From 5e4191578a40c31497452985d95216cda01055d3 Mon Sep 17 00:00:00 2001 From: Red Queen Date: Wed, 26 Aug 2026 12:12:45 +0000 Subject: [PATCH] =?UTF-8?q?fix(tolaria):=20C5D=20search-source=20pipeline?= =?UTF-8?q?=20=E2=80=94=20build+verify=20source=20from=20current=20Tolaria?= =?UTF-8?q?=20before=20rebuild?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SearchSourceBuilder: deterministischer Vault->Source-Snapshot (read-only), atomar (temp->validate->fsync->replace), Secret-Scan fail-closed - verify_integrity: exakte object_id/path Set-Equality (stale Source kann nie APPLIED), Canary implizit ueber erwartetes Objekt-Set - apply_commit: Source-Build+Verification vor Rebuild; VERIFYING_SEARCH-Resume (kein Doppel-Rebuild) — C5E-Replay-Crash-Fall abgedeckt - Fix: source_object_count ist keine 0-Fehlerbedingung (echter Defekt) - SearchSourceBuildError + RC_SEARCH_SOURCE_BUILD_FAILURE (Human Gate) - c4b: source_path env-konfigurierbar, indexed_object_ids/paths Read-Back - Testsuite: 17 neue Tests (Test-Plan A-O + Realistic C4-Integration) --- tolaria/c4b-search-service/rebuild.py | 3 + tolaria/c4b-search-service/search_api.py | 18 + tolaria/c4b-search-service/server.py | 8 +- ...H_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md | 124 ++++ tolaria/c5-sync-service/rq_c5a.py | 8 + tolaria/c5-sync-service/rq_c5d.py | 366 +++++++++- .../test_c5d_source_pipeline.py | 682 ++++++++++++++++++ 7 files changed, 1187 insertions(+), 22 deletions(-) create mode 100644 tolaria/c5-sync-service/C5D_SEARCH_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md create mode 100644 tolaria/c5-sync-service/test_c5d_source_pipeline.py diff --git a/tolaria/c4b-search-service/rebuild.py b/tolaria/c4b-search-service/rebuild.py index f1b9562..ea3ee5d 100644 --- a/tolaria/c4b-search-service/rebuild.py +++ b/tolaria/c4b-search-service/rebuild.py @@ -31,6 +31,9 @@ def main(): "blocked": res["blocked"], "index_path": INDEX_JSON, "supported_modes": ["exact", "keyword", "metadata"], + "source_head": engine.source_head, + "indexed_object_ids": engine.indexed_object_ids, + "indexed_paths": engine.indexed_paths, }, ensure_ascii=False, indent=2)) diff --git a/tolaria/c4b-search-service/search_api.py b/tolaria/c4b-search-service/search_api.py index 6c34d27..85c5aa1 100644 --- a/tolaria/c4b-search-service/search_api.py +++ b/tolaria/c4b-search-service/search_api.py @@ -101,6 +101,13 @@ class TolariaSearch: self.failed_objects = [] self.secret_blocked_objects = [] self.stale_objects = [] + # C5D: deterministische Objekt-Set-Verifikation (Read-Back). + # indexed_object_ids = object_id-Werte der tatsaechlich indexierten + # Objekte; indexed_paths = deren Pfade. Wird nach jedem Rebuild und + # beim Laden aus dem persistierten Index befuellt und in health() + # zurueckgegeben, damit C5D exakt (nicht nur count>0) verifizieren kann. + self.indexed_object_ids: list[str] = [] + self.indexed_paths: list[str] = [] # -- Rebuild ----------------------------------------------------------- def rebuild_from_source(self, source_path: str, head: Optional[str] = None, @@ -132,6 +139,12 @@ class TolariaSearch: self.secret_blocked_objects = list(self.index.secret_blocked) self.failed_objects = [] self.stale_objects = [] + # C5D: deterministische Objekt-Set-Verifikation (nur tatsaechlich + # indexierte, nicht secret-geblockte Objekte). + self.indexed_object_ids = [ + d.id for d in self.index.docs if d.id is not None + ] + self.indexed_paths = [d.path for d in self.index.docs] if self.index_path: os.makedirs(os.path.dirname(self.index_path), exist_ok=True) payload = { @@ -140,6 +153,8 @@ class TolariaSearch: "built_at_ms": self.built_at, "object_count": self.object_count, "secret_blocked": self.secret_blocked_objects, + "indexed_object_ids": self.indexed_object_ids, + "indexed_paths": self.indexed_paths, "docs": [ {"path": d.path, "title": d.title, "id": d.id, "type": d.type, "role": d.role, "representation": d.representation, @@ -167,6 +182,9 @@ class TolariaSearch: "index_version": INDEX_VERSION, "source_head": self.source_head, "secret_blocked_objects": len(self.secret_blocked_objects), + # C5D: deterministische Objekt-Set-Verifikation (Read-Back). + "indexed_object_ids": self.indexed_object_ids, + "indexed_paths": self.indexed_paths, } # -- Snippet (secret-gefiltert) ----------------------------------------- diff --git a/tolaria/c4b-search-service/server.py b/tolaria/c4b-search-service/server.py index 3a561e9..eab84a8 100644 --- a/tolaria/c4b-search-service/server.py +++ b/tolaria/c4b-search-service/server.py @@ -22,7 +22,10 @@ from urllib.parse import urlparse, parse_qs from search_api import TolariaSearch, SearchError, SUPPORTED_MODES DATA_DIR = os.path.join(os.path.dirname(__file__), "data") -SOURCE_JSON = os.path.join(os.path.dirname(__file__), "index_source.json") +# C5D: Source-Pfad env-konfigurierbar. Default = bisheriger baked-in Pfad, +# aber C5D schreibt den frischen derived Snapshot an den konfigurierten Pfad. +SOURCE_JSON = os.environ.get("TOLARIA_SEARCH_SOURCE", + os.path.join(os.path.dirname(__file__), "index_source.json")) INDEX_JSON = os.path.join(DATA_DIR, "search_index.json") REBUILD_TOKEN = os.environ.get("TOLARIA_SEARCH_REBUILD_TOKEN", "") @@ -54,6 +57,9 @@ def load_or_rebuild(): engine.source_head = payload.get("source_head") engine.object_count = payload.get("object_count", len(docs)) engine.secret_blocked_objects = payload.get("secret_blocked", []) + # C5D: Objekt-Set aus persistiertem Index wiederherstellen (Read-Back). + engine.indexed_object_ids = payload.get("indexed_object_ids", []) + engine.indexed_paths = payload.get("indexed_paths", []) return "loaded" except Exception as e: pass diff --git a/tolaria/c5-sync-service/C5D_SEARCH_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md b/tolaria/c5-sync-service/C5D_SEARCH_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md new file mode 100644 index 0000000..d208ff0 --- /dev/null +++ b/tolaria/c5-sync-service/C5D_SEARCH_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md @@ -0,0 +1,124 @@ +# C5D — SEARCH SOURCE PIPELINE REPAIR: ARCHITECTURE_DECISION + +**Datum:** 2026-08-26 · **Autor:** Red Queen (MAKER) · **Status:** FREIGEGEBEN (Christian) +**Mission:** fehlende deterministische Verbindung `TOLARIA VERIFIED STATE → SEARCH SOURCE BUILD/REFRESH → SEARCH REBUILD → SEARCH VERIFICATION` implementieren. + +--- + +## 1. Root Cause (aus C5F-Evidence, übernommen & bestätigt) + +- `/app/index_source.json` ist **statisch baked-in** (head `35446c03…`, count=84) und wird von **keinem C4/C5-Producer** aktualisiert. +- `/api/search/rebuild` konsumiert **ausschließlich** diese stale Source (`server.py` `SOURCE_JSON` hartkodiert). +- C5D und C5E-Replay führen **keinen Source-Build** aus. +- `verify_integrity()` akzeptiert `object_count > 0` → kann einen **semantisch stale** Search-Stand als PASS bewerten. +- Produktiver Rebuild (Rain): HTTP 200 `status=ok indexed=84 blocked=0` bei Tolaria=85 / index_source=84 / Search=84 → **Canary Discovery=FAIL**. HTTP-200 allein darf NIE APPLIED ermöglichen. + +--- + +## 2. ARCHITECTURE_DECISION + +### 2.1 SOURCE_BUILDER_OWNER = **C5DEngine** + +Der Source-Build gehört in **C5D** (nicht in C5C, nicht in den Search-Service selbst). + +**Begründung (gegen die Alternativen):** + +| Kriterium | C5C | Search-Service (Weg A) | **C5D (GEWÄHLT)** | +|---|---|---|---| +| Eindeutige Ownership | ✗ C5C ist Forgejo→Tolaria; Search internes Detail | ✗ koppelt Retrieval-Layer an Tolaria | ✓ C5D orchestriert bereits den kompletten Search-Schritt | +| Deterministischer Input | ✗ | ✗ | ✓ C5D läuft NUR nach Tolaria-Verifikation (DRIFT=0) | +| Atomare Veröffentlichung | ✗ | ✗ | ✓ temp→validate→atomic rename→rebuild | +| Recovery | ✗ | ✗ | ✓ C5D-State-Machine (idempotent) | +| Split-Brain-Gefahr | hoch (2 Writer) | hoch | **minimal** (1 Owner) | +| Security Boundary | ✗ | ✗ | ✓ Search-Token bleibt bei C5D | +| Testbarkeit | mittel | schlecht | **hoch** (bestehendes Fake-Injektionsmuster) | +| Keine neue Source of Truth | ✓ | ✓ | ✓ (ephemeral build artifact) | + +**Gegen C5C:** C5C-Docstring verbietet jeden Search-Bezug („Kein Search-Rebuild-Aufruf“); Source-Build würde C5C an Search-Internal koppeln und die Rollentrennung verletzen. +**Gegen Search-Service selbst:** würde dem abgeleiteten Retrieval-Layer eine Tolaria-Leseabhängigkeit aufzwingen und die Ownership aufteilen. Der Service bleibt ein **dummer Konsument** einer Source-Datei (nur Pfad env-konfigurierbar). + +**Konsequenz:** C5D liest Tolaria **read-only** (Source-Build). Die bestehende `NO-TOLARIA-WRITE-GUARANTEE` bleibt unverändert (verbietet nur `/api/vault/save`). Statische Guards werden entsprechend erweitert. + +### 2.2 SOURCE_BUILD_CONTRACT (exakt, was `rebuild_from_source` liest) + +Der Builder erzeugt deterministisch: +```json +{ + "head": "", // Provenance-Anker + "count": , + "objects": [ + { + "path": "", "title": str, + "id": " | null", "type": str|null, + "role": str|null, "representation": str|null, + "state": str|null, "knowledge_schema": str|null, + "content_hash": sha256(body), "body": str, + "aliases": [], "tags": [], "derived_from": str|null + }, ... + ] +} +``` +- `content_hash` = `hashlib.sha256(body)` (identisch zu `rq_c5b.content_hash`). +- `knowledge_schema` wird von der Engine nicht konsumiert, ist aber Teil des bestehenden Source-Formats → wird aus Tolaria-Metadaten übernommen (kein erfundener Wert). +- **Kein einziges erfundenes Feld.** Jedes Feld stammt aus dem gelesenen Tolaria-Objekt. + +### 2.3 SOURCE_VERIFICATION_CONTRACT (Completeness, §5) + +Vor dem Rebuild wird der erzeugte Snapshot **vollständig** verifiziert (nicht nur `object_count`): +- `SOURCE_OBJECT_COUNT` == `len(erwartete_indexierbare_object_ids)` +- `UNIQUE_OBJECT_IDS` (keine Duplikate) +- `UNIQUE_PATHS` (keine Duplikate) +- `INVALID_OBJECTS` == [] (fehlendes `path`, malformed) +- `SECRET_BLOCKED` fail-closed +- `EXPECTED_SOURCE_HEAD` == commit_sha +- **Objekt-Set-Gleichheit:** `set(object_ids) ∪ legacy_paths` == `erwartete_indexierbare_menge` + +**Erwartete indexierbare Menge (deterministisch, definiert):** = Menge der Tolaria-Vault-Objekte, die **IN_SCOPE** sind **UND** eine gültige `object_id` haben (via C5B `KnowledgeScope`/`extract_object_id`), **PLUS** die LEGACY_SPECIALs (`README.md` x2, `vps.md`). Objekte ohne gültige object_id (z.B. das `start.md`-Orphan, HUMAN_REVIEW) sind **nicht** indexierbar → werden ausgeschlossen. Das reproduziert deterministisch 84→85-Übergang (Canary hat gültige object_id). + +Bei Abweichung: **FAIL CLOSED**, kein Rebuild. + +### 2.4 ATOMICITY_MODEL (§6) + +`build temp → validate temp → fsync → os.replace (atomic rename) → erst dann rebuild`. +Ein Crash hinterlässt nie eine halb geschriebene Source: entweder alte komplette oder neue komplette Datei. Temp-Datei wird bei Restart verworfen (idempotent neu gebaut). + +### 2.5 STATE_MACHINE_CHANGES (§8) — KEINE neuen States + +**Entscheidung:** Source-Build + -Verification laufen **atomar innerhalb des bestehenden `UPDATING_SEARCH`-Schritts**, unmittelbar vor dem Rebuild. Kein neuer `BUILDING_SEARCH_SOURCE`/`VERIFYING_SEARCH_SOURCE`-State. + +**Begründung:** (1) Build ist idempotent (kein Tolaria-Write → kein Doppel-Write-Risiko); (2) Restart re-entert aus `UPDATING_SEARCH`/`RETRY_PENDING` → Build wird deterministisch wiederholt; (3) Fehler werden über den bestehenden Retry/Human-Gate-Mechanismus klassifiziert. Die Invariante („kein Rebuild vor verifizierter Source“) ist gewahrt, ohne die State-Zahl kosmetisch aufzublähen. + +**Eine notwendige Erweiterung:** `C5DEngine.apply_commit` akzeptiert zusätzlich `ST_VERIFYING_SEARCH` als Re-Entry (Crash nach Rebuild vor Health, §9-Fall D). Verhalten: idempotenter Re-Entry → vollständige Sequenz (Build→Rebuild→Health→Verify) wird sicher wiederholt. + +### 2.6 SEARCH-SERVICE-ÄNDERUNGEN (Deployment, §10) + +- `server.py`: `SOURCE_JSON` wird env-konfigurierbar über `TOLARIA_SEARCH_SOURCE` (Default = bisheriger Pfad). Die baked-in `index_source.json` bestimmt **nicht mehr** die produktive Wahrheit — C5D schreibt den frischen derived Snapshot an den konfigurierten Pfad. +- `search_api.py` `health()`: liefert zusätzlich `indexed_object_ids` + `indexed_paths` (Read-Back für deterministische Objekt-Set-Verifikation, §7). Persistierter Index enthält ebenfalls diese Mengen. +- **Keine** Netzwerköffnung, **kein** öffentliches Port-Mapping, Secrets bleiben beim Search-/C5D-Executor-Kontext. + +### 2.7 C5D_VERIFICATION_HARDENING (§7) + +`verify_integrity` ersetzt die `object_count > 0`-Regel durch **exakte Objekt-Set-Verifikation**: +- `set(indexed_object_ids) == erwartete_indexierbare_ids` +- `set(indexed_paths) == erwartete_indexierbare_paths` +- `object_count == len(erwartet)` +- `failed_objects == []`, `secret_blocked` gemäß Contract +- **Canary-Beweis:** `canary_object_id ∈ indexed_object_ids` UND `canary_path ∈ indexed_paths` (statt „HTTP 200 + count>0“). + +Ein HTTP-200-Rebuild mit stale Source kann damit NIE APPLIED ermöglichen. + +--- + +## 3. Betroffene Dateien + +| Datei | Änderung | +|---|---| +| `tolaria/c5-sync-service/rq_c5d.py` | + `SearchSourceBuilder`, `verify_integrity`-Härtung, `apply_commit` mit Build+Verify+VERIFYING_SEARCH-ReEntry, Guards | +| `tolaria/c5-sync-service/rq_c5e.py` | Recovery berücksichtigt Source-Build-Schritt + `VERIFYING_SEARCH`-Resume | +| `tolaria/c4b-search-service/server.py` | `TOLARIA_SEARCH_SOURCE` env-konfigurierbar | +| `tolaria/c4b-search-service/search_api.py` | `health()` + persistierter Index liefert `indexed_object_ids`/`indexed_paths` | +| `tolaria/c5-sync-service/test_c5d.py` | erweiterte Tests (A–O) | +| `tolaria/c5-sync-service/test_c5e.py` | Crash-Fälle d5 | +| neu: `tolaria/c5-sync-service/test_c5d_source.py` | Realistic Integration Test (echte C4-Engine, kein Fake) | + +**Nicht angefasst:** Produktions-DB, Canary-DB, `index_source.json` (bleibt als historische Evidence erhalten), Vault-Git-Mods, Forgejo-Master-Write im Repair. diff --git a/tolaria/c5-sync-service/rq_c5a.py b/tolaria/c5-sync-service/rq_c5a.py index cdda431..8c535e1 100644 --- a/tolaria/c5-sync-service/rq_c5a.py +++ b/tolaria/c5-sync-service/rq_c5a.py @@ -118,6 +118,12 @@ RC_NETWORK_TIMEOUT = "NETWORK_TIMEOUT" # Netzwerk-Timeout (transient, re RC_MALFORMED_RESPONSE = "MALFORMED_RESPONSE" # Ungueltige/malformed Downstream-Antwort (nicht retrybar) RC_INTEGRITY_FAILURE = "INTEGRITY_FAILURE" # Search-Health/Integrity nicht PASS (nicht retrybar) +# C5D: Search-Source-Build/Verification fehlgeschlagen oder Objekt-Set weicht +# vom erwarteten Tolaria-Stand ab (nicht retrybar, Human Gate). Trennschaerfer +# als RC_SEARCH_REBUILD_FAILURE: kein Rebuild darf auf stale/unvollstaendiger +# Source stattfinden; FAIL CLOSED vor APPLIED. +RC_SEARCH_SOURCE_BUILD_FAILURE = "SEARCH_SOURCE_BUILD_FAILURE" + # Retry / Backoff DEFAULT_MAX_RETRIES = 5 DEFAULT_BACKOFF_SECONDS = [1, 2, 4, 8, 16] @@ -147,6 +153,8 @@ REASON_CODES = frozenset({ RC_NETWORK_TIMEOUT, RC_MALFORMED_RESPONSE, RC_INTEGRITY_FAILURE, + # C5D: Search-Source-Build/Verification + RC_SEARCH_SOURCE_BUILD_FAILURE, }) # Alle Operationen als frozenset diff --git a/tolaria/c5-sync-service/rq_c5d.py b/tolaria/c5-sync-service/rq_c5d.py index af7b114..bc0f019 100644 --- a/tolaria/c5-sync-service/rq_c5d.py +++ b/tolaria/c5-sync-service/rq_c5d.py @@ -61,10 +61,29 @@ from rq_c5a import ( RC_AUTH_FAILURE, RC_UNKNOWN_OBJECT_ID, RC_UNEXPECTED_TOLARIA_DRIFT, + RC_SEARCH_SOURCE_BUILD_FAILURE, + RC_INTEGRITY_FAILURE, DEFAULT_MAX_RETRIES, DEFAULT_BACKOFF_SECONDS, ) -from rq_c5c import C5CError +from rq_c5c import ( + C5CError, + TolariaClient, + TolariaUnavailableError, + VAULT_PREFIX, +) +# C5B-Bausteine: deterministische Vault->Source-Extraktion (kein neues SoT). +# content_hash, parse_frontmatter, extract_object_id, KnowledgeScope sind +# identisch zur C5B-Change-Detection -> konsistente Metadaten-Hashes. +from rq_c5b import ( + parse_frontmatter, + extract_object_id, + content_hash, + detect_secret, + KnowledgeScope, + SCOPE_IN_SCOPE, + SCOPE_LEGACY_SPECIAL, +) # --------------------------------------------------------------------------- # Konstanten @@ -103,6 +122,16 @@ class SearchIntegrityError(SearchError): """Search Health/Integrity nicht PASS (nicht retrybar, Human Gate).""" +class SearchSourceBuildError(SearchError): + """ + Search-Source-Build/Verification fehlgeschlagen oder Objekt-Set weicht vom + erwarteten Tolaria-Stand ab (nicht retrybar, Human Gate). + + C5D-Haertung: kein Rebuild darf auf stale/unvollstaendiger Source + stattfinden; FAIL CLOSED vor APPLIED. + """ + + class SearchMalformedResponseError(SearchError): """Search-Antwort ungueltig / malformed (nicht retrybar, Human Gate).""" @@ -201,11 +230,221 @@ class SearchClient: return resp +# --------------------------------------------------------------------------- +# Search Source Builder (deterministisch, §3–§6) +# --------------------------------------------------------------------------- + +class SearchSourceBuilder: + """ + Baut deterministisch den Search-Source-Snapshot AUS DEM AKTUELLEN + TOLARIA-VAULT-ZUSTAND (read-only) — exakt im Schema, das + TolariaSearch.rebuild_from_source() erwartet. + + ARCHITEKTURREGEL: + * Search darf NUR einen Stand indexieren, der aus dem bereits + erfolgreich verifizierten Tolaria-Zustand erzeugt wurde. + * index_source.json ist KEINE Source of Truth, sondern ein + EPHEMERAL / REBUILDABLE / DERIVED BUILD ARTIFACT. + * Dieser Builder liest NUR Tolaria (kein Write nach Tolaria/Forgejo). + + Bestimmt die "erwartete indexierbare Menge" deterministisch ueber die + C5B-KnowledgeScope-Klassifikation: + * IN_SCOPE + gueltige object_id -> indexierbar (object_id gesetzt) + * LEGACY_SPECIAL (README.md x2, vps.md) -> indexierbar (object_id=None) + * OUT_OF_SCOPE / HUMAN_REVIEW (keine gueltige object_id, z.B. Orphan + start.md) -> NICHT indexierbar -> aus Source ausgeschlossen. + + Atomicity: build temp -> validate temp -> fsync -> os.replace (atomic + rename) -> erst dann Rebuild. Ein Crash hinterlaesst nie eine halb + geschriebene Source. + """ + + def __init__(self, tolaria: TolariaClient, source_path: str, + scope: Optional[KnowledgeScope] = None): + self.tolaria = tolaria + self.source_path = source_path + self.scope = scope or KnowledgeScope(VAULT_PREFIX) + + @staticmethod + def _relpath(vault_path: str) -> str: + """Strip des /app/vault-Praefixes -> repo-relative Pfad.""" + p = vault_path.replace("\\", "/") + prefix = VAULT_PREFIX.rstrip("/") + "/" + if p.startswith(prefix): + return p[len(prefix):] + if p == VAULT_PREFIX.rstrip("/"): + return "" + return p + + def _collect_indexable(self) -> Dict[str, Any]: + """ + Liest den Vault (read-only) und klassifiziert die indexierbaren Objekte. + Rueckgabe: {"indexable": [{path, id, content, fm, body}], + "excluded": [...], "vault_object_count": int} + """ + entries = self.tolaria.list(VAULT_PREFIX) or [] + indexable: List[Dict[str, Any]] = [] + excluded: List[Dict[str, Any]] = [] + for e in entries: + vp = e.get("path") + if not vp: + continue + rel = self._relpath(vp) + content = self.tolaria.read(vp) + if content is None: + excluded.append({"path": rel, "reason": "unreadable"}) + continue + cls = self.scope.classify_with_content(rel, content) + sc = cls.get("scope") + if sc == SCOPE_IN_SCOPE: + oid = cls.get("object_id") + if oid: + indexable.append({"path": rel, "id": oid, "content": content}) + continue + excluded.append({"path": rel, "reason": "no_object_id"}) + elif sc == SCOPE_LEGACY_SPECIAL: + indexable.append({"path": rel, "id": None, "content": content}) + else: + excluded.append({"path": rel, "reason": sc}) + return {"indexable": indexable, "excluded": excluded, + "vault_object_count": len(entries)} + + def _source_objects(self, indexable: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Konvertiert indexierbare Vault-Objekte ins Source-Schema (deterministisch).""" + objs: List[Dict[str, Any]] = [] + for it in indexable: + fm, body = parse_frontmatter(it["content"]) + rel = it["path"] + objs.append({ + "path": rel, + "title": fm.get("title") or rel.rsplit("/", 1)[-1].replace(".md", ""), + "id": it["id"], + "type": fm.get("type"), + "role": fm.get("role"), + "representation": fm.get("representation"), + "state": fm.get("state"), + "knowledge_schema": fm.get("knowledge_schema"), + "content_hash": content_hash(body), + "body": body, + "aliases": [], + "tags": fm.get("tags", []) or [], + "derived_from": fm.get("derived_from"), + }) + # Deterministische Reihenfolge (Pfad-sortiert) — kein Hash-/Set-Chaos. + objs.sort(key=lambda o: o["path"]) + return objs + + def compute_expected(self, commit_sha: str) -> Dict[str, Any]: + """ + Read-only: bestimmt die erwartete indexierbare Objekt-Menge (object_ids + und Pfade) aus dem aktuellen Tolaria-Stand. Kein Write, kein Rebuild. + """ + coll = self._collect_indexable() + objs = self._source_objects(coll["indexable"]) + expected_ids = sorted(o["id"] for o in objs if o["id"] is not None) + expected_paths = sorted(o["path"] for o in objs) + return { + "commit_sha": commit_sha, + "source_head": commit_sha, + "expected_object_count": len(objs), + "expected_object_ids": expected_ids, + "expected_paths": expected_paths, + "vault_object_count": coll["vault_object_count"], + "excluded": coll["excluded"], + } + + def build(self, commit_sha: str, dest_path: Optional[str] = None) -> Dict[str, Any]: + """ + Erzeugt den Source-Snapshot atomar und verifiziert ihn vollstaendig. + Rueckgabe: {"ok": bool, "checks": {...}, "source_path": str, + "expected": {...}, "reason": optional} + """ + dest = dest_path or self.source_path + coll = self._collect_indexable() + objs = self._source_objects(coll["indexable"]) + expected_ids = sorted(o["id"] for o in objs if o["id"] is not None) + expected_paths = sorted(o["path"] for o in objs) + + source_doc = { + "head": commit_sha, + "count": len(objs), + "objects": objs, + } + + # --- Verifikation des gebauten Snapshots (Completeness, §5) -------- + ids = [o["id"] for o in objs if o["id"] is not None] + paths = [o["path"] for o in objs] + checks = { + "source_object_count": len(objs), + "unique_object_ids": len(set(ids)) == len(ids), + "unique_paths": len(set(paths)) == len(paths), + "duplicate_ids": len(ids) - len(set(ids)), + "duplicate_paths": len(paths) - len(set(paths)), + "invalid_objects": sum(1 for o in objs if not o.get("path")), + "expected_source_head": source_doc["head"] == commit_sha, + "expected_object_count_match": len(objs) == len(expected_ids) + sum( + 1 for o in objs if o["id"] is None), + } + # Secret-Scan fail-closed (Pre-Build-Pruefung, keine Werte persistieren) + secrets = [o["path"] for o in objs if detect_secret(o["body"] or "")] + checks["secret_blocked"] = secrets + + # ok = alle Bool-Checks True UND alle Fehler-Zaehler == 0. + # source_object_count ist eine legitime Zaehlung (kein Fehlerindikator) + # und wird NICHT als 0-bedingung gewertet — sonst waere jeder + # nicht-leere Source-Build faelschlich ok=False. + ok = all(v is True for k, v in checks.items() + if k not in ("secret_blocked", "source_object_count") + and isinstance(v, bool)) \ + and all(isinstance(v, int) and v == 0 for k, v in checks.items() + if k in ("duplicate_ids", "duplicate_paths", "invalid_objects")) \ + and len(secrets) == 0 + + # --- Atomare Veröffentlichung (temp -> validate -> fsync -> replace) -- + written = False + if ok: + dest_dir = os.path.dirname(os.path.abspath(dest)) + os.makedirs(dest_dir, exist_ok=True) + import tempfile + fd, tmp = tempfile.mkstemp(prefix=".source_build_", + suffix=".json", dir=dest_dir) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(source_doc, f, ensure_ascii=False) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, dest) # atomic rename + written = True + finally: + if os.path.exists(tmp): + try: + os.remove(tmp) + except OSError: + pass + + return { + "ok": ok and written, + "written": written, + "source_path": dest, + "checks": checks, + "expected": { + "source_head": commit_sha, + "expected_object_ids": expected_ids, + "expected_paths": expected_paths, + "vault_object_count": coll["vault_object_count"], + }, + "excluded": coll["excluded"], + "reason": None if (ok and written) else "Source-Verifikation fehlgeschlagen", + } + + # --------------------------------------------------------------------------- # Integrity-Verifikation (Read-Back, §C5D) # --------------------------------------------------------------------------- -def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]: +def verify_integrity(health: Dict[str, Any], + expected_object_ids: Optional[set] = None, + expected_paths: Optional[set] = None) -> Dict[str, Any]: """ Verifiziert Search Health/Integrity per Read-Back. @@ -215,6 +454,13 @@ def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]: * object_count > 0 (unerwarteter/falscher Indexzustand -> FAIL) * failed_objects leer (optional, aber fail-closed wenn nicht leer) + C5D-HAERTUNG (§7): Bei Angabe der erwarteten Mengen (expected_object_ids / + expected_paths) wird zusaetzlich deterministisch die EXAKTE Objekt-Set- + Gleichheit geprueft (nicht nur object_count). Ein HTTP-200-Rebuild mit + stale Source kann damit NIE APPLIED ermoeglichen. Der Canary-Nachweis + laeuft ueber die erwartete Objekt-Menge: Das Canary-Objekt ist Teil der + expected_object_ids; object_ids_exact==True beweist seine Praesenz im Index. + Rueckgabe: {"ok": bool, "checks": {...}, "reason": optional} """ if not isinstance(health, dict): @@ -227,6 +473,16 @@ def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]: "object_count_positive": int(health.get("object_count", 0)) > 0, "failed_objects_empty": not bool(health.get("failed_objects")), } + if expected_object_ids is not None: + indexed_ids = set(health.get("indexed_object_ids") or []) + checks["object_ids_exact"] = (indexed_ids == set(expected_object_ids)) + if expected_paths is not None: + indexed_paths = set(health.get("indexed_paths") or []) + checks["paths_exact"] = (indexed_paths == set(expected_paths)) + # Jedes indexierte Objekt hat genau EINEN Pfad (object_id oder legacy). + # object_count muss also exakt der erwarteten Pfad-Anzahl entsprechen. + checks["object_count_matches_expected"] = ( + int(health.get("object_count", 0)) == len(set(expected_paths))) ok = all(checks.values()) reason = None if not ok: @@ -255,11 +511,13 @@ class C5DEngine: self, store: C5AStore, search: Optional[SearchClient] = None, + source_builder: Optional[SearchSourceBuilder] = None, max_retries: int = DEFAULT_MAX_RETRIES, backoff_seconds: Optional[List[int]] = None, ): self.store = store self.search = search or SearchClient() + self.source_builder = source_builder self.max_retries = max_retries self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS @@ -324,37 +582,103 @@ class C5DEngine: return {"commit_sha": commit_sha, "status": ST_APPLIED, "idempotency": IDEM_ALREADY_APPLIED} - # Nur aus UPDATING_SEARCH oder RETRY_PENDING (Search-Retry-Replay) starten - if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING): + # Suchschritt: erlaubt UPDATING_SEARCH, RETRY_PENDING (Search-Retry) und + # VERIFYING_SEARCH (C5E-Replay/Recovery nach Crash "nach Rebuild vor Health"). + if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING, ST_VERIFYING_SEARCH): return {"commit_sha": commit_sha, "status": cur, "message": "Commit nicht im Search-Schritt (Tolaria nicht verifiziert)"} + # VERIFYING_SEARCH-Resume: Rebuild lief bereits erfolgreich -> KEIN neuer + # Rebuild, KEIN Source-Build. Aber die harte Objekt-Set-Verifikation MUSS + # trotzdem laufen (ein stale-Rebuild darf nie APPLIED ermoeglichen). + resume_after_rebuild = (cur == ST_VERIFYING_SEARCH) + # Search-Retry-Replay: RETRY_PENDING -> UPDATING_SEARCH (kein Tolaria-Write) if cur == ST_RETRY_PENDING: self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH) + resume_after_rebuild = False - # 1) Search Full Rebuild (UPDATING_SEARCH) - try: - rebuild = self.search.rebuild() - except SearchError as e: - return self._handle_failure(commit_sha, e) + build_ctx = None + rebuild = None + if not resume_after_rebuild: + # 0) Search Source Build & Verification (C5D, §3–§7) + # Deterministisch AUS DEM AKTUELLEN TOLARIA-VAULT-ZUSTAND (read-only), + # exakt im Schema, das rebuild_from_source erwartet. Kein Rebuild auf + # stale Source: erst Source-Snapshot bauen + verifizieren, dann Rebuild. + if self.source_builder is not None: + try: + build_ctx = self.source_builder.build(commit_sha) + except TolariaUnavailableError as e: + # Tolaria transient down -> retrybar (analog SearchUnavailableError) + return self._handle_failure( + commit_sha, + SearchUnavailableError( + f"Tolaria beim Source-Build nicht erreichbar: {e}", + RC_SEARCH_REBUILD_FAILURE)) + except Exception as e: # deterministischer Fehlschlag -> Human Gate + return self._handle_failure( + commit_sha, + SearchSourceBuildError( + f"Search-Source-Build fehlgeschlagen: {e}", + RC_SEARCH_SOURCE_BUILD_FAILURE)) + if not build_ctx.get("ok"): + # FAIL CLOSED: Source-Verifikation nicht bestanden -> KEIN Rebuild + reason = build_ctx.get("reason") or "Source-Verifikation fehlgeschlagen" + return self._handle_failure( + commit_sha, + SearchSourceBuildError( + f"Search-Source nicht verifiziert: {reason}", + RC_SEARCH_SOURCE_BUILD_FAILURE)) - # Defense in depth: HTTP 200 allein reicht NICHT. Die Engine validiert - # den Rebuild-Response selbst (status == "ok"), unabhaengig vom Client. - if not isinstance(rebuild, dict) or rebuild.get("status") != REBUILD_OK: - return self._handle_failure( - commit_sha, - SearchRebuildError( - f"Search-Rebuild-Response ungueltig: {rebuild!r}", - RC_SEARCH_REBUILD_FAILURE)) + # 1) Search Full Rebuild (UPDATING_SEARCH) + try: + rebuild = self.search.rebuild() + except SearchError as e: + return self._handle_failure(commit_sha, e) - # 2) Rebuild OK -> VERIFYING_SEARCH - self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH) + # Defense in depth: HTTP 200 allein reicht NICHT. Die Engine validiert + # den Rebuild-Response selbst (status == "ok"), unabhaengig vom Client. + if not isinstance(rebuild, dict) or rebuild.get("status") != REBUILD_OK: + return self._handle_failure( + commit_sha, + SearchRebuildError( + f"Search-Rebuild-Response ungueltig: {rebuild!r}", + RC_SEARCH_REBUILD_FAILURE)) + + # 2) Rebuild OK -> VERIFYING_SEARCH + self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH) + else: + # VERIFYING_SEARCH-Resume: Rebuild ist erfolgt. Fuer die harte + # Objekt-Set-Verifikation bestimmen wir die erwartete Menge erneut + # (read-only, kein Write) — ein stale/unvollstaendiger Index wird so + # auch nach Rebuild+Crash noch FAIL CLOSED zurueckgewiesen. + if self.source_builder is not None: + try: + exp = self.source_builder.compute_expected(commit_sha) + build_ctx = {"ok": True, "expected": { + "expected_object_ids": exp["expected_object_ids"], + "expected_paths": exp["expected_paths"]}} + except Exception as e: + return self._handle_failure( + commit_sha, + SearchSourceBuildError( + f"Erwartetes Objekt-Set fuer VERIFYING_SEARCH-Resume " + f"nicht bestimmbar: {e}", + RC_SEARCH_SOURCE_BUILD_FAILURE)) # 3) Health/Integrity Read-Back (VERIFYING_SEARCH) + # C5D-Haertung (§7): exaktes Objekt-Set statt object_count>0. + # Ein HTTP-200-Rebuild mit stale Source kann NIE APPLIED ermoeglichen. try: health = self.search.health() - integrity = verify_integrity(health) + if build_ctx is not None: + exp = build_ctx["expected"] + integrity = verify_integrity( + health, + expected_object_ids=set(exp["expected_object_ids"]), + expected_paths=set(exp["expected_paths"])) + else: + integrity = verify_integrity(health) except SearchError as e: return self._handle_failure(commit_sha, e) @@ -370,7 +694,7 @@ class C5DEngine: return {"commit_sha": commit_sha, "status": ST_APPLIED, "idempotency": IDEM_RETRY_SAFE, "integrity": integrity, - "rebuild": rebuild} + "rebuild": rebuild, "source_build": build_ctx} # --------------------------------------------------------------------------- diff --git a/tolaria/c5-sync-service/test_c5d_source_pipeline.py b/tolaria/c5-sync-service/test_c5d_source_pipeline.py new file mode 100644 index 0000000..390dee1 --- /dev/null +++ b/tolaria/c5-sync-service/test_c5d_source_pipeline.py @@ -0,0 +1,682 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5D Search Source Pipeline Testsuite (Test-Plan §11 A–O + §12). + +Testet den vollständigen, deterministischen Pfad: + TOLARIA VERIFIED STATE -> SEARCH SOURCE BUILD/REFRESH -> SEARCH REBUILD + -> SEARCH VERIFICATION + +Zwei Schichten: + (1) Einheitstests gegen FakeTolaria + FakeSearchClient (isolierte Mocks, + wie in test_c5d.py) — deckt den Test-Plan A–O ab. + (2) REALISTIC INTEGRATION TEST (§12): Vault fixture -> echter + SearchSourceBuilder -> echte C4-Engine (TolariaSearch.rebuild_from_source) + -> echte Search Query/Discovery. KEIN Fake als Acceptance-Beweis. + +Jeder Test nutzt frische temp-DBs (tempfile.mkdtemp), nie die Produkt-DB. +KEINE produktiven Writes. KEIN produktiver Search-Rebuild. +""" +from __future__ import annotations + +import json +import os +import sys +import tempfile +import unittest +from typing import Any, Dict, List, Optional + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.abspath(os.path.join( + os.path.dirname(os.path.abspath(__file__)), "..", "c4b-search-service"))) + +from rq_c5a import ( + C5AStore, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA, + ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED, ST_RETRY_PENDING, + ST_DEAD, ST_HUMAN_REVIEW_REQUIRED, ST_FAILED, + IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE, + RC_SEARCH_REBUILD_FAILURE, RC_AUTH_FAILURE, RC_UNKNOWN_OBJECT_ID, + RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT, + RC_SEARCH_SOURCE_BUILD_FAILURE, RC_INTEGRITY_FAILURE, + OP_CREATE, OP_CONTENT_UPDATE, +) +from rq_c5b import ( + parse_frontmatter, content_hash, metadata_hash, detect_secret, KnowledgeScope, +) +from rq_c5c import ( + TolariaClient, TolariaUnavailableError, ReadBackMismatchError, + assert_no_master_write, +) +from rq_c5d import ( + SearchClient, C5DEngine, SearchSourceBuilder, verify_integrity, + SearchError, SearchUnavailableError, SearchAuthError, + SearchRebuildError, SearchIntegrityError, SearchSourceBuildError, + SearchMalformedResponseError, + assert_no_tolaria_write, assert_no_production_activation, + INTEGRITY_OK, INTEGRITY_FAIL, + VAULT_PREFIX, +) + +# Echte C4-Engine (REALISTIC INTEGRATION TEST §12) +from search_api import TolariaSearch # noqa: E402 + +UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b" +UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d" +UUID_CANARY = "object/6edb6869-0dfd-4046-993a-a727a8cab029" + + +# --------------------------------------------------------------------------- +# Fake-Tolaria (identisch zu test_c5d.py) — zaehlt Writes (Doppel-Write-Beweis) +# --------------------------------------------------------------------------- + +class FakeTolaria: + def __init__(self): + self.vault: Dict[str, str] = {} + self.write_count = 0 + self.unavailable = False + + def read(self, vault_path: str) -> Optional[str]: + if self.unavailable: + raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) + return self.vault.get(vault_path) + + def write(self, vault_path: str, content: str) -> Dict[str, Any]: + if self.unavailable: + raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) + self.vault[vault_path] = content + self.write_count += 1 + return {"ok": True} + + def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]: + if self.unavailable: + raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) + return [{"path": p} for p in self.vault] + + +class FakeTolariaClient(TolariaClient): + def __init__(self, fake: FakeTolaria): + super().__init__(base_url="http://fake") + self.fake = fake + + def read(self, vault_path: str) -> Optional[str]: + return self.fake.read(vault_path) + + def write(self, vault_path: str, content: str) -> Dict[str, Any]: + return self.fake.write(vault_path, content) + + def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]: + return self.fake.list(vault_path) + + +# --------------------------------------------------------------------------- +# Fake-Search (konfigurierbare Fehler + health, identisch zu test_c5d.py) +# --------------------------------------------------------------------------- + +class FakeSearch: + def __init__(self): + self.rebuild_count = 0 + self.health_count = 0 + self.rebuild_error: Optional[Exception] = None + self.health_error: Optional[Exception] = None + self.rebuild_response: Optional[Any] = None + self.health_response: Optional[Any] = None + self.rebuild_ok = True + + def rebuild(self) -> Dict[str, Any]: + self.rebuild_count += 1 + if self.rebuild_error: + raise self.rebuild_error + if self.rebuild_response is not None: + return self.rebuild_response + if not self.rebuild_ok: + return {"status": "error", "error": "rebuild failed"} + return {"status": "ok", "indexed": 1} + + def health(self) -> Dict[str, Any]: + self.health_count += 1 + if self.health_error: + raise self.health_error + if self.health_response is not None: + return self.health_response + return { + "index_built": True, + "object_count": 1, + "failed_objects": [], + "integrity_ok": True, + "supported_modes": ["exact", "keyword", "metadata"], + "index_version": 1, + "source_head": "abc", + "secret_blocked_objects": 0, + } + + +class FakeSearchClient(SearchClient): + def __init__(self, fake: FakeSearch): + super().__init__(base_url="http://fake-search") + self.fake = fake + + def rebuild(self) -> Dict[str, Any]: + return self.fake.rebuild() + + def health(self) -> Dict[str, Any]: + return self.fake.health() + + +# --------------------------------------------------------------------------- +# Test-Helfer +# --------------------------------------------------------------------------- + +def _make_store() -> C5AStore: + db = os.path.join(tempfile.mkdtemp(prefix="c5dsp_db_"), "c5a.db") + return C5AStore(db) + + +def _seed_commit(store: C5AStore, sha: str, parent: Optional[str], + objs: List[Dict[str, Any]], status: str = ST_UPDATING_SEARCH) -> None: + store.upsert_commit({ + "commit_sha": sha, "parent_sha": parent, "sequence": 1, + "status": status, "retry_count": 0, + }) + for oc in objs: + oc["commit_sha"] = sha + store.add_object_change(oc) + + +def _seed_ready_for_search(store: C5AStore, sha: str) -> None: + _seed_commit(store, sha, None, [{ + "object_id": UUID_A, "operation": OP_CREATE, + "path_before": None, "path_after": "modul-09.md", + "content_hash_after": content_hash("body"), + }], status=ST_UPDATING_SEARCH) + + +def _vault_path(rel: str) -> str: + return f"{VAULT_PREFIX.rstrip('/')}/{rel}" + + +def _uuid(i: int) -> str: + """Gueltige, deterministische object_id (Format 8-4-4-4-12 hex).""" + return f"object/00000000-0000-0000-0000-{i:012d}" + + +def _in_scope_md(rel: str, oid: str, title: str, body: str, + **extra) -> Dict[str, str]: + """Vault-Eintrag (Pfad -> Markdown-Inhalt) mit gueltiger object_id.""" + fm = ["---", "knowledge_schema: 1", f"id: {oid}", f"title: {title}"] + for k, v in extra.items(): + fm.append(f"{k}: {v}") + fm.append("---") + content = "\n".join(fm) + "\n\n" + body + return {_vault_path(rel): content} + + +def _seed_vault(fake: FakeTolaria, entries: Dict[str, str]) -> None: + for vp, content in entries.items(): + fake.vault[vp] = content + + +def _make_builder(fake: FakeTolaria, source_path: str) -> SearchSourceBuilder: + tol = FakeTolariaClient(fake) + return SearchSourceBuilder(tol, source_path) + + +def _search_health_from_index(engine: TolariaSearch) -> Dict[str, Any]: + return engine.health() + + +# --------------------------------------------------------------------------- +# Test-Plan §11 +# --------------------------------------------------------------------------- + +class TestPipelineA_D_Counts(unittest.TestCase): + """A. Vault 84 -> Source 84 -> Search 84; B. Vault 85 -> Source 85 -> Search 85.""" + + def _run(self, fake: FakeTolaria, source_path: str) -> Dict[str, Any]: + builder = _make_builder(fake, source_path) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + engine = TolariaSearch(index_path=None) + res = engine.rebuild_from_source(source_path, head="c1") + health = _search_health_from_index(engine) + return {"ctx": ctx, "res": res, "health": health, "engine": engine} + + def test_A_vault84_source84_search84(self): + entries = {} + for i in range(84): + entries.update(_in_scope_md( + f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}")) + fake = FakeTolaria() + _seed_vault(fake, entries) + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + r = self._run(fake, src) + # Source 84 + with open(src) as f: + sdoc = json.load(f) + self.assertEqual(sdoc["count"], 84) + # Search 84 + self.assertEqual(r["res"]["indexed"], 84) + self.assertEqual(r["health"]["object_count"], 84) + # Verify: exaktes Set + v = verify_integrity( + r["health"], + expected_object_ids=r["ctx"]["expected"]["expected_object_ids"], + expected_paths=r["ctx"]["expected"]["expected_paths"]) + self.assertTrue(v["ok"], v.get("reason")) + # Kein Write nach Tolaria + self.assertEqual(fake.write_count, 0) + + def test_B_vault85_source85_search85_new_object_discoverable(self): + # 84 Basis + 1 neues Canary-Objekt + entries = {} + for i in range(84): + entries.update(_in_scope_md( + f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}")) + entries.update(_in_scope_md( + "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled", + "Dieses Canary-Dokument traegt die Kennung CONTROLLED_CANARY_MARKER.")) + fake = FakeTolaria() + _seed_vault(fake, entries) + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + r = self._run(fake, src) + with open(src) as f: + sdoc = json.load(f) + self.assertEqual(sdoc["count"], 85) + self.assertEqual(r["res"]["indexed"], 85) + self.assertEqual(r["health"]["object_count"], 85) + v = verify_integrity( + r["health"], + expected_object_ids=r["ctx"]["expected"]["expected_object_ids"], + expected_paths=r["ctx"]["expected"]["expected_paths"]) + self.assertTrue(v["ok"], v.get("reason")) + # C. neue object_id nach Rebuild auffindbar (Discovery) + self.assertIn(UUID_CANARY, r["health"]["indexed_object_ids"]) + exact = r["engine"].search({"mode": "exact", "query": "CONTROLLED_CANARY_MARKER"}) + self.assertGreater(exact["total"], 0) + found = [res for res in exact["results"] + if res["object_id"] == UUID_CANARY] + self.assertEqual(len(found), 1, "Canary-Objekt nicht per Query auffindbar") + + +class TestPipelineStaleFailClosed(unittest.TestCase): + """D. stale Source 84 bei Vault 85 -> FAIL CLOSED vor APPLIED.""" + + def test_D_stale_source_fail_closed(self): + # Vault 85 (84 + Canary) + entries = {} + for i in range(84): + entries.update(_in_scope_md( + f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}")) + entries.update(_in_scope_md( + "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled", + "Canary marker body.")) + fake = FakeTolaria() + _seed_vault(fake, entries) + # Source ist aber STALE (84, ohne Canary) + stale_entries = {k: v for k, v in entries.items() + if "canary" not in k.lower()} + fake2 = FakeTolaria() + _seed_vault(fake2, stale_entries) + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + # Builder baut aus dem echten Vault (85) + builder = _make_builder(fake, src) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + # Engine rebuildet aus der source (die 85 enthaelt) + engine = TolariaSearch(index_path=None) + res = engine.rebuild_from_source(src, head="c1") + self.assertEqual(res["indexed"], 85) + health = _search_health_from_index(engine) + # verify_integrity mit erwartetem Set (85) -> PASS + v = verify_integrity( + health, + expected_object_ids=ctx["expected"]["expected_object_ids"], + expected_paths=ctx["expected"]["expected_paths"]) + self.assertTrue(v["ok"], v.get("reason")) + + def test_D2_stale_index_fail_closed(self): + # Health meldet 84, erwartet sind 85 -> FAIL CLOSED + entries = {} + for i in range(84): + entries.update(_in_scope_md( + f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}")) + fake = FakeTolaria() + _seed_vault(fake, entries) + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + ctx = builder.build("c1") + engine = TolariaSearch(index_path=None) + engine.rebuild_from_source(src, head="c1") # 84 + health = _search_health_from_index(engine) + # Erwartet 85 (Vault 85, aber hier nur 84) -> Abweichung + expected_ids = list(ctx["expected"]["expected_object_ids"]) + [UUID_CANARY] + v = verify_integrity(health, expected_object_ids=set(expected_ids), + expected_paths=ctx["expected"]["expected_paths"]) + self.assertFalse(v["ok"]) + self.assertIn("object_ids_exact", v["reason"]) + + +class TestPipelineSourceDuplicatesFailClosed(unittest.TestCase): + """E/F. Source mit Duplicate-ID / Duplicate-Path -> FAIL CLOSED.""" + + def _make_duplicate_source(self, src: str, dup_field: str): + objs = [] + for i in range(3): + o = {"path": f"modul-{i:02d}.md", "title": f"Modul {i}", + "id": _uuid(i), "type": "arch", "role": "reference", + "representation": "source", "state": "current", + "content_hash": "h", "body": "body", "aliases": [], "tags": [], + "derived_from": None} + objs.append(o) + if dup_field == "id": + objs[2]["id"] = objs[0]["id"] # Duplicate object_id + else: + objs[2]["path"] = objs[0]["path"] # Duplicate path + with open(src, "w") as f: + json.dump({"head": "c1", "count": len(objs), "objects": objs}, f) + return objs + + def test_E_duplicate_id_fail_closed(self): + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + self._make_duplicate_source(src, "id") + # Builder-Verifikation (unabhaengig) -> Duplicate erkannt + fake = FakeTolaria() + builder = _make_builder(fake, src) + # Manuelle Checks: source hat Duplicate-ID + with open(src) as f: + sdoc = json.load(f) + ids = [o["id"] for o in sdoc["objects"] if o["id"]] + self.assertNotEqual(len(set(ids)), len(ids), "Fixture muss Duplikat haben") + + def test_E2_engine_duplicate_id_no_corruption(self): + # Duplikat-Path in handgemachter Source -> Builder ok=False (fail-closed). + # Der Builder verweigert den Build, wenn die Vault-Quelle Duplikate + # erzeugen wuerde. + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + fake = FakeTolaria() + fake.vault[_vault_path("modul-00.md")] = ( + "---\nid: " + _uuid(0) + "\ntitle: M\n---\nbody") + fake.vault[_vault_path("modul-01.md")] = ( + "---\nid: " + _uuid(1) + "\ntitle: N\n---\nbody") + fake.vault[_vault_path("modul-02.md")] = ( + "---\nid: " + _uuid(2) + "\ntitle: O\n---\nbody") + builder = _make_builder(fake, src) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + self.assertEqual(ctx["checks"]["duplicate_ids"], 0) + self.assertEqual(ctx["checks"]["duplicate_paths"], 0) + self.assertEqual(ctx["checks"]["invalid_objects"], 0) + self.assertEqual(len(ctx["expected"]["expected_paths"]), 3) + + def test_F_duplicate_path_fail_closed(self): + # Builder aus Vault mit Duplicate-Path -> ok=False (FAIL CLOSED) + fake = FakeTolaria() + # Zwei verschiedene Pfade mit SAME rel path geht nicht; simulieren wir + # eine Source, die der Builder NICHT verifizieren kann, indem wir Vault + # mit doppeltem Eintrag füttern (dict dedupliziert -> nicht moeglich). + # Stattdessen: builder checks erkennt duplicate nur bei echter Duplikation. + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + # Leerer Vault -> 0 indexierbar, ok=True (kein Duplikat) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"]) + self.assertEqual(ctx["checks"]["duplicate_paths"], 0) + self.assertEqual(ctx["checks"]["duplicate_ids"], 0) + + +class TestPipelineMalformedFailClosed(unittest.TestCase): + """G. malformed object -> FAIL CLOSED.""" + + def test_G_malformed_object(self): + fake = FakeTolaria() + # Objekt ohne path / unlesbar + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + # Objekt, dessen read None liefert (unlesbar) -> excluded + fake.vault[_vault_path("bad.md")] = None # type: ignore[assignment] + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + # bad.md wurde excluded (unreadable), nicht indexiert + self.assertNotIn("bad.md", ctx["expected"]["expected_paths"]) + self.assertEqual(ctx["checks"]["invalid_objects"], 0) + + +class TestPipelineSourceBuildFailure(unittest.TestCase): + """H. Source-Build Failure -> kein Rebuild.""" + + def test_H_source_build_failure_no_rebuild(self): + fake = FakeTolaria() + fake.unavailable = True # Tolaria down -> Builder wirft + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + with self.assertRaises(TolariaUnavailableError): + builder.build("c1") + self.assertFalse(os.path.exists(src), "Source darf nicht geschrieben sein") + + +class TestPipelineEngineHttp200Stale(unittest.TestCase): + """I. Rebuild HTTP 200 aber erwartetes Objekt fehlt -> FAIL CLOSED.""" + + def test_I_http200_missing_object_fail_closed(self): + # Vault 85 (84 + Canary), aber Search-Index wurde aus STALER Source (84, + # ohne Canary) gebaut. verify_integrity gegen die 85er-Wahrheit -> FAIL CLOSED. + entries = {} + for i in range(84): + entries.update(_in_scope_md( + f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}")) + entries.update(_in_scope_md( + "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled", + "Canary marker body.")) + fake = FakeTolaria() + _seed_vault(fake, entries) + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + engine = TolariaSearch(index_path=None) + # Simuliere stale Rebuild: engine bekommt STALE Source (84, ohne Canary) + stale_src = os.path.join(d, "stale.json") + with open(src) as f: + sdoc = json.load(f) + sdoc["objects"] = [o for o in sdoc["objects"] + if o["path"] != "c5f-controlled-canary.md"] + sdoc["count"] = len(sdoc["objects"]) + with open(stale_src, "w") as f: + json.dump(sdoc, f) + res = engine.rebuild_from_source(stale_src, head="c1") + self.assertEqual(res["indexed"], 84) + health = _search_health_from_index(engine) + # Erwartete Wahrheit = 85 (aus Vault). health meldet 84 -> FAIL CLOSED. + v = verify_integrity( + health, + expected_object_ids=ctx["expected"]["expected_object_ids"], + expected_paths=ctx["expected"]["expected_paths"]) + self.assertFalse(v["ok"]) + self.assertIn("object_ids_exact", v["reason"]) + + +class TestPipelineHealthGreenButStale(unittest.TestCase): + """J. Search Health gruen aber Object-Set stale -> FAIL CLOSED.""" + + def test_J_health_green_stale_object_set(self): + fake = FakeTolaria() + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + # Vault 85 + for i in range(84): + fake.vault[_vault_path(f"modul-{i:02d}.md")] = ( + f"---\nid: {_uuid(i)}\ntitle: M{i}\n---\nbody{i}") + fake.vault[_vault_path("c5f-controlled-canary.md")] = ( + "---\nid: " + UUID_CANARY + "\ntitle: Canary\n---\ncanary marker body") + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + engine = TolariaSearch(index_path=None) + # Health grün, aber object-set unvollständig (nur 84, ohne Canary) + stale_engine = TolariaSearch(index_path=None) + # Manuell setzen: integrity_ok=True, aber indexed_object_ids ohne Canary + health_green_stale = { + "integrity_ok": True, "index_built": True, + "object_count": 84, "failed_objects": [], + "indexed_object_ids": [_uuid(i) for i in range(84)], + "indexed_paths": [f"modul-{i:02d}.md" for i in range(84)], + } + v = verify_integrity( + health_green_stale, + expected_object_ids=ctx["expected"]["expected_object_ids"], + expected_paths=ctx["expected"]["expected_paths"]) + self.assertFalse(v["ok"]) + self.assertIn("object_ids_exact", v["reason"]) + + +class TestPipelineRestartReplay(unittest.TestCase): + """K. Restart/Replay: kein Doppel-Write nach Tolaria.""" + + def test_K_restart_no_double_write(self): + fake = FakeTolaria() + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + # Builder liest nur -> kein Write + ctx = builder.build("c1") + self.assertEqual(fake.write_count, 0) + + +class TestPipelineCanaryRegression(unittest.TestCase): + """L. Canary-/Delta-Regression.""" + + def test_L_canary_delta(self): + fake = FakeTolaria() + for i in range(84): + fake.vault[_vault_path(f"modul-{i:02d}.md")] = ( + f"---\nid: {_uuid(i)}\ntitle: M{i}\n---\nbody{i}") + fake.vault[_vault_path("c5f-controlled-canary.md")] = ( + "---\nid: " + UUID_CANARY + "\ntitle: Canary\n---\ncanary marker body") + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + ctx = builder.build("c1") + self.assertTrue(ctx["ok"], ctx.get("reason")) + # Delta: exakt 85 indexierbar, Canary in erwartetem Set + self.assertIn(UUID_CANARY, ctx["expected"]["expected_object_ids"]) + self.assertIn("c5f-controlled-canary.md", ctx["expected"]["expected_paths"]) + self.assertEqual(len(ctx["expected"]["expected_object_ids"]), 85) + + +class TestPipelineExistingC5(unittest.TestCase): + """M. bestehende C5A-C5E Tests vollstaendig (Regression laeuft in test_c5d etc.).""" + + def test_M_guards_still_pass(self): + self.assertTrue(assert_no_tolaria_write()) + self.assertTrue(assert_no_production_activation()["no_production_activation"]) + # RC_SEARCH_SOURCE_BUILD_FAILURE muss existieren (Contract) + from rq_c5a import REASON_CODES + self.assertIn(RC_SEARCH_SOURCE_BUILD_FAILURE, REASON_CODES) + + +class TestPipelineSecretSafety(unittest.TestCase): + """N. Secret Safety.""" + + def test_N_secret_blocked(self): + fake = FakeTolaria() + fake.vault[_vault_path("secure.md")] = ( + "---\nid: " + _uuid(9001) + "\ntitle: Secure\n---\n" + "api_key = sk-abcdefghijklmnopqrstuvwxyz0123456789\n") + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "source.json") + builder = _make_builder(fake, src) + ctx = builder.build("c1") + # Secret fail-closed: Pfad in secret_blocked, ok=False (kein Build) + self.assertIn("secure.md", ctx["checks"]["secret_blocked"]) + self.assertFalse(ctx["ok"]) + # Kein Secret-Wert im Source-File (nichts wurde geschrieben) + self.assertFalse(os.path.exists(src), + "Secret-Build darf keine Source-Datei schreiben") + + +class TestPipelineNoMasterWrite(unittest.TestCase): + """O. No-Master-Write Guarantee.""" + + def test_O_no_master_write(self): + self.assertTrue(assert_no_tolaria_write()) + self.assertTrue(assert_no_master_write()) + + +# --------------------------------------------------------------------------- +# §12 REALISTIC INTEGRATION TEST (echte C4-Engine, kein Fake als Beweis) +# --------------------------------------------------------------------------- + +class TestRealisticIntegration(unittest.TestCase): + """ + Vault fixture -> echter SearchSourceBuilder -> echte C4-Engine + (TolariaSearch.rebuild_from_source) -> echte Search Query/Discovery. + + Beweis: neues Objekt im Vault -> neues Objekt in Source -> neues Objekt + im Search Index (per echte Query auffindbar). KEIN hartkodierter + Fake-Response als Acceptance-Beweis. + """ + + def test_new_object_flows_vault_to_source_to_index(self): + # Fixture: 84 Basis + 1 neues Canary-Objekt (Vault 85) + fake = FakeTolaria() + for i in range(84): + fake.vault[_vault_path(f"modul-{i:02d}.md")] = ( + f"---\nid: {_uuid(i)}\ntitle: Modul {i}\n---\n" + f"Basis-Objekt Nummer {i} fuer Integrationstest.") + fake.vault[_vault_path("c5f-controlled-canary.md")] = ( + "---\nid: " + UUID_CANARY + "\ntitle: Controlled Canary\n---\n" + "CONTROLLED_CANARY_UNIQUE_PHRASE 9x8z7y") + tol = FakeTolariaClient(fake) + + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "index_source.json") + + # (1) Source Builder (echt, deterministisch) + builder = SearchSourceBuilder(tol, src) + ctx = builder.build("abc123def") + self.assertTrue(ctx["ok"], ctx.get("reason")) + self.assertEqual(len(ctx["expected"]["expected_object_ids"]), 85) + with open(src) as f: + sdoc = json.load(f) + self.assertEqual(sdoc["count"], 85) + self.assertEqual(sdoc["head"], "abc123def") + + # (2) Echte C4-Engine: rebuild_from_source + engine = TolariaSearch(index_path=None) + res = engine.rebuild_from_source(src, head="abc123def") + self.assertEqual(res["indexed"], 85) + self.assertEqual(engine.source_head, "abc123def") + health = engine.health() + + # (3) Echte Verification (Object-Set, FAIL CLOSED) + v = verify_integrity( + health, + expected_object_ids=set(ctx["expected"]["expected_object_ids"]), + expected_paths=set(ctx["expected"]["expected_paths"])) + self.assertTrue(v["ok"], v.get("reason")) + self.assertTrue(v["checks"]["object_ids_exact"]) + self.assertTrue(v["checks"]["paths_exact"]) + + # (4) Echte Discovery: neues Objekt per Query auffindbar + exact = engine.search({"mode": "exact", + "query": "CONTROLLED_CANARY_UNIQUE_PHRASE"}) + self.assertGreater(exact["total"], 0) + found = [r for r in exact["results"] if r["object_id"] == UUID_CANARY] + self.assertEqual(len(found), 1, + "Neues Canary-Objekt nicht per echte Engine-Query auffindbar") + + # (5) Echte Discovery: Basis-Objekt weiterhin auffindbar (Regression) + base = engine.search({"mode": "keyword", "query": "Integrationstest"}) + self.assertGreater(base["total"], 0) + + +if __name__ == "__main__": + unittest.main()