c5d: search integration + commit completion (UPDATING_SEARCH->VERIFYING_SEARCH->APPLIED)

tolaria/c5-sync-service/rq_c5_cli.py

@@ -23,7 +23,7 @@ import os
 import sys
 from typing import Any, Dict, List, Optional
 
-from rq_c5a import C5AStore
+from rq_c5a import C5AStore, ST_APPLIED
 from rq_c5b import (
     C5BPoller,
     GitReader,
@@ -40,6 +40,16 @@ from rq_c5c import (
     DEFAULT_TOLARIA_BASE,
     ENV_TOLARIA_BASE,
 )
+from rq_c5d import (
+    C5DEngine,
+    C5DDryRun,
+    SearchClient,
+    assert_no_tolaria_write,
+    assert_no_production_activation,
+    DEFAULT_SEARCH_BASE,
+    ENV_SEARCH_BASE,
+    ENV_SEARCH_TOKEN,
+)
 
 
 def _store(db_path: str) -> C5AStore:
@@ -172,6 +182,45 @@ def cmd_c5c_guarantees(args: argparse.Namespace) -> int:
     return 0 if ok else 1
 
 
+def _c5d_search_client(args: argparse.Namespace) -> SearchClient:
+    return SearchClient(base_url=args.search_base, token=args.search_token)
+
+
+def cmd_apply_commit(args: argparse.Namespace) -> int:
+    """Vervollstaendigt einen Commit: Search-Rebuild -> Health/Integrity -> APPLIED.
+
+    NUR aus ST_UPDATING_SEARCH (Tolaria verifiziert) oder ST_RETRY_PENDING
+    (Search-Retry-Replay). Fuehrt KEINEN Tolaria-Write aus.
+    """
+    store = _store(args.db)
+    engine = C5DEngine(store, search=_c5d_search_client(args))
+    result = engine.apply_commit(args.sha)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    store.close()
+    return 0 if result.get("status") == ST_APPLIED else 1
+
+
+def cmd_c5d_plan(args: argparse.Namespace) -> int:
+    """Read-only C5D-Plan: Commits in UPDATING_SEARCH / RETRY_PENDING anzeigen."""
+    store = _store(args.db)
+    dry = C5DDryRun(store)
+    print(json.dumps(dry.plan(), ensure_ascii=False, indent=2))
+    store.close()
+    return 0
+
+
+def cmd_c5d_guarantees(args: argparse.Namespace) -> int:
+    """Prueft C5D-Guarantees (statisch): No-Tolaria-Write, No-Master-Write, No-Production."""
+    tol = assert_no_tolaria_write()
+    master = assert_no_master_write()
+    prod = assert_no_production_activation()
+    result = {"no_tolaria_write": tol, "no_master_write": master,
+              "no_production_activation": prod}
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    ok = tol["no_tolaria_write"] and master["no_master_write"] and prod["no_production_activation"]
+    return 0 if ok else 1
+
+
 def main(argv: Optional[List[str]] = None) -> int:
     parser = argparse.ArgumentParser(
         prog="rq_c5_cli",
@@ -193,6 +242,14 @@ def main(argv: Optional[List[str]] = None) -> int:
         "--tolaria-base", default=os.environ.get(ENV_TOLARIA_BASE, DEFAULT_TOLARIA_BASE),
         help=f"Tolaria-Vault-API-Basis (Env {ENV_TOLARIA_BASE}, Default {DEFAULT_TOLARIA_BASE})",
     )
+    parser.add_argument(
+        "--search-base", default=os.environ.get(ENV_SEARCH_BASE, DEFAULT_SEARCH_BASE),
+        help=f"Search-API-Basis (Env {ENV_SEARCH_BASE}, Default {DEFAULT_SEARCH_BASE})",
+    )
+    parser.add_argument(
+        "--search-token", default=os.environ.get(ENV_SEARCH_TOKEN, ""),
+        help=f"Search-Rebuild-Token (Env {ENV_SEARCH_TOKEN})",
+    )
 
     sub = parser.add_subparsers(dest="command", required=True)
 
@@ -226,6 +283,16 @@ def main(argv: Optional[List[str]] = None) -> int:
     p = sub.add_parser("c5c-guarantees", help="No-Search- und No-Master-Write-Guarantee pruefen (statisch)")
     p.set_defaults(func=cmd_c5c_guarantees)
 
+    p = sub.add_parser("apply-commit", help="Commit vervollstaendigen: Search-Rebuild -> Health/Integrity -> APPLIED")
+    p.add_argument("sha", help="Commit-SHA")
+    p.set_defaults(func=cmd_apply_commit)
+
+    p = sub.add_parser("c5d-plan", help="Read-only C5D-Plan: Commits in UPDATING_SEARCH / RETRY_PENDING")
+    p.set_defaults(func=cmd_c5d_plan)
+
+    p = sub.add_parser("c5d-guarantees", help="C5D-Guarantees pruefen (statisch): No-Tolaria-Write, No-Master-Write, No-Production")
+    p.set_defaults(func=cmd_c5d_guarantees)
+
     args = parser.parse_args(argv)
     return args.func(args)
 

tolaria/c5-sync-service/rq_c5a.py

@@ -188,6 +188,7 @@ _ALLOWED_TRANSITIONS = {
     (ST_VERIFYING_SEARCH, ST_HUMAN_REVIEW_REQUIRED),
     (ST_RETRY_PENDING, ST_READY),                # Retry -> erneut propagieren
     (ST_RETRY_PENDING, ST_PROPAGATING_TOLARIA),  # C5C: Retry-Replay -> erneut propagieren
+    (ST_RETRY_PENDING, ST_UPDATING_SEARCH),      # C5D: Search-Retry-Replay -> ab Search-Schritt fortsetzen (kein Tolaria-Doppel-Write)
     (ST_RETRY_PENDING, ST_RETRY_PENDING),        # idempotenter erneuter Retry-Versuch
     (ST_RETRY_PENDING, ST_FAILED),               # Retry-Limit erreicht
     (ST_RETRY_PENDING, ST_DEAD),                  # Max-Retry -> DEAD

tolaria/c5-sync-service/rq_c5d.py

@@ -0,0 +1,512 @@
+#!/usr/bin/env python3
+"""
+Red Queen — C5D: SEARCH INTEGRATION + COMMIT COMPLETION v1 (deterministische, inaktive Library).
+
+C5D implementiert ausschliesslich:
+  UPDATING_SEARCH -> SEARCH FULL REBUILD -> VERIFYING_SEARCH
+  -> SEARCH HEALTH/INTEGRITY READ-BACK -> APPLIED -> last_applied_commit fortschreiben
+
+VERBINDLICHE REIHENFOLGE (ORDERING INVARIANT):
+  Forgejo Commit -> Tolaria propagiert (C5C) -> Tolaria Read-Back DRIFT=0
+  -> Search Full Rebuild -> Search Health/Integrity PASS -> Commit APPLIED
+  -> last_applied_commit fortschreiben
+
+INVARIANTE: Search darf NIEMALS vor erfolgreicher Tolaria-Verifikation aktualisiert
+werden. APPLIED und last_applied_commit duerfen ausschliesslich gesetzt werden,
+wenn Tolaria UND Search vollstaendig erfolgreich und verifiziert sind.
+
+C5D startet NUR aus ST_UPDATING_SEARCH (Tolaria bereits verifiziert) oder
+ST_RETRY_PENDING (Search-Retry-Replay). C5D ruft NIE Tolaria auf -> kein
+Tolaria-Doppel-Write beim Search-Retry.
+
+ARCHITEKTURREGEL (verbindlich):
+  * Forgejo bleibt MASTER / Source of Truth.
+  * Tolaria ist ausschliesslich DERIVED.
+  * C5D darf: Forgejo lesen, Search kontrolliert rebuilden (mit Token).
+  * C5D darf NIEMALS: Tolaria -> Forgejo schreiben, Forgejo veraendern,
+    Tolaria schreiben, IDs neu vergeben, Knowledge Content umformulieren.
+
+NO-TOLARIA-WRITE-GUARANTEE: C5D ruft NICHT /api/vault/save auf.
+NO-MASTER-WRITE-GUARANTEE: C5D fuehrt KEIN git push / Forgejo-Write aus.
+NO-PRODUCTION-ACTIVATION: C5D fuehrt KEINE produktive Propagation / Search-Rebuild
+  aus (nur Tests gegen Fake/Mock + read-only Dry-Run).
+
+FAIL-CLOSED: Bei Search-Ausfall, Timeout, Auth-Fehler, ungueltiger Antwort,
+Rebuild-Fehler oder fehlerhafter Integrity -> Commit NICHT APPLIED,
+last_applied_commit NICHT veraendert, Fehler klassifiziert, Retry/Human-Gate
+gemaess C5A-Contract.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+# C5A / C5C wiederverwenden (keine konkurrierende State Machine)
+from rq_c5a import (
+    C5AStore,
+    ST_UPDATING_SEARCH,
+    ST_VERIFYING_SEARCH,
+    ST_APPLIED,
+    ST_RETRY_PENDING,
+    ST_DEAD,
+    ST_HUMAN_REVIEW_REQUIRED,
+    IDEM_ALREADY_APPLIED,
+    IDEM_RETRY_SAFE,
+    RC_SEARCH_REBUILD_FAILURE,
+    RC_AUTH_FAILURE,
+    RC_UNKNOWN_OBJECT_ID,
+    RC_UNEXPECTED_TOLARIA_DRIFT,
+    DEFAULT_MAX_RETRIES,
+    DEFAULT_BACKOFF_SECONDS,
+)
+from rq_c5c import C5CError
+
+# ---------------------------------------------------------------------------
+# Konstanten
+# ---------------------------------------------------------------------------
+
+# Search-API-Basis (Default; per Env C5_SEARCH_BASE ueberschreibbar)
+DEFAULT_SEARCH_BASE = "http://127.0.0.1:8325/api/search"
+ENV_SEARCH_BASE = "C5_SEARCH_BASE"
+ENV_SEARCH_TOKEN = "TOLARIA_SEARCH_REBUILD_TOKEN"
+
+# Search-Rebuild-Ergebnis
+REBUILD_OK = "ok"
+
+# Integrity-Verifikations-Ergebnisse
+INTEGRITY_OK = "INTEGRITY_OK"
+INTEGRITY_FAIL = "INTEGRITY_FAIL"
+
+
+class SearchError(C5CError):
+    """Basis-Fehler fuer C5D Search-Integration."""
+
+
+class SearchUnavailableError(SearchError):
+    """Search nicht erreichbar / Timeout / transienter HTTP-Fehler (retrybar)."""
+
+
+class SearchAuthError(SearchError):
+    """Search-Rebuild Auth-Fehler (nicht retrybar, Human Gate)."""
+
+
+class SearchRebuildError(SearchError):
+    """Search-Rebuild fehlgeschlagen (nicht retrybar, Human Gate)."""
+
+
+class SearchIntegrityError(SearchError):
+    """Search Health/Integrity nicht PASS (nicht retrybar, Human Gate)."""
+
+
+class SearchMalformedResponseError(SearchError):
+    """Search-Antwort ungueltig / malformed (nicht retrybar, Human Gate)."""
+
+
+# ---------------------------------------------------------------------------
+# Search Client (isoliert: rebuild / health)
+# ---------------------------------------------------------------------------
+
+class SearchClient:
+    """
+    Isolierter Search-Client. Nur C5D-Writer darf rebuild() verwenden.
+
+    rebuild() — POST /api/search/rebuild (Bearer-Token, kontrolliert)
+    health()  — GET  /api/search/health  (Read-Back, oeffentlich)
+
+    Keine generische Agent-Write-Funktion. Kein Tolaria-Write.
+    """
+
+    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None,
+                 timeout: float = 15.0):
+        self.base_url = (base_url or os.environ.get(ENV_SEARCH_BASE)
+                         or DEFAULT_SEARCH_BASE).rstrip("/")
+        self.token = token if token is not None else os.environ.get(ENV_SEARCH_TOKEN, "")
+        self.timeout = timeout
+
+    # -- HTTP-Helfer --------------------------------------------------------
+
+    def _post(self, endpoint: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        data = json.dumps(payload or {}).encode("utf-8")
+        headers = {"Content-Type": "application/json"}
+        if self.token:
+            headers["Authorization"] = "Bearer " + self.token
+        req = urllib.request.Request(url, data=data, headers=headers, method="POST")
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                body = resp.read().decode("utf-8")
+                return json.loads(body) if body else {}
+        except urllib.error.HTTPError as e:
+            if e.code in (401, 403):
+                raise SearchAuthError(
+                    f"Search Auth-Fehler HTTP {e.code} auf {endpoint}",
+                    RC_AUTH_FAILURE)
+            if e.code >= 500:
+                raise SearchUnavailableError(
+                    f"Search HTTP {e.code} auf {endpoint}", RC_SEARCH_REBUILD_FAILURE)
+            raise SearchRebuildError(
+                f"Search HTTP {e.code} auf {endpoint}: {e.read().decode('utf-8', 'replace')[:200]}",
+                RC_SEARCH_REBUILD_FAILURE)
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            raise SearchUnavailableError(
+                f"Search nicht erreichbar ({endpoint}): {e}", RC_SEARCH_REBUILD_FAILURE)
+
+    def _get(self, endpoint: str) -> Dict[str, Any]:
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        req = urllib.request.Request(url, method="GET")
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                body = resp.read().decode("utf-8")
+                return json.loads(body) if body else {}
+        except urllib.error.HTTPError as e:
+            if e.code >= 500:
+                raise SearchUnavailableError(
+                    f"Search HTTP {e.code} auf {endpoint}", RC_SEARCH_REBUILD_FAILURE)
+            raise SearchRebuildError(
+                f"Search HTTP {e.code} auf {endpoint}: {e.read().decode('utf-8', 'replace')[:200]}",
+                RC_SEARCH_REBUILD_FAILURE)
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            raise SearchUnavailableError(
+                f"Search nicht erreichbar ({endpoint}): {e}", RC_SEARCH_REBUILD_FAILURE)
+
+    # -- Rebuild (NUR C5D-Writer) -------------------------------------------
+
+    def rebuild(self) -> Dict[str, Any]:
+        """Fuehrt einen Search Full Rebuild aus (POST /api/search/rebuild)."""
+        resp = self._post("rebuild")
+        if not isinstance(resp, dict):
+            raise SearchMalformedResponseError(
+                f"Search-Rebuild-Antwort ungueltig: {type(resp).__name__}",
+                RC_SEARCH_REBUILD_FAILURE)
+        if resp.get("status") != REBUILD_OK:
+            raise SearchRebuildError(
+                f"Search-Rebuild fehlgeschlagen: {resp.get('error', resp)}",
+                RC_SEARCH_REBUILD_FAILURE)
+        return resp
+
+    # -- Health (Read-Back) -------------------------------------------------
+
+    def health(self) -> Dict[str, Any]:
+        """Liest Search Health/Integrity (GET /api/search/health)."""
+        resp = self._get("health")
+        if not isinstance(resp, dict):
+            raise SearchMalformedResponseError(
+                f"Search-Health-Antwort ungueltig: {type(resp).__name__}",
+                RC_SEARCH_REBUILD_FAILURE)
+        return resp
+
+
+# ---------------------------------------------------------------------------
+# Integrity-Verifikation (Read-Back, §C5D)
+# ---------------------------------------------------------------------------
+
+def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Verifiziert Search Health/Integrity per Read-Back.
+
+    Ein HTTP 200 allein reicht NICHT als Erfolg. Health muss PASS sein:
+      * integrity_ok == True
+      * index_built == True
+      * object_count > 0  (unerwarteter/falscher Indexzustand -> FAIL)
+      * failed_objects leer (optional, aber fail-closed wenn nicht leer)
+
+    Rueckgabe: {"ok": bool, "checks": {...}, "reason": optional}
+    """
+    if not isinstance(health, dict):
+        raise SearchMalformedResponseError(
+            f"Search-Health-Antwort ungueltig: {type(health).__name__}",
+            RC_SEARCH_REBUILD_FAILURE)
+    checks = {
+        "integrity_ok": bool(health.get("integrity_ok")),
+        "index_built": bool(health.get("index_built")),
+        "object_count_positive": int(health.get("object_count", 0)) > 0,
+        "failed_objects_empty": not bool(health.get("failed_objects")),
+    }
+    ok = all(checks.values())
+    reason = None
+    if not ok:
+        failed = [k for k, v in checks.items() if not v]
+        reason = "Search-Health nicht PASS: " + ", ".join(failed)
+    return {"ok": ok, "checks": checks, "reason": reason}
+
+
+# ---------------------------------------------------------------------------
+# C5D Engine (Search Integration + Commit Completion)
+# ---------------------------------------------------------------------------
+
+class C5DEngine:
+    """
+    Vervollstaendigt einen Commit nach erfolgreicher Tolaria-Propagation (C5C).
+
+    Startet NUR aus ST_UPDATING_SEARCH (Tolaria verifiziert, READY_FOR_SEARCH)
+    oder ST_RETRY_PENDING (Search-Retry-Replay). Fuehrt Search-Rebuild aus,
+    verifiziert Health/Integrity per Read-Back, und setzt bei vollem PASS
+    ST_APPLIED + last_applied_commit.
+
+    C5D ruft NIE Tolaria auf -> kein Tolaria-Doppel-Write beim Search-Retry.
+    """
+
+    def __init__(
+        self,
+        store: C5AStore,
+        search: Optional[SearchClient] = None,
+        max_retries: int = DEFAULT_MAX_RETRIES,
+        backoff_seconds: Optional[List[int]] = None,
+    ):
+        self.store = store
+        self.search = search or SearchClient()
+        self.max_retries = max_retries
+        self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS
+
+    # -- Idempotenz ---------------------------------------------------------
+
+    def _commit_already_applied(self, commit_sha: str) -> bool:
+        """Commit bereits vollstaendig APPLIED -> kein Downstream-Write."""
+        return self.store.commit_status(commit_sha) == ST_APPLIED
+
+    # -- Retry-Klassifikation (§17) -----------------------------------------
+
+    def _is_retryable(self, err: SearchError) -> bool:
+        """Nur technische, retrybare Fehler nutzen das Retry-Modell."""
+        return isinstance(err, SearchUnavailableError)
+
+    def _handle_failure(self, commit_sha: str, err: SearchError) -> Dict[str, Any]:
+        """Behandelt einen Search-Fehler: Retry, DEAD oder Human Gate."""
+        cur = self.store.commit_status(commit_sha)
+        if self._is_retryable(err):
+            retry = self.store.increment_retry(commit_sha)
+            # Sicherstellen, dass wir in RETRY_PENDING sind (erlaubt von
+            # UPDATING_SEARCH und VERIFYING_SEARCH), bevor wir ggf. zu DEAD wechseln.
+            if cur in (ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH):
+                self.store.transition_commit(commit_sha, ST_RETRY_PENDING)
+            if retry >= self.max_retries:
+                self.store.set_commit_error(commit_sha, err.reason_code or RC_SEARCH_REBUILD_FAILURE, err.message)
+                self.store.transition_commit(commit_sha, ST_DEAD)
+                return {"commit_sha": commit_sha, "status": ST_DEAD,
+                        "reason_code": err.reason_code, "retry_count": retry}
+            self.store.set_commit_error(commit_sha, err.reason_code or RC_SEARCH_REBUILD_FAILURE, err.message)
+            return {"commit_sha": commit_sha, "status": ST_RETRY_PENDING,
+                    "reason_code": err.reason_code, "retry_count": retry}
+        # Nicht-retrybar -> Human Gate / Fail Closed
+        rc = err.reason_code or RC_SEARCH_REBUILD_FAILURE
+        self.store.set_commit_error(commit_sha, rc, err.message)
+        self.store.transition_commit(commit_sha, ST_HUMAN_REVIEW_REQUIRED)
+        return {"commit_sha": commit_sha, "status": ST_HUMAN_REVIEW_REQUIRED,
+                "reason_code": rc}
+
+    # -- Commit anwenden ----------------------------------------------------
+
+    def apply_commit(self, commit_sha: str) -> Dict[str, Any]:
+        """
+        Vervollstaendigt einen Commit: Search-Rebuild -> Health/Integrity -> APPLIED.
+
+        VERBINDLICHE REIHENFOLGE:
+          UPDATING_SEARCH -> (Search Rebuild) -> VERIFYING_SEARCH
+          -> (Health/Integrity Read-Back) -> APPLIED -> last_applied_commit
+
+        FAIL-CLOSED: Bei jedem Search-Fehler bleibt der Commit NICHT APPLIED und
+        last_applied_commit unveraendert.
+        """
+        commit = self.store.get_commit(commit_sha)
+        if commit is None:
+            return {"commit_sha": commit_sha, "status": ST_HUMAN_REVIEW_REQUIRED,
+                    "reason_code": RC_UNKNOWN_OBJECT_ID}
+
+        cur = commit.get("status")
+
+        # Idempotenz: bereits APPLIED -> kein Downstream-Write
+        if cur == ST_APPLIED:
+            return {"commit_sha": commit_sha, "status": ST_APPLIED,
+                    "idempotency": IDEM_ALREADY_APPLIED}
+
+        # Nur aus UPDATING_SEARCH oder RETRY_PENDING (Search-Retry-Replay) starten
+        if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING):
+            return {"commit_sha": commit_sha, "status": cur,
+                    "message": "Commit nicht im Search-Schritt (Tolaria nicht verifiziert)"}
+
+        # Search-Retry-Replay: RETRY_PENDING -> UPDATING_SEARCH (kein Tolaria-Write)
+        if cur == ST_RETRY_PENDING:
+            self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH)
+
+        # 1) Search Full Rebuild (UPDATING_SEARCH)
+        try:
+            rebuild = self.search.rebuild()
+        except SearchError as e:
+            return self._handle_failure(commit_sha, e)
+
+        # Defense in depth: HTTP 200 allein reicht NICHT. Die Engine validiert
+        # den Rebuild-Response selbst (status == "ok"), unabhaengig vom Client.
+        if not isinstance(rebuild, dict) or rebuild.get("status") != REBUILD_OK:
+            return self._handle_failure(
+                commit_sha,
+                SearchRebuildError(
+                    f"Search-Rebuild-Response ungueltig: {rebuild!r}",
+                    RC_SEARCH_REBUILD_FAILURE))
+
+        # 2) Rebuild OK -> VERIFYING_SEARCH
+        self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH)
+
+        # 3) Health/Integrity Read-Back (VERIFYING_SEARCH)
+        try:
+            health = self.search.health()
+            integrity = verify_integrity(health)
+        except SearchError as e:
+            return self._handle_failure(commit_sha, e)
+
+        if not integrity["ok"]:
+            return self._handle_failure(
+                commit_sha,
+                SearchIntegrityError(integrity["reason"] or "Search-Integrity nicht PASS",
+                                     RC_SEARCH_REBUILD_FAILURE))
+
+        # 4) Vollstaendiger PASS -> APPLIED + last_applied_commit fortschreiben
+        self.store.transition_commit(commit_sha, ST_APPLIED)
+        self.store.mark_applied(commit_sha)
+
+        return {"commit_sha": commit_sha, "status": ST_APPLIED,
+                "idempotency": IDEM_RETRY_SAFE, "integrity": integrity,
+                "rebuild": rebuild}
+
+
+# ---------------------------------------------------------------------------
+# C5D Dry-Run (read-only) — zeigt, welche Commits fuer Search bereit sind
+# ---------------------------------------------------------------------------
+
+class C5DDryRun:
+    """
+    Read-only Dry-Run: identifiziert Commits in UPDATING_SEARCH (READY_FOR_SEARCH)
+    und RETRY_PENDING (Search-Retry-Replay). Fuehrt KEINEN Search-Rebuild aus.
+    """
+
+    def __init__(self, store: C5AStore):
+        self.store = store
+
+    def plan(self) -> Dict[str, Any]:
+        """Erzeugt den C5D-Plan (read-only, KEIN Search-Rebuild)."""
+        ready = []
+        retry = []
+        for c in self.store.list_commits():
+            status = c.get("status")
+            if status == ST_UPDATING_SEARCH:
+                ready.append(c.get("commit_sha"))
+            elif status == ST_RETRY_PENDING:
+                retry.append(c.get("commit_sha"))
+        return {
+            "ready_for_search": ready,
+            "search_retry_pending": retry,
+            "note": "read-only Dry-Run: KEIN produktiver Search-Rebuild",
+        }
+
+
+# ---------------------------------------------------------------------------
+# Guarantees (statische Pruefung)
+# ---------------------------------------------------------------------------
+
+def _collect_docstrings(tree: Any) -> set:
+    """Sammelt alle Docstring-Strings (Modul-/Funktions-/Klassen-Docstrings)."""
+    import ast
+    docstrings = set()
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef,
+                             ast.ClassDef)):
+            body = node.body
+            if body and isinstance(body[0], ast.Expr):
+                val = body[0].value
+                if isinstance(val, ast.Constant) and isinstance(val.value, str):
+                    docstrings.add(val.value)
+    return docstrings
+
+
+def assert_no_tolaria_write() -> Dict[str, Any]:
+    """
+    Beweist statisch, dass C5D NICHT nach Tolaria schreibt (kein /api/vault/save).
+
+    Prueft, dass keine Tolaria-Write-Endpoint-Strings in echten Code-Ausdruecken
+    (nicht Docstrings/Kommentaren) vorkommen.
+    """
+    import ast
+    this_file = Path(__file__).resolve()
+    tree = ast.parse(this_file.read_text(encoding="utf-8"))
+    docstrings = _collect_docstrings(tree)
+    # Fragmentierte banned-Strings (Selbstreferenz vermeiden)
+    banned = ["/api/" + "vault/" + "save", "vault/" + "save", "api/" + "vault"]
+    found = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            if node.value in docstrings:
+                continue
+            for b in banned:
+                if b in node.value:
+                    found.append(node.value)
+    return {
+        "no_tolaria_write": len(found) == 0,
+        "tolaria_write_endpoints_found": sorted(set(found)),
+    }
+
+
+def assert_no_master_write() -> Dict[str, Any]:
+    """
+    Beweist statisch, dass C5D KEIN git push / Forgejo-Write ausfuehrt.
+    """
+    import ast
+    this_file = Path(__file__).resolve()
+    tree = ast.parse(this_file.read_text(encoding="utf-8"))
+    docstrings = _collect_docstrings(tree)
+    banned_imports = {"subprocess", "os.system", "git"}
+    found_imports = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                root = alias.name.split(".")[0]
+                if root in banned_imports:
+                    found_imports.add(root)
+        elif isinstance(node, ast.ImportFrom):
+            if node.module:
+                root = node.module.split(".")[0]
+                if root in banned_imports:
+                    found_imports.add(root)
+    banned_calls = ["git " + "pus" + "h", "git " + "comm" + "it", "git " + "ad" + "d", "pus" + "h"]
+    found_calls = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            if node.value in docstrings:
+                continue
+            for b in banned_calls:
+                if b in node.value:
+                    found_calls.append(node.value)
+    return {
+        "no_master_write": len(found_imports) == 0 and len(found_calls) == 0,
+        "banned_imports_found": sorted(found_imports),
+        "write_commands_found": sorted(set(found_calls)),
+    }
+
+
+def assert_no_production_activation() -> Dict[str, Any]:
+    """
+    Beweist statisch, dass C5D KEINE produktive Aktivierung enthaelt
+    (kein Daemon/Deploy/Canary/Thread/Cron/Self-Schedule).
+    """
+    import ast
+    this_file = Path(__file__).resolve()
+    tree = ast.parse(this_file.read_text(encoding="utf-8"))
+    docstrings = _collect_docstrings(tree)
+    banned = ["daem" + "on", "serve_" + "forever", "Threading" + "HTTPServer",
+              "cro" + "n", "sched" + "ule", "depl" + "oy", "can" + "ary",
+              "threading." + "Thread", "while " + "True"]
+    found = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            if node.value in docstrings:
+                continue
+            for b in banned:
+                if b in node.value:
+                    found.append(node.value)
+    return {
+        "no_production_activation": len(found) == 0,
+        "activation_patterns_found": sorted(set(found)),
+    }

tolaria/c5-sync-service/test_c5d.py

@@ -0,0 +1,684 @@
+#!/usr/bin/env python3
+"""
+Red Queen — C5D Testsuite (Search Integration + Commit Completion).
+
+Testet gegen einen Fake-Search-Client + Fake-Tolaria-Client (isolierte Mocks).
+Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB.
+KEINE produktiven Writes. KEIN produktiver Search-Rebuild.
+
+Abgedeckte Faelle (C5D-Prompt):
+  erfolgreicher Search-Rebuild
+  erfolgreicher Health-/Integrity-Read-Back
+  Search Auth Failure
+  Search unavailable
+  Network Timeout
+  malformed Search Response
+  integrity_ok=false
+  unerwarteter/falscher Indexzustand
+  Search Rebuild Failure
+  Retry -> erfolgreiche Recovery
+  Max Retry -> DEAD/Human Gate gemaess Contract
+  Replay nach bereits erfolgreicher Tolaria-Phase
+  kein Tolaria-Doppel-Write beim Search-Retry
+  Search darf bei fehlgeschlagener Tolaria-Verifikation NIEMALS aufgerufen werden
+  last_applied_commit bleibt bei JEDEM Failure unveraendert
+  APPLIED darf ohne Search PASS niemals erreicht werden
+  Duplicate/Replay Idempotenz
+  Restart-/Persistence-Verhalten soweit C5D betroffen
+  Secret Safety
+  Regression der bestehenden C5A/C5B/C5C-Contracts
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from typing import Any, Dict, List, Optional
+
+# C5D importieren (aus demselben Verzeichnis)
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from rq_c5a import (
+    C5AStore, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA,
+    ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED, ST_RETRY_PENDING,
+    ST_DEAD, ST_HUMAN_REVIEW_REQUIRED, ST_FAILED,
+    IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE,
+    RC_SEARCH_REBUILD_FAILURE, RC_AUTH_FAILURE, RC_UNKNOWN_OBJECT_ID,
+    RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT,
+    OP_CREATE, OP_CONTENT_UPDATE,
+)
+from rq_c5b import (
+    GitReader, parse_frontmatter, content_hash, metadata_hash, detect_secret,
+)
+from rq_c5c import (
+    TolariaClient, C5CPropagator, C5CDryRun,
+    pre_write_drift_check, _vault_path,
+    DRIFT_WRITE_ALLOWED, DRIFT_ALREADY_AT_TARGET, DRIFT_UNEXPECTED,
+    PLAN_WOULD_WRITE, PLAN_ALREADY_AT_TARGET, PLAN_HUMAN_REVIEW, PLAN_DRIFT,
+    TolariaUnavailableError, ReadBackMismatchError, UnexpectedDriftError,
+    assert_no_search_calls, assert_no_master_write,
+)
+from rq_c5d import (
+    SearchClient, C5DEngine, C5DDryRun,
+    verify_integrity,
+    SearchError, SearchUnavailableError, SearchAuthError,
+    SearchRebuildError, SearchIntegrityError, SearchMalformedResponseError,
+    assert_no_tolaria_write, assert_no_production_activation,
+    INTEGRITY_OK, INTEGRITY_FAIL,
+)
+
+UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
+UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
+
+
+# ---------------------------------------------------------------------------
+# Fixture: kontrolliertes Git-Repo (Test-Helfer darf schreiben)
+# ---------------------------------------------------------------------------
+
+def _git(repo: str, *args: str) -> str:
+    proc = subprocess.run(["git", "-C", repo] + list(args),
+                          capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} fehlgeschlagen: {proc.stderr}")
+    return proc.stdout
+
+
+def _write(repo: str, path: str, content: str) -> None:
+    full = os.path.join(repo, path)
+    os.makedirs(os.path.dirname(full), exist_ok=True)
+    with open(full, "w", encoding="utf-8") as f:
+        f.write(content)
+
+
+def _commit(repo: str, message: str) -> str:
+    _git(repo, "add", "-A")
+    _git(repo, "commit", "-m", message, "--no-verify")
+    return _git(repo, "rev-parse", "HEAD").strip()
+
+
+def _fm(id_: str, **extra) -> str:
+    lines = ["---", "knowledge_schema: 1", f"id: {id_}"]
+    for k, v in extra.items():
+        if isinstance(v, list):
+            lines.append(f"{k}: [{', '.join(v)}]")
+        else:
+            lines.append(f"{k}: {v}")
+    lines.append("---")
+    return "\n".join(lines) + "\n"
+
+
+class FixtureRepo:
+    def __init__(self):
+        self.dir = tempfile.mkdtemp(prefix="c5d_fixture_")
+        _git(self.dir, "init", "-q", "-b", "main")
+        _git(self.dir, "config", "user.email", "test@test")
+        _git(self.dir, "config", "user.name", "Test")
+        self.commits: Dict[str, str] = {}
+
+    def write(self, path: str, content: str) -> None:
+        _write(self.dir, path, content)
+
+    def commit(self, message: str) -> str:
+        sha = _commit(self.dir, message)
+        self.commits[message] = sha
+        return sha
+
+    def cleanup(self) -> None:
+        shutil.rmtree(self.dir, ignore_errors=True)
+
+
+# ---------------------------------------------------------------------------
+# Fake-Tolaria-API (Mock, isolierter Test-Vault) — zaehlt Writes
+# ---------------------------------------------------------------------------
+
+class FakeTolaria:
+    """In-Memory Fake der Tolaria-Vault-API. Zaehlt Writes, um Doppel-Writes zu beweisen."""
+
+    def __init__(self):
+        self.vault: Dict[str, str] = {}
+        self.write_count = 0
+        self.unavailable = False
+
+    def read(self, vault_path: str) -> Optional[str]:
+        if self.unavailable:
+            raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
+        return self.vault.get(vault_path)
+
+    def write(self, vault_path: str, content: str) -> Dict[str, Any]:
+        if self.unavailable:
+            raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
+        self.vault[vault_path] = content
+        self.write_count += 1
+        return {"ok": True}
+
+    def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
+        return [{"path": p} for p in self.vault]
+
+
+class FakeTolariaClient(TolariaClient):
+    """TolariaClient, der gegen FakeTolaria statt HTTP arbeitet."""
+
+    def __init__(self, fake: FakeTolaria):
+        super().__init__(base_url="http://fake")
+        self.fake = fake
+
+    def read(self, vault_path: str) -> Optional[str]:
+        return self.fake.read(vault_path)
+
+    def write(self, vault_path: str, content: str) -> Dict[str, Any]:
+        return self.fake.write(vault_path, content)
+
+    def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
+        return self.fake.list(vault_path)
+
+
+# ---------------------------------------------------------------------------
+# Fake-Search-API (Mock) — konfigurierbare Fehler
+# ---------------------------------------------------------------------------
+
+class FakeSearch:
+    """In-Memory Fake der Search-API. Konfigurierbare Fehler + Zaehler."""
+
+    def __init__(self):
+        self.rebuild_count = 0
+        self.health_count = 0
+        self.rebuild_error: Optional[Exception] = None
+        self.health_error: Optional[Exception] = None
+        self.rebuild_response: Optional[Any] = None
+        self.health_response: Optional[Any] = None
+        self.rebuild_ok = True
+
+    def rebuild(self) -> Dict[str, Any]:
+        self.rebuild_count += 1
+        if self.rebuild_error:
+            raise self.rebuild_error
+        if self.rebuild_response is not None:
+            return self.rebuild_response
+        if not self.rebuild_ok:
+            return {"status": "error", "error": "rebuild failed"}
+        return {"status": "ok", "indexed": 1}
+
+    def health(self) -> Dict[str, Any]:
+        self.health_count += 1
+        if self.health_error:
+            raise self.health_error
+        if self.health_response is not None:
+            return self.health_response
+        return {
+            "index_built": True,
+            "object_count": 1,
+            "failed_objects": [],
+            "integrity_ok": True,
+            "supported_modes": ["exact", "keyword", "metadata"],
+            "index_version": 1,
+            "source_head": "abc",
+            "secret_blocked_objects": 0,
+        }
+
+
+class FakeSearchClient(SearchClient):
+    """SearchClient, der gegen FakeSearch statt HTTP arbeitet."""
+
+    def __init__(self, fake: FakeSearch):
+        super().__init__(base_url="http://fake-search")
+        self.fake = fake
+
+    def rebuild(self) -> Dict[str, Any]:
+        return self.fake.rebuild()
+
+    def health(self) -> Dict[str, Any]:
+        return self.fake.health()
+
+
+# ---------------------------------------------------------------------------
+# Test-Helfer: C5A-Store + Commit mit ObjectChanges
+# ---------------------------------------------------------------------------
+
+def _make_store() -> C5AStore:
+    db = os.path.join(tempfile.mkdtemp(prefix="c5d_db_"), "c5a.db")
+    return C5AStore(db)
+
+
+def _seed_commit(store: C5AStore, sha: str, parent: Optional[str],
+                 objs: List[Dict[str, Any]], status: str = ST_UPDATING_SEARCH) -> None:
+    """Legt einen Commit + ObjectChanges im Store an (Status UPDATING_SEARCH)."""
+    store.upsert_commit({
+        "commit_sha": sha, "parent_sha": parent, "sequence": 1,
+        "status": status, "retry_count": 0,
+    })
+    for oc in objs:
+        oc["commit_sha"] = sha
+        store.add_object_change(oc)
+
+
+def _seed_ready_for_search(store: C5AStore, sha: str) -> None:
+    """Legt einen Commit an, der Tolaria-verifiziert ist (READY_FOR_SEARCH)."""
+    _seed_commit(store, sha, None, [{
+        "object_id": UUID_A, "operation": OP_CREATE,
+        "path_before": None, "path_after": "modul-09.md",
+        "content_hash_after": content_hash("body"),
+    }], status=ST_UPDATING_SEARCH)
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+class TestVerifyIntegrity(unittest.TestCase):
+    """§C5D Health/Integrity Read-Back-Verifikation."""
+
+    def test_integrity_ok(self):
+        health = {"integrity_ok": True, "index_built": True,
+                  "object_count": 5, "failed_objects": []}
+        res = verify_integrity(health)
+        self.assertTrue(res["ok"])
+        self.assertIsNone(res["reason"])
+
+    def test_integrity_ok_false(self):
+        health = {"integrity_ok": False, "index_built": True,
+                  "object_count": 5, "failed_objects": []}
+        res = verify_integrity(health)
+        self.assertFalse(res["ok"])
+        self.assertIn("integrity_ok", res["reason"])
+
+    def test_index_not_built(self):
+        health = {"integrity_ok": True, "index_built": False,
+                  "object_count": 5, "failed_objects": []}
+        res = verify_integrity(health)
+        self.assertFalse(res["ok"])
+        self.assertIn("index_built", res["reason"])
+
+    def test_object_count_zero(self):
+        # unerwarteter/falscher Indexzustand: 0 Objekte
+        health = {"integrity_ok": True, "index_built": True,
+                  "object_count": 0, "failed_objects": []}
+        res = verify_integrity(health)
+        self.assertFalse(res["ok"])
+        self.assertIn("object_count_positive", res["reason"])
+
+    def test_failed_objects_present(self):
+        health = {"integrity_ok": True, "index_built": True,
+                  "object_count": 5, "failed_objects": ["x.md"]}
+        res = verify_integrity(health)
+        self.assertFalse(res["ok"])
+        self.assertIn("failed_objects_empty", res["reason"])
+
+
+class TestC5DEngine(unittest.TestCase):
+    """C5D-Engine: Search Integration + Commit Completion."""
+
+    def setUp(self):
+        self.fake_tol = FakeTolaria()
+        self.tol_client = FakeTolariaClient(self.fake_tol)
+        self.fake_search = FakeSearch()
+        self.search_client = FakeSearchClient(self.fake_search)
+        self.store = _make_store()
+        self.engine = C5DEngine(self.store, search=self.search_client)
+
+    def tearDown(self):
+        self.store.close()
+
+    def _last_applied(self) -> Optional[str]:
+        return self.store.health().get("last_applied_commit")
+
+    # -- Erfolgreicher Pfad ------------------------------------------------
+
+    def test_successful_rebuild_and_apply(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(self.fake_search.rebuild_count, 1)
+        self.assertEqual(self.fake_search.health_count, 1)
+        self.assertEqual(self._last_applied(), sha)
+        # C5D darf NIE Tolaria schreiben
+        self.assertEqual(self.fake_tol.write_count, 0)
+
+    def test_apply_requires_verifying_search_transition(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.engine.apply_commit(sha)
+        # Nach APPLIED ist der Commit im Endzustand
+        self.assertEqual(self.store.commit_status(sha), ST_APPLIED)
+
+    # -- Idempotenz / Replay ----------------------------------------------
+
+    def test_already_applied_no_downstream_write(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.engine.apply_commit(sha)
+        self.assertEqual(self.fake_search.rebuild_count, 1)
+        # Erneuter Aufruf -> kein weiterer Search-Rebuild
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(res["idempotency"], IDEM_ALREADY_APPLIED)
+        self.assertEqual(self.fake_search.rebuild_count, 1)
+        self.assertEqual(self.fake_tol.write_count, 0)
+
+    def test_replay_after_successful_tolaria_phase(self):
+        # Commit ist in UPDATING_SEARCH (Tolaria bereits verifiziert).
+        # C5D darf NIE Tolaria erneut aufrufen (kein Doppel-Write).
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.engine.apply_commit(sha)
+        self.assertEqual(self.fake_tol.write_count, 0)
+        self.assertEqual(self.fake_search.rebuild_count, 1)
+
+    # -- Search Auth Failure ----------------------------------------------
+
+    def test_auth_failure_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_error = SearchAuthError("auth", RC_AUTH_FAILURE)
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertEqual(res["reason_code"], RC_AUTH_FAILURE)
+        self.assertIsNone(self._last_applied())
+        self.assertEqual(self.store.commit_status(sha), ST_HUMAN_REVIEW_REQUIRED)
+
+    # -- Search unavailable (retrybar) -------------------------------------
+
+    def test_search_unavailable_retry_pending(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE)
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_RETRY_PENDING)
+        self.assertEqual(res["retry_count"], 1)
+        self.assertIsNone(self._last_applied())
+        self.assertEqual(self.store.commit_status(sha), ST_RETRY_PENDING)
+
+    # -- Network Timeout (retrybar) ----------------------------------------
+
+    def test_network_timeout_retry_pending(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_error = SearchUnavailableError("timeout", RC_SEARCH_REBUILD_FAILURE)
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_RETRY_PENDING)
+        self.assertEqual(res["retry_count"], 1)
+        self.assertIsNone(self._last_applied())
+
+    # -- Malformed Search Response ----------------------------------------
+
+    def test_malformed_rebuild_response_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_response = "not-a-dict"
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied())
+
+    def test_malformed_health_response_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.health_response = "not-a-dict"
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied())
+
+    # -- integrity_ok=false ------------------------------------------------
+
+    def test_integrity_false_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.health_response = {
+            "index_built": True, "object_count": 1, "failed_objects": [],
+            "integrity_ok": False,
+        }
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied())
+        self.assertEqual(self.store.commit_status(sha), ST_HUMAN_REVIEW_REQUIRED)
+
+    # -- unerwarteter/falscher Indexzustand --------------------------------
+
+    def test_unexpected_index_state_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # object_count=0 -> falscher Indexzustand
+        self.fake_search.health_response = {
+            "index_built": True, "object_count": 0, "failed_objects": [],
+            "integrity_ok": True,
+        }
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied())
+
+    # -- Search Rebuild Failure -------------------------------------------
+
+    def test_rebuild_failure_human_gate(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_ok = False  # status != "ok"
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied())
+
+    # -- Retry -> erfolgreiche Recovery -----------------------------------
+
+    def test_retry_then_recovery(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # 1. Versuch: Search down -> RETRY_PENDING
+        self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE)
+        res1 = self.engine.apply_commit(sha)
+        self.assertEqual(res1["status"], ST_RETRY_PENDING)
+        self.assertEqual(res1["retry_count"], 1)
+        # 2. Versuch (Replay): Search wieder da -> APPLIED
+        self.fake_search.rebuild_error = None
+        res2 = self.engine.apply_commit(sha)
+        self.assertEqual(res2["status"], ST_APPLIED)
+        self.assertEqual(self._last_applied(), sha)
+        # Kein Tolaria-Write waehrend des gesamten Retry-Zyklus
+        self.assertEqual(self.fake_tol.write_count, 0)
+
+    # -- Max Retry -> DEAD ------------------------------------------------
+
+    def test_max_retry_reaches_dead(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE)
+        # max_retries=2 -> nach 2 Fehlversuchen DEAD
+        engine = C5DEngine(self.store, search=self.search_client, max_retries=2)
+        res1 = engine.apply_commit(sha)
+        self.assertEqual(res1["status"], ST_RETRY_PENDING)
+        res2 = engine.apply_commit(sha)
+        self.assertEqual(res2["status"], ST_DEAD)
+        self.assertIsNone(self._last_applied())
+        self.assertEqual(self.store.commit_status(sha), ST_DEAD)
+
+    # -- Search darf bei fehlgeschlagener Tolaria-Verifikation NIE aufgerufen werden
+
+    def test_search_never_called_when_tolaria_not_verified(self):
+        # Commit ist in ST_READY (Tolaria NICHT verifiziert) -> C5D darf nicht starten
+        sha = "c1"
+        _seed_commit(self.store, sha, None, [{
+            "object_id": UUID_A, "operation": OP_CREATE,
+            "path_before": None, "path_after": "modul-09.md",
+            "content_hash_after": content_hash("body"),
+        }], status=ST_READY)
+        res = self.engine.apply_commit(sha)
+        # C5D startet NICHT aus ST_READY -> kein Search-Rebuild
+        self.assertEqual(self.fake_search.rebuild_count, 0)
+        self.assertEqual(self.fake_search.health_count, 0)
+        self.assertNotEqual(res["status"], ST_APPLIED)
+        self.assertIsNone(self._last_applied())
+
+    def test_search_never_called_when_tolaria_drift(self):
+        # Commit in ST_HUMAN_REVIEW_REQUIRED (Tolaria-Drift) -> C5D darf nicht starten
+        sha = "c1"
+        _seed_commit(self.store, sha, None, [{
+            "object_id": UUID_A, "operation": OP_CREATE,
+            "path_before": None, "path_after": "modul-09.md",
+            "content_hash_after": content_hash("body"),
+        }], status=ST_HUMAN_REVIEW_REQUIRED)
+        res = self.engine.apply_commit(sha)
+        self.assertEqual(self.fake_search.rebuild_count, 0)
+        self.assertNotEqual(res["status"], ST_APPLIED)
+
+    # -- last_applied_commit bleibt bei JEDEM Failure unveraendert ---------
+
+    def test_last_applied_unchanged_on_all_failures(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # Auth-Fehler
+        self.fake_search.rebuild_error = SearchAuthError("auth", RC_AUTH_FAILURE)
+        self.engine.apply_commit(sha)
+        self.assertIsNone(self._last_applied())
+        # Integrity-Fehler
+        sha2 = "c2"
+        _seed_ready_for_search(self.store, sha2)
+        self.fake_search.rebuild_error = None
+        self.fake_search.health_response = {"integrity_ok": False, "index_built": True,
+                                            "object_count": 1, "failed_objects": []}
+        self.engine.apply_commit(sha2)
+        self.assertIsNone(self._last_applied())
+
+    # -- APPLIED darf ohne Search PASS niemals erreicht werden -------------
+
+    def test_applied_requires_search_pass(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # Search-Rebuild schlaegt fehl -> kein APPLIED
+        self.fake_search.rebuild_ok = False
+        res = self.engine.apply_commit(sha)
+        self.assertNotEqual(res["status"], ST_APPLIED)
+        self.assertIsNone(self._last_applied())
+
+    # -- Restart-/Persistence-Verhalten ------------------------------------
+
+    def test_persistence_across_engine_instances(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # Engine 1: Search down -> RETRY_PENDING (persistiert)
+        self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE)
+        self.engine.apply_commit(sha)
+        self.assertEqual(self.store.commit_status(sha), ST_RETRY_PENDING)
+        # Engine 2 (neue Instanz, gleiche DB): Replay -> APPLIED
+        engine2 = C5DEngine(self.store, search=self.search_client)
+        self.fake_search.rebuild_error = None
+        res = engine2.apply_commit(sha)
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(self._last_applied(), sha)
+
+    # -- kein Tolaria-Doppel-Write beim Search-Retry ----------------------
+
+    def test_no_tolaria_double_write_on_search_retry(self):
+        sha = "c1"
+        _seed_ready_for_search(self.store, sha)
+        # Mehrere Search-Fehler + Recovery -> Tolaria-Write bleibt 0
+        self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE)
+        self.engine.apply_commit(sha)
+        self.engine.apply_commit(sha)
+        self.fake_search.rebuild_error = None
+        self.engine.apply_commit(sha)
+        self.assertEqual(self.fake_tol.write_count, 0)
+
+
+class TestC5DDryRun(unittest.TestCase):
+    """C5D Dry-Run (read-only)."""
+
+    def setUp(self):
+        self.store = _make_store()
+
+    def tearDown(self):
+        self.store.close()
+
+    def test_plan_lists_ready_and_retry(self):
+        _seed_ready_for_search(self.store, "c1")
+        _seed_commit(self.store, "c2", None, [{
+            "object_id": UUID_A, "operation": OP_CREATE,
+            "path_before": None, "path_after": "modul-09.md",
+            "content_hash_after": content_hash("body"),
+        }], status=ST_RETRY_PENDING)
+        dry = C5DDryRun(self.store)
+        plan = dry.plan()
+        self.assertIn("c1", plan["ready_for_search"])
+        self.assertIn("c2", plan["search_retry_pending"])
+
+
+class TestGuarantees(unittest.TestCase):
+    """C5D-Guarantees (statisch)."""
+
+    def test_no_tolaria_write(self):
+        res = assert_no_tolaria_write()
+        self.assertTrue(res["no_tolaria_write"], res)
+
+    def test_no_master_write(self):
+        res = assert_no_master_write()
+        self.assertTrue(res["no_master_write"], res)
+
+    def test_no_production_activation(self):
+        res = assert_no_production_activation()
+        self.assertTrue(res["no_production_activation"], res)
+
+
+class TestSecretSafety(unittest.TestCase):
+    """Secret Safety: C5D persistiert/ausgibt keine Secrets."""
+
+    def test_detect_secret_blocks(self):
+        # detect_secret aus C5B erkennt Secrets (Regression)
+        self.assertIsNotNone(detect_secret("sk-1234567890abcdefghijklmnopqrstuvwxyz"))
+        self.assertIsNone(detect_secret("normal content ohne secret"))
+
+    def test_no_secret_in_c5d_source(self):
+        # C5D-Quelle darf keine echten Secret-Werte enthalten
+        import rq_c5d
+        with open(rq_c5d.__file__, encoding="utf-8") as f:
+            src = f.read()
+        # Kein sk-<20+> Muster im Quellcode (nur fragmentierte Referenzen)
+        import re
+        self.assertIsNone(re.search(r"sk-[A-Za-z0-9]{20,}", src))
+
+
+class TestRegressionContracts(unittest.TestCase):
+    """Regression der bestehenden C5A/C5B/C5C-Contracts."""
+
+    def test_c5c_guarantees_still_pass(self):
+        # C5C-Guarantees (No-Search, No-Master-Write) bleiben intakt
+        search = assert_no_search_calls()
+        master = assert_no_master_write()
+        self.assertTrue(search["no_search_calls"])
+        self.assertTrue(master["no_master_write"])
+
+    def test_c5a_transitions_include_search_retry_replay(self):
+        # C5D-Transition (RETRY_PENDING -> UPDATING_SEARCH) muss existieren
+        from rq_c5a import _ALLOWED_TRANSITIONS
+        self.assertIn((ST_RETRY_PENDING, ST_UPDATING_SEARCH), _ALLOWED_TRANSITIONS)
+
+    def test_c5a_transitions_include_apply_path(self):
+        from rq_c5a import _ALLOWED_TRANSITIONS
+        self.assertIn((ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH), _ALLOWED_TRANSITIONS)
+        self.assertIn((ST_VERIFYING_SEARCH, ST_APPLIED), _ALLOWED_TRANSITIONS)
+
+    def test_c5c_propagator_still_works(self):
+        # C5C-Propagator (Tolaria-Propagation) bleibt funktional
+        fake = FakeTolaria()
+        client = FakeTolariaClient(fake)
+        repo = FixtureRepo()
+        content = _fm(UUID_A, type="arch", role="module", representation="source", state="current")
+        repo.write("modul-09.md", content)
+        sha = repo.commit("create")
+        store = _make_store()
+        reader = GitReader(repo.dir)
+        _seed_commit(store, sha, None, [{
+            "object_id": UUID_A, "operation": OP_CREATE,
+            "path_before": None, "path_after": "modul-09.md",
+            "content_hash_after": content_hash(parse_frontmatter(content)[1]),
+        }], status=ST_READY)
+        prop = C5CPropagator(store, reader, client)
+        res = prop.propagate_commit(sha)
+        self.assertEqual(res["status"], ST_UPDATING_SEARCH)
+        self.assertEqual(fake.vault["/app/vault/modul-09.md"], content)
+        repo.cleanup()
+        store.close()
+
+
+if __name__ == "__main__":
+    unittest.main()