diff --git a/tolaria/c5-sync-service/rq_c5_cli.py b/tolaria/c5-sync-service/rq_c5_cli.py index 9e55e9b..be1cf37 100644 --- a/tolaria/c5-sync-service/rq_c5_cli.py +++ b/tolaria/c5-sync-service/rq_c5_cli.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -Red Queen — C5 CLI (C5A + C5B): kontrollierte Inspektion & Dry-Run. +Red Queen — C5 CLI (C5A + C5B + C5C): kontrollierte Inspektion & Dry-Run. KEIN produktiver Poll-Daemon. KEINE daemon/start-forever-Funktion. Nur explizite, begrenzte Befehle: @@ -10,6 +10,9 @@ Nur explizite, begrenzte Befehle: * dry-run-diff — Diff eines Commits anzeigen (read-only) * health — Health-Contract anzeigen * no-write-check — No-Downstream-Write-Guarantee pruefen + * propagate-plan — Propagation-Plan fuer einen Commit (read-only, KEIN Write) + * live-dry-run — Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand + * c5c-guarantees — No-Search- und No-Master-Write-Guarantee pruefen (statisch) """ from __future__ import annotations @@ -28,6 +31,15 @@ from rq_c5b import ( DEFAULT_POLL_INTERVAL_SECONDS, ENV_POLL_INTERVAL, ) +from rq_c5c import ( + C5CPropagator, + C5CDryRun, + TolariaClient, + assert_no_search_calls, + assert_no_master_write, + DEFAULT_TOLARIA_BASE, + ENV_TOLARIA_BASE, +) def _store(db_path: str) -> C5AStore: @@ -91,6 +103,75 @@ def cmd_no_write_check(args: argparse.Namespace) -> int: return 0 if result["no_downstream_write"] else 1 +def _c5c_client(args: argparse.Namespace) -> TolariaClient: + return TolariaClient(base_url=args.tolaria_base) + + +def cmd_propagate_plan(args: argparse.Namespace) -> int: + """Erzeugt den Propagation-Plan fuer einen Commit (read-only, KEIN Write).""" + store = _store(args.db) + reader = GitReader(args.repo) + dry = C5CDryRun(store, reader, client=_c5c_client(args)) + plan = dry.plan_commit(args.sha) + print(json.dumps(plan, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_live_dry_run(args: argparse.Namespace) -> int: + """Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand. + + Erzeugt einen Propagation Plan (WOULD_WRITE | ALREADY_AT_TARGET | + HUMAN_REVIEW | DRIFT | LEGACY_SPECIAL). Fuehrt KEINE produktive + Propagation aus. KEINE Writes. + """ + store = _store(args.db) + reader = GitReader(args.repo) + dry = C5CDryRun(store, reader, client=_c5c_client(args)) + commits = store.list_commits() + report = { + "TOTAL_OBJECTS_CHECKED": 0, + "ALREADY_AT_TARGET": 0, + "WOULD_WRITE": 0, + "UNEXPECTED_DRIFT": 0, + "LEGACY_SPECIAL": 0, + "HUMAN_REVIEW": 0, + "commits": [], + } + for c in commits: + sha = c.get("commit_sha") or c.get("sha") + if not sha: + continue + plan = dry.plan_commit(sha) + for item in plan.get("plan", []): + report["TOTAL_OBJECTS_CHECKED"] += 1 + p = item.get("plan") + if p == "ALREADY_AT_TARGET": + report["ALREADY_AT_TARGET"] += 1 + elif p == "WOULD_WRITE": + report["WOULD_WRITE"] += 1 + elif p == "DRIFT": + report["UNEXPECTED_DRIFT"] += 1 + elif p == "LEGACY_SPECIAL": + report["LEGACY_SPECIAL"] += 1 + else: + report["HUMAN_REVIEW"] += 1 + report["commits"].append({"commit_sha": sha, "plan": plan.get("plan", [])}) + print(json.dumps(report, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_c5c_guarantees(args: argparse.Namespace) -> int: + """Prueft No-Search- und No-Master-Write-Guarantee (statisch).""" + search = assert_no_search_calls() + master = assert_no_master_write() + result = {"no_search_calls": search, "no_master_write": master} + print(json.dumps(result, ensure_ascii=False, indent=2)) + ok = search["no_search_calls"] and master["no_master_write"] + return 0 if ok else 1 + + def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser( prog="rq_c5_cli", @@ -108,6 +189,10 @@ def main(argv: Optional[List[str]] = None) -> int: "--interval", type=int, default=None, help=f"Poll-Intervall in Sekunden (Env {ENV_POLL_INTERVAL}, Default {DEFAULT_POLL_INTERVAL_SECONDS})", ) + parser.add_argument( + "--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})", + ) sub = parser.add_subparsers(dest="command", required=True) @@ -131,6 +216,16 @@ def main(argv: Optional[List[str]] = None) -> int: p = sub.add_parser("no-write-check", help="No-Downstream-Write-Guarantee pruefen") p.set_defaults(func=cmd_no_write_check) + p = sub.add_parser("propagate-plan", help="Propagation-Plan fuer einen Commit (read-only, KEIN Write)") + p.add_argument("sha", help="Commit-SHA") + p.set_defaults(func=cmd_propagate_plan) + + p = sub.add_parser("live-dry-run", help="Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand") + p.set_defaults(func=cmd_live_dry_run) + + p = sub.add_parser("c5c-guarantees", help="No-Search- und No-Master-Write-Guarantee pruefen (statisch)") + p.set_defaults(func=cmd_c5c_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 ec3ab0a..ddca63b 100644 --- a/tolaria/c5-sync-service/rq_c5a.py +++ b/tolaria/c5-sync-service/rq_c5a.py @@ -175,6 +175,7 @@ _ALLOWED_TRANSITIONS = { (ST_READY, ST_PROPAGATING_TOLARIA), (ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA), (ST_PROPAGATING_TOLARIA, ST_RETRY_PENDING), # Tolaria-Write-Fehler + (ST_PROPAGATING_TOLARIA, ST_HUMAN_REVIEW_REQUIRED), # C5C: Drift/Secret/dangling waehrend Propagation (ST_VERIFYING_TOLARIA, ST_UPDATING_SEARCH), (ST_VERIFYING_TOLARIA, ST_RETRY_PENDING), # DRIFT != 0 / Tolaria-Fehler (ST_VERIFYING_TOLARIA, ST_HUMAN_REVIEW_REQUIRED), # unexpected drift diff --git a/tolaria/c5-sync-service/rq_c5c.py b/tolaria/c5-sync-service/rq_c5c.py new file mode 100644 index 0000000..91132bd --- /dev/null +++ b/tolaria/c5-sync-service/rq_c5c.py @@ -0,0 +1,820 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5C Tolaria Propagation & Drift Verification Engine. + +C5C implementiert ausschliesslich: + VALIDIERTE OBJECT CHANGES -> TOLARIA PROPAGATION -> READ-BACK + -> HASH-/METADATA-VERIFY -> DRIFT DETECTION -> STATE-MACHINE UPDATE + +ARCHITEKTURREGEL (verbindlich): + * Forgejo bleibt MASTER / Source of Truth. + * Tolaria ist ausschliesslich DERIVED. + * C5C darf: Forgejo lesen, Tolaria kontrolliert schreiben. + * C5C darf NIEMALS: Tolaria -> Forgejo schreiben, Forgejo veraendern, + IDs neu vergeben, Knowledge Content eigenmaessig umformulieren. + +NO-SEARCH-GUARANTEE: C5C ruft NICHT /api/search/rebuild auf. +NO-MASTER-WRITE-GUARANTEE: C5C fuehrt KEIN git push / Forgejo-Write aus. +NO-PRODUCTION-PROPAGATION: C5C fuehrt KEINE produktive Propagation aus + (nur Tests gegen Fake/Mock + read-only Live-Dry-Run). + +Write-Komponente (TolariaClient) ist klar isoliert. Nur C5C-Writer darf den +Tolaria-Write-Pfad verwenden. Keine generische Agent-Write-Funktion. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +# C5A / C5B wiederverwenden (keine konkurrierende State Machine) +from rq_c5a import ( + C5AStore, + ST_READY, + ST_PROPAGATING_TOLARIA, + ST_VERIFYING_TOLARIA, + ST_UPDATING_SEARCH, + ST_HUMAN_REVIEW_REQUIRED, + ST_RETRY_PENDING, + ST_DEAD, + ST_APPLIED, + IDEM_ALREADY_AT_TARGET, + IDEM_ALREADY_APPLIED, + IDEM_RETRY_SAFE, + IDEM_CONFLICT, + RC_UNEXPECTED_TOLARIA_DRIFT, + RC_UNKNOWN_OBJECT_ID, + RC_INVALID_SCHEMA, + RC_DANGLING_DERIVED_FROM, + RC_AMBIGUOUS_DELETE, + RC_UNKNOWN_LEGACY_OBJECT, + RC_SECRET_DETECTED, + RC_TOLARIA_UNAVAILABLE, + RC_ID_COLLISION, + RC_AUTH_FAILURE, + DEFAULT_MAX_RETRIES, + DEFAULT_BACKOFF_SECONDS, + OP_CREATE, + OP_CONTENT_UPDATE, + OP_METADATA_UPDATE, + OP_STATE_UPDATE, + OP_TAGS_UPDATE, + OP_SOURCE_CANONICAL_RELATION_UPDATE, + OP_RENAME, + OP_MOVE, + OP_DELETE_REQUEST, + OP_SUPERSEDE, +) +from rq_c5b import ( + GitReader, + parse_frontmatter, + extract_object_id, + content_hash, + metadata_hash, + detect_secret, + SCOPE_LEGACY_SPECIAL, +) + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +# Tolaria-API-Basis (Default; per Env C5_TOLARIA_BASE ueberschreibbar) +DEFAULT_TOLARIA_BASE = "http://187.124.31.123:5173/api/vault" +ENV_TOLARIA_BASE = "C5_TOLARIA_BASE" + +# Vault-Pfad-Praefix (verbindlich — ohne Praefix liest /content das Tolaria-eigene +# README statt des Vault-Objekts -> false drift) +VAULT_PREFIX = "/app/vault" + +# Pre-Write-Drift-Ergebnisse +DRIFT_WRITE_ALLOWED = "WRITE_ALLOWED" +DRIFT_ALREADY_AT_TARGET = "ALREADY_AT_TARGET" +DRIFT_UNEXPECTED = "UNEXPECTED_TOLARIA_DRIFT" + +# Propagation-Plan-Klassifikation (Live-Dry-Run, §20) +PLAN_WOULD_WRITE = "WOULD_WRITE" +PLAN_ALREADY_AT_TARGET = "ALREADY_AT_TARGET" +PLAN_HUMAN_REVIEW = "HUMAN_REVIEW" +PLAN_DRIFT = "DRIFT" +PLAN_LEGACY_SPECIAL = "LEGACY_SPECIAL" + +# Operationen, die C5C automatisch propagieren darf (kein Human Gate) +_AUTO_PROPAGATE_OPS = frozenset({ + OP_CREATE, + OP_CONTENT_UPDATE, + OP_METADATA_UPDATE, + OP_STATE_UPDATE, + OP_TAGS_UPDATE, + OP_SOURCE_CANONICAL_RELATION_UPDATE, + OP_SUPERSEDE, +}) + +# Operationen, die IMMER Human Gate erfordern (C5C darf sie nie automatisch anwenden) +_HUMAN_GATE_OPS = frozenset({ + OP_DELETE_REQUEST, # §11: niemals automatisch + OP_RENAME, # §10: API-Semantik nicht sicher verifizierbar -> Human Gate + OP_MOVE, # §10: dito +}) + +# Reason Codes, die NICHT retrybar sind (direkt Human Gate / Fail Closed, §17) +_NON_RETRYABLE_REASONS = frozenset({ + RC_UNEXPECTED_TOLARIA_DRIFT, + RC_INVALID_SCHEMA, + RC_UNKNOWN_OBJECT_ID, + RC_SECRET_DETECTED, + RC_DANGLING_DERIVED_FROM, + RC_AMBIGUOUS_DELETE, + RC_ID_COLLISION, + RC_UNKNOWN_LEGACY_OBJECT, + RC_AUTH_FAILURE, +}) + + +class C5CError(Exception): + """Basis-Fehler fuer C5C.""" + + def __init__(self, message: str, reason_code: Optional[str] = None): + super().__init__(message) + self.message = message + self.reason_code = reason_code + + def to_dict(self) -> Dict[str, Any]: + return {"error": self.message, "reason_code": self.reason_code} + + +class TolariaUnavailableError(C5CError): + """Tolaria nicht erreichbar / Timeout / transienter HTTP-Fehler (retrybar).""" + + +class TolariaWriteError(C5CError): + """Tolaria-Write fehlgeschlagen (nicht retrybar, z.B. Auth/Schema).""" + + +class UnexpectedDriftError(C5CError): + """Tolaria CURRENT entspricht weder BEFORE noch TARGET (fail closed).""" + + +class ReadBackMismatchError(C5CError): + """Read-Back nach Write stimmt nicht mit dem erwarteten Ziel ueberein.""" + + +class NoSearchGuaranteeError(C5CError): + """C5C darf Search-Rebuild nicht aufrufen.""" + + +class NoMasterWriteGuaranteeError(C5CError): + """C5C darf nicht nach Forgejo schreiben.""" + + +# --------------------------------------------------------------------------- +# Tolaria Client (isoliert: read / write / verify) +# --------------------------------------------------------------------------- + +class TolariaClient: + """ + Isolierter Tolaria-Client. Nur C5C-Writer darf write() verwenden. + + read() — POST /content (Read-Back) + list() — POST /list (Metadaten) + write() — POST /save (NUR C5C-Writer; isolierte Write-Komponente) + verify() — Read-Back + Hash-/Metadata-Vergleich + + Keine generische Agent-Write-Funktion. Kein Search-Rebuild-Aufruf. + """ + + def __init__(self, base_url: Optional[str] = None, timeout: float = 15.0): + self.base_url = (base_url or os.environ.get(ENV_TOLARIA_BASE) + or DEFAULT_TOLARIA_BASE).rstrip("/") + self.timeout = timeout + + # -- HTTP-Helfer -------------------------------------------------------- + + def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: + url = f"{self.base_url}/{endpoint.lstrip('/')}" + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, data=data, headers={"Content-Type": "application/json"}, + 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: + # 4xx/5xx: transient (5xx) vs. nicht-retrybar (4xx Auth/Schema) + if e.code >= 500: + raise TolariaUnavailableError( + f"Tolaria HTTP {e.code} auf {endpoint}", RC_TOLARIA_UNAVAILABLE) + raise TolariaWriteError( + f"Tolaria HTTP {e.code} auf {endpoint}: {e.read().decode('utf-8', 'replace')[:200]}", + RC_AUTH_FAILURE if e.code in (401, 403) else RC_INVALID_SCHEMA) + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise TolariaUnavailableError( + f"Tolaria nicht erreichbar ({endpoint}): {e}", RC_TOLARIA_UNAVAILABLE) + + # -- Read --------------------------------------------------------------- + + def read(self, vault_path: str) -> Optional[str]: + """Read-Back eines Vault-Objekts. Gibt Inhalt oder None (nicht vorhanden).""" + resp = self._post("content", {"path": vault_path}) + if "error" in resp: + return None # Invalid or missing path -> Objekt nicht vorhanden + return resp.get("content") + + def list(self, vault_path: str = VAULT_PREFIX) -> List[Dict[str, Any]]: + """Listet Vault-Metadaten (read-only).""" + resp = self._post("list", {"path": vault_path}) + if isinstance(resp, list): + return resp + return [] + + # -- Write (NUR C5C-Writer) -------------------------------------------- + + def write(self, vault_path: str, content: str) -> Dict[str, Any]: + """Schreibt ein Vault-Objekt (POST /save). Isolierte Write-Komponente.""" + resp = self._post("save", {"path": vault_path, "content": content}) + if "error" in resp: + raise TolariaWriteError( + f"Tolaria save fehlgeschlagen: {resp['error']}", RC_INVALID_SCHEMA) + return resp + + # -- Verify (Read-Back + Hash) ------------------------------------------ + + def verify(self, vault_path: str, expected_content: str) -> Dict[str, Any]: + """ + Read-Back + Hash-/Metadata-Verify nach einem Write. + + Vergleicht: path, content_hash, metadata_hash, representation, state, + derived_from, tags. Kein Write gilt ohne Read-Back als erfolgreich. + """ + rb = self.read(vault_path) + if rb is None: + raise ReadBackMismatchError( + f"Read-Back lieferte kein Objekt fuer {vault_path}", + RC_UNEXPECTED_TOLARIA_DRIFT) + fm_rb, body_rb = parse_frontmatter(rb) + fm_exp, body_exp = parse_frontmatter(expected_content) + + content_ok = content_hash(body_rb) == content_hash(body_exp) + metadata_ok = metadata_hash(fm_rb) == metadata_hash(fm_exp) + + # Feld-fuer-Feld-Vergleich (fuer Diagnose) + fields = { + "object_id": (fm_rb.get("id"), fm_exp.get("id")), + "knowledge_schema": (fm_rb.get("knowledge_schema"), fm_exp.get("knowledge_schema")), + "type": (fm_rb.get("type"), fm_exp.get("type")), + "role": (fm_rb.get("role"), fm_exp.get("role")), + "representation": (fm_rb.get("representation"), fm_exp.get("representation")), + "state": (fm_rb.get("state"), fm_exp.get("state")), + "derived_from": (fm_rb.get("derived_from"), fm_exp.get("derived_from")), + "tags": (fm_rb.get("tags"), fm_exp.get("tags")), + } + mismatches = {k: v for k, v in fields.items() if v[0] != v[1]} + + ok = content_ok and metadata_ok and not mismatches + return { + "ok": ok, + "path": vault_path, + "content_hash_match": content_ok, + "metadata_hash_match": metadata_ok, + "field_mismatches": mismatches, + } + + +# --------------------------------------------------------------------------- +# Pre-Write Drift Check (harte Invariante, §5) +# --------------------------------------------------------------------------- + +def pre_write_drift_check( + client: TolariaClient, + vault_path: str, + before_content: Optional[str], + target_content: str, +) -> Tuple[str, Optional[str]]: + """ + Vor JEDEM Tolaria-Write: Forgejo TARGET gegen Tolaria CURRENT vergleichen. + + * Tolaria CURRENT == erwarteter BEFORE-Zustand -> WRITE_ALLOWED + * Tolaria CURRENT == TARGET exakt -> ALREADY_AT_TARGET (kein Write) + * Tolaria CURRENT weder BEFORE noch TARGET -> UNEXPECTED_TOLARIA_DRIFT (FAIL CLOSED) + + Rueckgabe: (DRIFT_*, current_content) + """ + current = client.read(vault_path) + + # Tolaria CURRENT == TARGET -> bereits am Ziel, kein Write + if current is not None and _content_equal(current, target_content): + return DRIFT_ALREADY_AT_TARGET, current + + # Tolaria CURRENT == erwarteter BEFORE -> Write erlaubt + if before_content is not None and current is not None and _content_equal(current, before_content): + return DRIFT_WRITE_ALLOWED, current + + # CREATE-Fall: kein BEFORE, Tolaria hat das Objekt noch nicht -> Write erlaubt + if before_content is None and current is None: + return DRIFT_WRITE_ALLOWED, current + + # Alles andere -> UNEXPECTED_TOLARIA_DRIFT (fail closed, kein Ueberschreiben) + return DRIFT_UNEXPECTED, current + + +def _content_equal(a: str, b: str) -> bool: + """Byte-genauer Vergleich (keine Transformation, kein LLM Rewrite).""" + return a == b + + +# --------------------------------------------------------------------------- +# Propagation-Modelle (pro Operation) +# --------------------------------------------------------------------------- + +def _vault_path(rel_path: str) -> str: + """Konvertiert einen Repo-relativen Pfad in einen Vault-Pfad.""" + return f"{VAULT_PREFIX}/{rel_path.lstrip('/')}" + + +def _read_forgejo_after(reader: GitReader, sha: str, path_after: Optional[str]) -> Optional[str]: + """Liest den exakten Forgejo-AFTER-Inhalt (keine Transformation).""" + if not path_after: + return None + return reader.file_content(sha, path_after) + + +def _read_forgejo_before(reader: GitReader, parent_sha: Optional[str], path_before: Optional[str]) -> Optional[str]: + """Liest den Forgejo-BEFORE-Inhalt (fuer Pre-Write-Drift-Check).""" + if not path_before or not parent_sha: + return None + return reader.file_content(parent_sha, path_before) + + +def _validate_create(obj: Dict[str, Any], after_content: str) -> Optional[str]: + """§6 CREATE-Validierung. Gibt reason_code zurueck oder None (gueltig).""" + oid = obj.get("object_id") + if not oid or not oid.startswith("object/"): + return RC_UNKNOWN_OBJECT_ID + fm, _ = parse_frontmatter(after_content) + if not fm.get("knowledge_schema"): + return RC_INVALID_SCHEMA + if detect_secret(after_content): + return RC_SECRET_DETECTED + derived = fm.get("derived_from") + if derived and not derived.startswith("object/"): + return RC_DANGLING_DERIVED_FROM + return None + + +def _validate_derived_from(fm: Dict[str, Any], existing_ids: set) -> Optional[str]: + """§9: Keine dangling derived_from. Ziel muss produktiv existieren.""" + derived = fm.get("derived_from") + if derived and derived not in existing_ids: + return RC_DANGLING_DERIVED_FROM + return None + + +# --------------------------------------------------------------------------- +# C5C Propagation Engine +# --------------------------------------------------------------------------- + +class C5CPropagator: + """ + Propagiert validierte ObjectChanges eines Forgejo-Commits nach Tolaria. + + Commit-Atomicity (§15): Ein Commit ist die Processing Unit. Alle Tolaria-Steps + muessen PASS sein, bevor der Commit zum naechsten C5-Schritt darf. Wenn ein + Objekt fehlschlaegt, wird der Commit NICHT als Tolaria-complete markiert. + + C5C darf States bewegen: READY -> PROPAGATING_TOLARIA -> VERIFYING_TOLARIA + -> UPDATING_SEARCH (READY_FOR_SEARCH). C5C darf NICHT VERIFYING_SEARCH -> + APPLIED durchfuehren. last_applied_commit bleibt unveraendert. + """ + + def __init__( + self, + store: C5AStore, + reader: GitReader, + client: Optional[TolariaClient] = None, + max_retries: int = DEFAULT_MAX_RETRIES, + backoff_seconds: Optional[List[int]] = None, + ): + self.store = store + self.reader = reader + self.client = client or TolariaClient() + self.max_retries = max_retries + self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS + + # -- Idempotenz --------------------------------------------------------- + + def _commit_already_propagated(self, commit_sha: str) -> bool: + """Commit wurde bereits vollstaendig propagiert (Tolaria-complete).""" + return self.store.commit_status(commit_sha) in ( + ST_UPDATING_SEARCH, ST_APPLIED) + + # -- Retry-Klassifikation (§17) ----------------------------------------- + + def _is_retryable(self, err: C5CError) -> bool: + """Nur technische, retryable Fehler nutzen das Retry-Modell.""" + if isinstance(err, TolariaUnavailableError): + return True + if err.reason_code in _NON_RETRYABLE_REASONS: + return False + # Default: nicht-retrybar (fail closed) + return False + + def _handle_failure(self, commit_sha: str, err: C5CError) -> Dict[str, Any]: + """Behandelt einen Fehler: Retry, DEAD oder Human Gate.""" + if self._is_retryable(err): + retry = self.store.increment_retry(commit_sha) + if retry >= self.max_retries: + self.store.set_commit_error(commit_sha, err.reason_code or RC_TOLARIA_UNAVAILABLE, 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_TOLARIA_UNAVAILABLE, err.message) + self.store.transition_commit(commit_sha, ST_RETRY_PENDING) + 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_UNEXPECTED_TOLARIA_DRIFT + 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} + + # -- Einzelnes Objekt propagieren --------------------------------------- + + def _propagate_object(self, commit_sha: str, parent_sha: Optional[str], + obj: Dict[str, Any], existing_ids: set) -> Dict[str, Any]: + """ + Propagiert EIN ObjectChange nach Tolaria. Gibt ein Ergebnis-Dict zurueck. + + Rueckgabe-Felder: object_id, operation, status, reason_code (optional), + drift (optional), idempotency (optional). + """ + oid = obj.get("object_id") + op = obj.get("operation") + + # Human-Gate-Operationen (DELETE/RENAME/MOVE) -> niemals automatisch + if op in _HUMAN_GATE_OPS: + rc = RC_AMBIGUOUS_DELETE if op == OP_DELETE_REQUEST else RC_UNKNOWN_OBJECT_ID + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": rc} + + # Nicht-auto-propagierbare Operation -> Human Gate + if op not in _AUTO_PROPAGATE_OPS: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_INVALID_SCHEMA} + + # object_id muss vorhanden sein + if not oid: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_UNKNOWN_OBJECT_ID} + + path_after = obj.get("path_after") + path_before = obj.get("path_before") + if not path_after: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_UNKNOWN_OBJECT_ID} + + # Forgejo-AFTER-Inhalt lesen (exakt, keine Transformation) + after_content = _read_forgejo_after(self.reader, commit_sha, path_after) + if after_content is None: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_UNKNOWN_OBJECT_ID} + + # Secret-Safety (fail-closed VOR Verarbeitung) + if detect_secret(after_content): + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_SECRET_DETECTED} + + # CREATE-Validierung (§6) + if op == OP_CREATE: + rc = _validate_create(obj, after_content) + if rc: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": rc} + + # derived_from-Validierung (§9): keine dangling relation + fm_after, _ = parse_frontmatter(after_content) + rc = _validate_derived_from(fm_after, existing_ids) + if rc: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": rc} + + vault_path = _vault_path(path_after) + before_content = _read_forgejo_before(self.reader, parent_sha, path_before) + + # Pre-Write Drift Check (§5) — harte Invariante. + # Idempotenz wird HIER bestimmt: Tolaria CURRENT == TARGET -> ALREADY_AT_TARGET + # (kein Write). Tolaria CURRENT == BEFORE -> WRITE_ALLOWED. + drift, current = pre_write_drift_check( + self.client, vault_path, before_content, after_content) + if drift == DRIFT_ALREADY_AT_TARGET: + return {"object_id": oid, "operation": op, "status": ST_VERIFYING_TOLARIA, + "idempotency": IDEM_ALREADY_AT_TARGET} + if drift == DRIFT_UNEXPECTED: + return {"object_id": oid, "operation": op, "status": ST_HUMAN_REVIEW_REQUIRED, + "reason_code": RC_UNEXPECTED_TOLARIA_DRIFT, "drift": True} + + # WRITE_ALLOWED -> Write (nur C5C-Writer) + self.client.write(vault_path, after_content) + + # Read-Back + Verify (§14) — kein Write gilt ohne Read-Back als erfolgreich + verify = self.client.verify(vault_path, after_content) + if not verify["ok"]: + raise ReadBackMismatchError( + f"Read-Back-Mismatch auf {vault_path}: {verify['field_mismatches']}", + RC_UNEXPECTED_TOLARIA_DRIFT) + + return {"object_id": oid, "operation": op, "status": ST_VERIFYING_TOLARIA, + "idempotency": IDEM_RETRY_SAFE} + + # -- Commit propagieren (Atomicity, §15) -------------------------------- + + def propagate_commit(self, commit_sha: str) -> Dict[str, Any]: + """ + Propagiert alle ObjectChanges eines Commits nach Tolaria. + + Commit-Atomicity: Alle Objekte muessen PASS sein, bevor der Commit zum + naechsten C5-Schritt (UPDATING_SEARCH / READY_FOR_SEARCH) darf. Wenn ein + Objekt fehlschlaegt, wird der Commit NICHT als Tolaria-complete markiert. + + C5C bewegt: READY -> PROPAGATING_TOLARIA -> VERIFYING_TOLARIA + -> UPDATING_SEARCH (READY_FOR_SEARCH). NICHT weiter zu APPLIED. + last_applied_commit bleibt 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") + # Nur aus READY oder RETRY_PENDING propagieren + if cur not in (ST_READY, ST_RETRY_PENDING): + return {"commit_sha": commit_sha, "status": cur, + "message": "Commit nicht im propagierbaren Zustand"} + + # READY -> PROPAGATING_TOLARIA + self.store.transition_commit(commit_sha, ST_PROPAGATING_TOLARIA) + + parent_sha = commit.get("parent_sha") + objs = self.store.list_object_changes(commit_sha) + + # Existierende IDs sammeln (fuer derived_from-Validierung, §9) + existing_ids = self._collect_existing_ids(commit_sha) + + results = [] + all_pass = True + for obj in objs: + try: + r = self._propagate_object(commit_sha, parent_sha, obj, existing_ids) + except C5CError as e: + r = self._handle_failure(commit_sha, e) + all_pass = False + results.append(r) + break # Commit-Atomicity: bei Fehler Commit nicht als complete markieren + results.append(r) + if r.get("status") == ST_HUMAN_REVIEW_REQUIRED: + all_pass = False + # Human Gate -> Commit-Status setzen (fail closed) + rc = r.get("reason_code") or RC_UNEXPECTED_TOLARIA_DRIFT + self.store.set_commit_error(commit_sha, rc, f"ObjectChange {r.get('object_id')} {r.get('operation')}") + self.store.transition_commit(commit_sha, ST_HUMAN_REVIEW_REQUIRED) + break # Commit nicht als complete markieren + + if not all_pass: + # Commit bleibt in PROPAGATING_TOLARIA oder wird Human Gate / Retry + # (transition_commit wurde bereits im Fehler-Handler gesetzt) + return {"commit_sha": commit_sha, "status": self.store.commit_status(commit_sha), + "results": results} + + # Alle Objekte PASS -> VERIFYING_TOLARIA + self.store.transition_commit(commit_sha, ST_VERIFYING_TOLARIA) + + # VERIFYING_TOLARIA -> UPDATING_SEARCH (READY_FOR_SEARCH) + # C5C darf NICHT weiter zu VERIFYING_SEARCH -> APPLIED. + self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH) + + return {"commit_sha": commit_sha, "status": ST_UPDATING_SEARCH, + "results": results, "ready_for_search": True} + + def _collect_existing_ids(self, commit_sha: str) -> set: + """Sammelt alle bereits produktiv existierenden object_ids (fuer §9).""" + ids = set() + for c in self.store.list_commits(): + for oc in self.store.list_object_changes(c["commit_sha"]): + oid = oc.get("object_id") + if oid: + ids.add(oid) + return ids + + +# --------------------------------------------------------------------------- +# Live-Dry-Run (read-only, §20) — erzeugt Propagation Plan, KEINE Writes +# --------------------------------------------------------------------------- + +class C5CDryRun: + """ + Read-only Live-Dry-Run gegen den realen Forgejo + Tolaria Stand. + + Erzeugt einen Propagation Plan: + WOULD_WRITE | ALREADY_AT_TARGET | HUMAN_REVIEW | DRIFT | LEGACY_SPECIAL + + Fuehrt KEINE produktive Propagation aus. KEINE Writes. + """ + + def __init__(self, store: C5AStore, reader: GitReader, client: Optional[TolariaClient] = None): + self.store = store + self.reader = reader + self.client = client or TolariaClient() + + def plan_commit(self, commit_sha: str) -> Dict[str, Any]: + """Erzeugt den Propagation-Plan fuer einen Commit (read-only).""" + commit = self.store.get_commit(commit_sha) + if commit is None: + return {"commit_sha": commit_sha, "error": "Commit nicht gefunden"} + parent_sha = commit.get("parent_sha") + objs = self.store.list_object_changes(commit_sha) + existing_ids = self._collect_existing_ids() + + plan = [] + for obj in objs: + plan.append(self._plan_object(commit_sha, parent_sha, obj, existing_ids)) + + return {"commit_sha": commit_sha, "plan": plan} + + def _plan_object(self, commit_sha: str, parent_sha: Optional[str], + obj: Dict[str, Any], existing_ids: set) -> Dict[str, Any]: + oid = obj.get("object_id") + op = obj.get("operation") + + # Human-Gate-Operationen + if op in _HUMAN_GATE_OPS: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_AMBIGUOUS_DELETE if op == OP_DELETE_REQUEST else RC_UNKNOWN_OBJECT_ID} + if op not in _AUTO_PROPAGATE_OPS: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_INVALID_SCHEMA} + if not oid: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_UNKNOWN_OBJECT_ID} + + path_after = obj.get("path_after") + if not path_after: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_UNKNOWN_OBJECT_ID} + + after_content = _read_forgejo_after(self.reader, commit_sha, path_after) + if after_content is None: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_UNKNOWN_OBJECT_ID} + if detect_secret(after_content): + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": RC_SECRET_DETECTED} + if op == OP_CREATE: + rc = _validate_create(obj, after_content) + if rc: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": rc} + fm_after, _ = parse_frontmatter(after_content) + rc = _validate_derived_from(fm_after, existing_ids) + if rc: + return {"object_id": oid, "operation": op, "plan": PLAN_HUMAN_REVIEW, + "reason_code": rc} + + vault_path = _vault_path(path_after) + before_content = _read_forgejo_before(self.reader, parent_sha, path_before=obj.get("path_before")) + + # Pre-Write Drift Check (read-only) + drift, _ = pre_write_drift_check(self.client, vault_path, before_content, after_content) + if drift == DRIFT_ALREADY_AT_TARGET: + return {"object_id": oid, "operation": op, "plan": PLAN_ALREADY_AT_TARGET} + if drift == DRIFT_UNEXPECTED: + return {"object_id": oid, "operation": op, "plan": PLAN_DRIFT, + "reason_code": RC_UNEXPECTED_TOLARIA_DRIFT} + return {"object_id": oid, "operation": op, "plan": PLAN_WOULD_WRITE} + + def _collect_existing_ids(self) -> set: + ids = set() + for c in self.store.list_commits(): + for oc in self.store.list_object_changes(c["commit_sha"]): + oid = oc.get("object_id") + if oid: + ids.add(oid) + return ids + + def drift_report(self, commits: List[str]) -> Dict[str, Any]: + """Erzeugt den Drift Report (§21) aus dem Live-Dry-Run.""" + total = 0 + counts = {PLAN_WOULD_WRITE: 0, PLAN_ALREADY_AT_TARGET: 0, + PLAN_HUMAN_REVIEW: 0, PLAN_DRIFT: 0, PLAN_LEGACY_SPECIAL: 0} + drift_objects = [] + for sha in commits: + res = self.plan_commit(sha) + for item in res.get("plan", []): + total += 1 + p = item.get("plan") + counts[p] = counts.get(p, 0) + 1 + if p == PLAN_DRIFT: + drift_objects.append(item) + return { + "TOTAL_OBJECTS_CHECKED": total, + "ALREADY_AT_TARGET": counts[PLAN_ALREADY_AT_TARGET], + "WOULD_WRITE": counts[PLAN_WOULD_WRITE], + "UNEXPECTED_DRIFT": counts[PLAN_DRIFT], + "LEGACY_SPECIAL": counts[PLAN_LEGACY_SPECIAL], + "HUMAN_REVIEW": counts[PLAN_HUMAN_REVIEW], + "drift_objects": drift_objects, + } + + +# --------------------------------------------------------------------------- +# No-Search / No-Master-Write Guarantee (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_search_calls() -> Dict[str, Any]: + """ + Beweist statisch, dass C5C NICHT /api/search/rebuild oder Search-Admin-API aufruft. + + Prueft, dass keine Search-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) + # Verbotene Endpoints aus Fragmenten zusammensetzen, damit sie nicht als + # zusammenhaengende wörtliche Strings im Quellcode stehen (Selbstreferenz vermeiden). + banned = ["/api/" + "search/" + "rebuild", "search/" + "rebuild", "api/" + "search"] + found = [] + for node in ast.walk(tree): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + if node.value in docstrings: + continue # Docstring, kein echter Code-Ausdruck + for b in banned: + if b in node.value: + found.append(node.value) + return { + "no_search_calls": len(found) == 0, + "search_endpoints_found": sorted(set(found)), + } + + +def assert_no_master_write() -> Dict[str, Any]: + """ + Beweist statisch, dass C5C KEIN git push / Forgejo-Write ausfuehrt. + + Prueft, dass keine git-push-/Forgejo-Write-Befehle in echten Code-Ausdruecken + vorkommen und keine subprocess-Mutationsfunktionen importiert werden. + """ + 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 # Docstring, kein echter Code-Ausdruck + 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)), + } diff --git a/tolaria/c5-sync-service/test_c5c.py b/tolaria/c5-sync-service/test_c5c.py new file mode 100644 index 0000000..febbf59 --- /dev/null +++ b/tolaria/c5-sync-service/test_c5c.py @@ -0,0 +1,868 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5C Testsuite (Tolaria Propagation & Drift Verification). + +Testet gegen ein kontrolliertes lokales Git-Fixture-Repo + Fake-Tolaria-API. +Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB. +KEINE produktiven Writes. KEINE produktiven Knowledge-Dokumente. + +Abgedeckte Faelle (§19): + CREATE success / already target / path collision + CONTENT_UPDATE success / already target / unexpected drift + METADATA_UPDATE / STATE_UPDATE / TAGS_UPDATE + source/canonical ordered success / canonical blocked if source missing / dangling derived_from + rename/move safe case / rename/move ambiguity + DELETE_REQUEST -> Human Gate + SUPERSEDE safe case + legacy object -> Human Gate + secret detected -> no write + Tolaria unavailable -> retry + Read-back mismatch -> fail closed + multi-object commit success + partial commit failure + replay + idempotent replay + no Forgejo write + no Search rebuild +""" + +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 + +# C5C 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_HUMAN_REVIEW_REQUIRED, ST_RETRY_PENDING, ST_DEAD, + ST_APPLIED, IDEM_ALREADY_AT_TARGET, IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE, + RC_UNEXPECTED_TOLARIA_DRIFT, RC_UNKNOWN_OBJECT_ID, RC_SECRET_DETECTED, + RC_DANGLING_DERIVED_FROM, RC_AMBIGUOUS_DELETE, RC_TOLARIA_UNAVAILABLE, + OP_CREATE, OP_CONTENT_UPDATE, OP_METADATA_UPDATE, OP_STATE_UPDATE, + OP_TAGS_UPDATE, OP_SOURCE_CANONICAL_RELATION_UPDATE, OP_RENAME, OP_MOVE, + OP_DELETE_REQUEST, OP_SUPERSEDE, +) +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, +) + +UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b" +UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d" +UUID_C = "object/deea2ed6-cd6f-b006-2e4e-949abea60904" + + +# --------------------------------------------------------------------------- +# 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="c5c_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) +# --------------------------------------------------------------------------- + +class FakeTolaria: + """In-Memory Fake der Tolaria-Vault-API (read/write/verify).""" + + def __init__(self): + self.vault: Dict[str, str] = {} # vault_path -> content + self.write_count = 0 + self.fail_next_write = False + self.fail_next_read = False + self.unavailable = False + + def read(self, vault_path: str) -> Optional[str]: + if self.unavailable: + raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) + if self.fail_next_read: + self.fail_next_read = False + raise TolariaUnavailableError("read timeout", 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) + if self.fail_next_write: + self.fail_next_write = False + raise TolariaUnavailableError("write timeout", 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) + + +# --------------------------------------------------------------------------- +# Test-Helfer: C5A-Store + Commit mit ObjectChanges +# --------------------------------------------------------------------------- + +def _make_store() -> C5AStore: + db = os.path.join(tempfile.mkdtemp(prefix="c5c_db_"), "c5a.db") + return C5AStore(db) + + +def _seed_commit(store: C5AStore, sha: str, parent: Optional[str], + objs: List[Dict[str, Any]]) -> None: + """Legt einen Commit + ObjectChanges im Store an (Status READY).""" + store.upsert_commit({ + "commit_sha": sha, "parent_sha": parent, "sequence": 1, + "status": ST_READY, "retry_count": 0, + }) + for oc in objs: + oc["commit_sha"] = sha + store.add_object_change(oc) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestPreWriteDriftCheck(unittest.TestCase): + """§5 Pre-Write Drift Check (harte Invariante).""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + + def test_write_allowed_when_current_is_before(self): + before = _fm(UUID_A, state="current") + "body v1" + target = _fm(UUID_A, state="current") + "body v2" + self.fake.vault["/app/vault/x.md"] = before + drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", before, target) + self.assertEqual(drift, DRIFT_WRITE_ALLOWED) + self.assertEqual(cur, before) + + def test_already_at_target_no_write(self): + target = _fm(UUID_A, state="current") + "body v2" + self.fake.vault["/app/vault/x.md"] = target + drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", None, target) + self.assertEqual(drift, DRIFT_ALREADY_AT_TARGET) + + def test_unexpected_drift_fail_closed(self): + before = _fm(UUID_A, state="current") + "body v1" + target = _fm(UUID_A, state="current") + "body v2" + # Tolaria hat einen DRITTEN Zustand (weder before noch target) + self.fake.vault["/app/vault/x.md"] = _fm(UUID_A, state="current") + "body v3" + drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", before, target) + self.assertEqual(drift, DRIFT_UNEXPECTED) + + def test_create_write_allowed_when_absent(self): + target = _fm(UUID_A, state="current") + "body" + drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", None, target) + self.assertEqual(drift, DRIFT_WRITE_ALLOWED) + self.assertIsNone(cur) + + +class TestCreate(unittest.TestCase): + """§6 CREATE.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def _create_commit(self, content: str) -> str: + self.repo.write("modul-09.md", content) + return self.repo.commit("create") + + def test_create_success(self): + content = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + sha = self._create_commit(content) + _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(parse_frontmatter(content)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], content) + self.assertEqual(self.fake.write_count, 1) + + def test_create_already_target(self): + content = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + sha = self._create_commit(content) + # Tolaria hat bereits exakt den Ziel-Inhalt + self.fake.vault["/app/vault/modul-09.md"] = content + _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(parse_frontmatter(content)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.write_count, 0) # kein Write + + def test_create_path_collision(self): + # Pfad ist bereits mit ANDEREM Inhalt belegt (weder before noch target) + content = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + sha = self._create_commit(content) + self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_B) + "anderer inhalt" + _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(parse_frontmatter(content)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) # kein Write bei Drift + + +class TestContentUpdate(unittest.TestCase): + """§7 CONTENT_UPDATE.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def _content_update_commit(self, before: str, after: str) -> tuple: + self.repo.write("modul-09.md", before) + parent = self.repo.commit("v1") + self.repo.write("modul-09.md", after) + sha = self.repo.commit("v2") + return sha, parent + + def test_content_update_success(self): + before = _fm(UUID_A, state="current") + "body v1" + after = _fm(UUID_A, state="current") + "body v2" + sha, parent = self._content_update_commit(before, after) + self.fake.vault["/app/vault/modul-09.md"] = before # Tolaria = BEFORE + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_CONTENT_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "content_hash_before": content_hash(parse_frontmatter(before)[1]), + "content_hash_after": content_hash(parse_frontmatter(after)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after) + self.assertEqual(self.fake.write_count, 1) + + def test_content_update_already_target(self): + before = _fm(UUID_A, state="current") + "body v1" + after = _fm(UUID_A, state="current") + "body v2" + sha, parent = self._content_update_commit(before, after) + self.fake.vault["/app/vault/modul-09.md"] = after # bereits am Ziel + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_CONTENT_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "content_hash_before": content_hash(parse_frontmatter(before)[1]), + "content_hash_after": content_hash(parse_frontmatter(after)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.write_count, 0) + + def test_content_update_unexpected_drift(self): + before = _fm(UUID_A, state="current") + "body v1" + after = _fm(UUID_A, state="current") + "body v2" + sha, parent = self._content_update_commit(before, after) + # Tolaria hat einen DRITTEN Zustand + self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "body v3" + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_CONTENT_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "content_hash_before": content_hash(parse_frontmatter(before)[1]), + "content_hash_after": content_hash(parse_frontmatter(after)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) # kein Ueberschreiben + + +class TestMetadataStateTags(unittest.TestCase): + """§8 METADATA / STATE / TAGS.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def _update_commit(self, before: str, after: str) -> tuple: + self.repo.write("modul-09.md", before) + parent = self.repo.commit("v1") + self.repo.write("modul-09.md", after) + sha = self.repo.commit("v2") + return sha, parent + + def test_metadata_update(self): + before = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + "body" + after = _fm(UUID_A, type="arch", role="module", representation="source", state="current", summary="neu") + "body" + sha, parent = self._update_commit(before, after) + self.fake.vault["/app/vault/modul-09.md"] = before + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_METADATA_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]), + "metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after) + + def test_state_update(self): + before = _fm(UUID_A, state="current") + "body" + after = _fm(UUID_A, state="historical") + "body" + sha, parent = self._update_commit(before, after) + self.fake.vault["/app/vault/modul-09.md"] = before + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_STATE_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]), + "metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after) + + def test_tags_update(self): + before = _fm(UUID_A, tags=["a"]) + "body" + after = _fm(UUID_A, tags=["a", "b"]) + "body" + sha, parent = self._update_commit(before, after) + self.fake.vault["/app/vault/modul-09.md"] = before + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_TAGS_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]), + "metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after) + + +class TestSourceCanonical(unittest.TestCase): + """§9 Source/Canonical Relation.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_source_canonical_ordered_success(self): + # Source (Root) zuerst, dann Canonical (derived_from=Source) + source = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + "body" + canonical = _fm(UUID_B, type="arch", role="module", representation="canonical", + state="current", derived_from=UUID_A) + "body" + self.repo.write("modul-09.md", source) + self.repo.write("notes/trading/system-docs/modul-09.md", canonical) + sha = self.repo.commit("pair") + # Beide Objekte im selben Commit + _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(parse_frontmatter(source)[1])}, + {"object_id": UUID_B, "operation": OP_CREATE, "path_before": None, + "path_after": "notes/trading/system-docs/modul-09.md", + "content_hash_after": content_hash(parse_frontmatter(canonical)[1])}, + ]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], source) + self.assertEqual(self.fake.vault["/app/vault/notes/trading/system-docs/modul-09.md"], canonical) + + def test_canonical_blocked_if_source_missing(self): + # Canonical referenziert Source, die NICHT existiert -> dangling + canonical = _fm(UUID_B, type="arch", role="module", representation="canonical", + state="current", derived_from=UUID_A) + "body" + self.repo.write("notes/trading/system-docs/modul-09.md", canonical) + sha = self.repo.commit("canonical-only") + _seed_commit(self.store, sha, None, [{ + "object_id": UUID_B, "operation": OP_CREATE, "path_before": None, + "path_after": "notes/trading/system-docs/modul-09.md", + "content_hash_after": content_hash(parse_frontmatter(canonical)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) # kein Write bei dangling + + def test_dangling_derived_from(self): + # derived_from zeigt auf nicht-existierende ID + content = _fm(UUID_B, type="arch", role="module", representation="canonical", + state="current", derived_from="object/00000000-0000-0000-0000-000000000000") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("dangling") + _seed_commit(self.store, sha, None, [{ + "object_id": UUID_B, "operation": OP_CREATE, "path_before": None, + "path_after": "modul-09.md", + "content_hash_after": content_hash(parse_frontmatter(content)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) + + +class TestRenameMove(unittest.TestCase): + """§10 RENAME / MOVE -> Human Gate (sichere v1-Policy).""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_rename_safe_case_human_gate(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + parent = self.repo.commit("v1") + self.repo.write("modul-09-new.md", content) + self.repo.write("modul-09.md", "") # alt entfernen + sha = self.repo.commit("rename") + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_RENAME, + "path_before": "modul-09.md", "path_after": "modul-09-new.md", + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) # kein automatischer Rename + + def test_move_ambiguity_human_gate(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + parent = self.repo.commit("v1") + self.repo.write("sub/modul-09.md", content) + self.repo.write("modul-09.md", "") + sha = self.repo.commit("move") + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_MOVE, + "path_before": "modul-09.md", "path_after": "sub/modul-09.md", + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) + + +class TestDeleteSupersedeLegacy(unittest.TestCase): + """§11 DELETE, §12 SUPERSEDE, §13 LEGACY.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_delete_request_human_gate(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + parent = self.repo.commit("v1") + self.repo.write("modul-09.md", "") + sha = self.repo.commit("delete") + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_DELETE_REQUEST, + "path_before": "modul-09.md", "path_after": None, + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) # kein Hard Delete + + def test_supersede_safe_case(self): + before = _fm(UUID_A, state="current") + "body" + after = _fm(UUID_A, state="superseded") + "body" + self.repo.write("modul-09.md", before) + parent = self.repo.commit("v1") + self.repo.write("modul-09.md", after) + sha = self.repo.commit("supersede") + self.fake.vault["/app/vault/modul-09.md"] = before + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_SUPERSEDE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]), + "metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after) + + def test_legacy_object_human_gate(self): + # Legacy-Objekt (keine gueltige object_id) -> Human Gate + content = "---\nknowledge_schema: 1\nid: legacy\n---\nbody" + self.repo.write("README.md", content) + sha = self.repo.commit("legacy") + _seed_commit(self.store, sha, None, [{ + "object_id": None, "operation": OP_METADATA_UPDATE, + "path_before": None, "path_after": "README.md", + "reason_code": "UNKNOWN_OBJECT_ID", + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) + + +class TestSecretSafety(unittest.TestCase): + """Secret detected -> no write.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_secret_detected_no_write(self): + content = _fm(UUID_A, state="current") + "body\nkey: sk-1234567890abcdefghijklmnopqrstuvwxyz" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("secret") + _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(parse_frontmatter(content)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.fake.write_count, 0) + + +class TestRetry(unittest.TestCase): + """§17 Retry: nur technische Fehler retrybar.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_tolaria_unavailable_retry(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + _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(parse_frontmatter(content)[1]), + }]) + # Tolaria down -> retrybar + self.fake.unavailable = True + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_RETRY_PENDING) + self.assertEqual(res["results"][0]["retry_count"], 1) # _handle_failure hat increment_retry aufgerufen + + def test_drift_not_retryable(self): + # Drift ist NICHT retrybar -> direkt Human Gate + before = _fm(UUID_A, state="current") + "body v1" + after = _fm(UUID_A, state="current") + "body v2" + self.repo.write("modul-09.md", before) + parent = self.repo.commit("v1") + self.repo.write("modul-09.md", after) + sha = self.repo.commit("v2") + self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "body v3" + _seed_commit(self.store, sha, parent, [{ + "object_id": UUID_A, "operation": OP_CONTENT_UPDATE, + "path_before": "modul-09.md", "path_after": "modul-09.md", + "content_hash_before": content_hash(parse_frontmatter(before)[1]), + "content_hash_after": content_hash(parse_frontmatter(after)[1]), + }]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertEqual(self.store.increment_retry(sha), 1) # kein Retry-Increment durch C5C + + +class TestReadBackMismatch(unittest.TestCase): + """Read-back mismatch -> fail closed.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_readback_mismatch_fail_closed(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + _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(parse_frontmatter(content)[1]), + }]) + # Fake schreibt, aber Read-Back liefert anderen Inhalt (Simulation) + self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "ANDERER body" + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + + +class TestCommitAtomicity(unittest.TestCase): + """§15 Commit-Atomicity + Replay.""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.prop = C5CPropagator(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_multi_object_commit_success(self): + c1 = _fm(UUID_A, state="current") + "body A" + c2 = _fm(UUID_B, state="current") + "body B" + self.repo.write("modul-09.md", c1) + self.repo.write("modul-10.md", c2) + sha = self.repo.commit("multi") + _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(parse_frontmatter(c1)[1])}, + {"object_id": UUID_B, "operation": OP_CREATE, "path_before": None, + "path_after": "modul-10.md", + "content_hash_after": content_hash(parse_frontmatter(c2)[1])}, + ]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.write_count, 2) + + def test_partial_commit_failure(self): + # Objekt 1 ok, Objekt 2 drift -> Commit NICHT als complete markieren + c1 = _fm(UUID_A, state="current") + "body A" + c2 = _fm(UUID_B, state="current") + "body B" + self.repo.write("modul-09.md", c1) + self.repo.write("modul-10.md", c2) + sha = self.repo.commit("multi") + # Objekt 2 hat Drift in Tolaria + self.fake.vault["/app/vault/modul-10.md"] = _fm(UUID_B, state="current") + "DRIFT" + _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(parse_frontmatter(c1)[1])}, + {"object_id": UUID_B, "operation": OP_CREATE, "path_before": None, + "path_after": "modul-10.md", + "content_hash_after": content_hash(parse_frontmatter(c2)[1])}, + ]) + res = self.prop.propagate_commit(sha) + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + # Commit NICHT als Tolaria-complete (nicht UPDATING_SEARCH) + self.assertNotEqual(res["status"], ST_UPDATING_SEARCH) + + def test_idempotent_replay(self): + # Nach erfolgreichem Write: Replay erkennt ALREADY_AT_TARGET, kein Doppel-Write + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + _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(parse_frontmatter(content)[1]), + }]) + # Erster Lauf: schreibt + res1 = self.prop.propagate_commit(sha) + self.assertEqual(res1["status"], ST_UPDATING_SEARCH) + writes_after_first = self.fake.write_count + # Zweiter Lauf (Replay): Tolaria hat bereits Ziel -> kein Write + res2 = self.prop.propagate_commit(sha) + self.assertEqual(res2["status"], ST_UPDATING_SEARCH) + self.assertEqual(self.fake.write_count, writes_after_first) # kein Doppel-Write + + +class TestGuarantees(unittest.TestCase): + """No-Search / No-Master-Write Guarantee (statisch).""" + + def test_no_search_calls(self): + res = assert_no_search_calls() + self.assertTrue(res["no_search_calls"], res) + + def test_no_master_write(self): + res = assert_no_master_write() + self.assertTrue(res["no_master_write"], res) + + +class TestDryRun(unittest.TestCase): + """§20 Live-Dry-Run (read-only, kein Write).""" + + def setUp(self): + self.fake = FakeTolaria() + self.client = FakeTolariaClient(self.fake) + self.repo = FixtureRepo() + self.store = _make_store() + self.reader = GitReader(self.repo.dir) + self.dry = C5CDryRun(self.store, self.reader, self.client) + + def tearDown(self): + self.repo.cleanup() + self.store.close() + + def test_dry_run_plan(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + _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(parse_frontmatter(content)[1]), + }]) + plan = self.dry.plan_commit(sha) + self.assertEqual(plan["plan"][0]["plan"], PLAN_WOULD_WRITE) + self.assertEqual(self.fake.write_count, 0) # KEIN Write im Dry-Run + + def test_dry_run_already_at_target(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + self.fake.vault["/app/vault/modul-09.md"] = content + _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(parse_frontmatter(content)[1]), + }]) + plan = self.dry.plan_commit(sha) + self.assertEqual(plan["plan"][0]["plan"], PLAN_ALREADY_AT_TARGET) + self.assertEqual(self.fake.write_count, 0) + + def test_dry_run_drift(self): + content = _fm(UUID_A, state="current") + "body" + self.repo.write("modul-09.md", content) + sha = self.repo.commit("create") + self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "DRIFT" + _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(parse_frontmatter(content)[1]), + }]) + plan = self.dry.plan_commit(sha) + self.assertEqual(plan["plan"][0]["plan"], PLAN_DRIFT) + self.assertEqual(self.fake.write_count, 0) + + +if __name__ == "__main__": + unittest.main(verbosity=2)