diff --git a/tolaria/c5-sync-service/rq_c5_cli.py b/tolaria/c5-sync-service/rq_c5_cli.py index be1cf37..5f2c5bc 100644 --- a/tolaria/c5-sync-service/rq_c5_cli.py +++ b/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) diff --git a/tolaria/c5-sync-service/rq_c5a.py b/tolaria/c5-sync-service/rq_c5a.py index f710d25..78ae472 100644 --- a/tolaria/c5-sync-service/rq_c5a.py +++ b/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 diff --git a/tolaria/c5-sync-service/rq_c5d.py b/tolaria/c5-sync-service/rq_c5d.py new file mode 100644 index 0000000..af7b114 --- /dev/null +++ b/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)), + } diff --git a/tolaria/c5-sync-service/test_c5d.py b/tolaria/c5-sync-service/test_c5d.py new file mode 100644 index 0000000..0d204b5 --- /dev/null +++ b/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()