feat(tolaria): add C5 Tolaria propagation and drift verification

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)
 

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

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)),
+    }

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)