diff --git a/tolaria/c5-sync-service/README.md b/tolaria/c5-sync-service/README.md index c5f2f29..9d2f7b1 100644 --- a/tolaria/c5-sync-service/README.md +++ b/tolaria/c5-sync-service/README.md @@ -1,32 +1,142 @@ -# C5A — CONTRACT & STATE MACHINE v1 +# C5 SYNC SERVICE — CONTRACT & STATE MACHINE (C5A) + FORGEJO POLLING & CHANGE DETECTION (C5B) -**Phase:** C5A (erste Phase des C5 Sync Service) -**Status:** DONE (Contract & State Machine) — **INAKTIV, NO-WRITE** +**Phase:** C5A (Contract & State Machine) + C5B (Forgejo Polling & Change Detection) +**Status:** C5A = DONE · C5B = DONE (Polling & Change Detection) — **INAKTIV, NO-WRITE** **Architektur:** FORGEJO MASTER → C5 SYNC SERVICE → TOLARIA DERIVED → SEARCH FULL REBUILD C5A implementiert **ausschließlich** den Contract + die Sync State Machine + die -persistente Progress-State-Struktur. Es ist eine **deterministische, inaktive -Python-Library + CLI + Testsuite** — es führt **keine** externen Writes aus. +persistente Progress-State-Struktur. C5B baut darauf die **Forgejo Polling & +Change Detection Engine** (read-only). Beide sind **deterministische, inaktive +Python-Libraries + CLI + Testsuite** — sie führen **keine** externen Writes aus. --- -## NO-WRITE GUARANTEE (verbindlich) +## NO-WRITE GUARANTEE (verbindlich, C5A + C5B) -C5A kann **NOCH NICHT**: +C5A + C5B können **NOCH NICHT**: - Tolaria schreiben - Search rebuilden - Forgejo schreiben -- produktiv pollen +- produktiv pollen (kein Poll-Daemon) - Netzwerk-Mutationen ausführen -Es gibt **keine** Netzwerk-/HTTP-/Socket-/Subprocess-Mutationsfunktionen in C5A. -Die No-Write-Guarantee wird statisch per `assert_no_write_guarantee()` bewiesen -(keine `requests`/`urllib`/`http`/`socket`/`subprocess`-Imports) und durch den -Test `test_no_write_guarantee` abgesichert. +**C5A** hat **keine** Netzwerk-/HTTP-/Socket-/Subprocess-Mutationsfunktionen +(`assert_no_write_guarantee()`). + +**C5B** führt **ausschließlich read-only git-Befehle** gegen den lokalen Clone aus +(strikte Whitelist: `rev-parse`, `log`, `show`, `diff-tree`, `ls-tree`, +`cat-file`, `merge-base`, `rev-list`, `diff`). Jeder Write-Befehl +(`push`/`commit`/`add`/`reset`/`checkout`/`merge`/`rebase`) wird mit +`GitWriteBlockedError` blockiert. Die No-Downstream-Write-Guarantee wird statisch +per `assert_no_downstream_write()` bewiesen (keine HTTP-Mutationsimporte, keine +`/api/vault/save`-/`/api/search/rebuild`-Endpoints im Code, keine Git-Write-Befehle +in der Whitelist) und durch den Test `test_no_downstream_write` abgesichert. --- -## State Machine +## C5B — FORGEJO POLLING & CHANGE DETECTION ENGINE + +### FORGEJO_READ_METHOD +**A) lokale Git-Mirror/Clone-Analyse** (bevorzugt). +- **AUTH_REQUIREMENT:** KEIN (read-only gegen lokalen Clone; kein Write-Token) +- **NETWORK_DEPENDENCY:** KEIN (lokale `.git`-Objekte; kein HTTP) + +### POLLING_MODEL +- **Default:** 60 Sekunden +- **Environment-konfigurierbar:** `C5_POLL_INTERVAL_SECONDS` (z.B. `=120`) +- C5B implementiert die **Polling-Logik**, startet aber **KEINEN** dauerhaften + Produktions-Daemon. Tests führen einzelne kontrollierte `poll_once`-Zyklen aus. + +### HEAD_DISCOVERY +`get_current_head()` → Commit-SHA + parent/timestamp/message. Bei nicht +erreichbarem Forgejo: `RC_FORGEJO_UNAVAILABLE`, kein State-Fortschritt. + +### COMMIT_RANGE_MODEL +Aus `last_seen_commit` (C5B-Baseline) bis HEAD deterministisch alle fehlenden +Commits in chronologischer (parent-korrekter) Reihenfolge. Kein Überspringen, +keine Doppelerfassung. Merge-Commits werden explizit behandelt (alle Parents). +Divergenz/History-Rewrite wird erkannt. + +> **Wichtig:** C5B-Baseline ist `last_seen_commit`. `last_applied_commit` bleibt +> laut C5A-Contract für die spätere Propagation (C5C) reserviert und wird von C5B +> **NIE** fortgeschrieben. + +### HISTORY_SAFETY +Wenn `last_seen_commit` kein Ancestor von HEAD ist: **FAIL CLOSED** +(`RC_OUT_OF_ORDER_COMMIT`). Keine automatische Rebase-/Reset-Logik. + +### DIFF_MODEL +Für jeden neuen Commit: `git diff-tree -r --name-status -M -C --root` → +`ADDED` / `MODIFIED` / `DELETED` / `RENAMED` / `MOVED`. Bei unsicherer +Rename-Detection wird **nicht geraten** — Ambiguität wird sauber klassifiziert +(Human Review). + +### KNOWLEDGE_SCOPE_MODEL +Nicht jede Forgejo-Datei ist automatisch Tolaria-Knowledge: +- **IN_SCOPE:** Markdown im Repo-Root ODER unter `notes/trading/system-docs/` +- **OUT_OF_SCOPE:** `tolaria/`, `a2-5/`, `red-queen-architecture/`, + `notion-safety-brain/`, `notion-command-center/`, `backup_patches/`, + `notes/reference/`, Nicht-Markdown, Build-Artefakte +- **LEGACY_SPECIAL:** `README.md` (x2, KEEP_DISTINCT) + `vps.md` (DEFERRED) +- **HUMAN_REVIEW:** Knowledge-Objekt ohne gültige `object_id` + +Keine Secrets/Build-Artefakte in Sync-Change-Sets. Nicht blind `.md` = Sync. + +### OBJECT_ID_MODEL +`object_id` wird **aus dem Frontmatter** gelesen (`id: object/`), +nicht aus dem Pfad erzeugt, nicht neu generiert, nicht geraten. Ohne gültige ID: +`UNKNOWN_OBJECT_ID` / `UNKNOWN_LEGACY_OBJECT` → Human Gate. + +### HASH_MODEL +- **`content_hash`:** SHA-256 des fachlichen Bodies (nach Frontmatter) +- **`metadata_hash`:** SHA-256 der relevanten Knowledge-Schema-Metadaten + (`id`, `type`, `role`, `representation`, `state`, `derived_from`, `tags`, + `knowledge_schema`). Pfad ist separat. +- Deterministisch und testbar. + +### CHANGE_CLASSIFICATION +Aus Before/After deterministisch: `CREATE`, `CONTENT_UPDATE`, +`METADATA_UPDATE`, `STATE_UPDATE`, `RENAME`, `MOVE`, +`SOURCE_CANONICAL_RELATION_UPDATE`, `TAGS_UPDATE`, `SUPERSEDE`, +`DELETE_REQUEST`. Bei mehreren geänderten Aspekten in EINEM Objekt werden +**mehrere ObjectChange-Einträge** erzeugt (keine Information verloren). + +### RENAME_MOVE_MODEL +Rename/Move nur automatisch, wenn **gleiche `object_id`** + eindeutige +Git-/Content-Evidence. ID bleibt stabil. Rename = Dateiname geändert (gleiches +Verzeichnis); Move = Verzeichnis geändert. Bei Ambiguität: Human Gate. + +### DELETE_MODEL +Git-Delete wird **NIEMALS** zu automatischem Hard Delete. C5B erzeugt +`DELETE_REQUEST` → C5A führt es in Human Review. Keine Tolaria-Aktion. + +### PAIR_MODEL +Wenn Root/source + canonical gemeinsam verändert werden, erkennt C5B die +Pair-Beziehung (`derived_from`). Ordnung für spätere C5C: Source zuerst, +Canonical danach. Noch nicht propagiert — Output enthält genug Information für +C5C. + +### SECRET_SAFETY +C5B schreibt **keine** Secret-Inhalte in Logs/SQLite/Evidence. Bei +Secret-/Forbidden-Dateien: nur sichere Metadaten (`path`, `reason_code`, +`commit_sha`). Enthält ein Knowledge-Dokument einen möglichen Secret-Wert: +`RC_SECRET_DETECTED` FAIL CLOSED. Kein Wert wird ausgegeben. + +### C5A_INTEGRATION +C5B speist gefundene Commits + ObjectChanges in den C5A Store ein (nur +State/Persistence, kein Network Write downstream). Flow: Poll once → discover +commits → classify changes → register commit → add object changes → Status +`DISCOVERED`/`VALIDATING` (bzw. `HUMAN_REVIEW_REQUIRED`). **Nicht** weiter zu +`PROPAGATING_TOLARIA`. + +### OBSERVABILITY +C5B erweitert sichere Statusdaten: `last_seen_commit`, `poll_timestamp`, +`commits_discovered`, `objects_discovered`, `out_of_scope_count`, +`last_poll_status`, `last_poll_error_code`. Keine Content-Logs. + +--- + +## State Machine (C5A) ### Normalzustände `DISCOVERED → VALIDATING → READY → PROPAGATING_TOLARIA → VERIFYING_TOLARIA → UPDATING_SEARCH → VERIFYING_SEARCH → APPLIED` @@ -169,6 +279,10 @@ C5A führt **keine echten externen Health-Probes** aus (rein State-Machine) — ```bash export C5A_DB=/path/to/c5a.db # default: ./c5a.db +export C5_REPO=/path/to/forgejo-clone # default: . +export C5_POLL_INTERVAL_SECONDS=60 # default: 60 + +# C5A python3 rq_c5a_cli.py health python3 rq_c5a_cli.py bootstrap-state python3 rq_c5a_cli.py bootstrap-transition --to RECONCILING @@ -178,30 +292,52 @@ python3 rq_c5a_cli.py process-commit --json python3 rq_c5a_cli.py list-commits [--status ] python3 rq_c5a_cli.py commit-status --sha python3 rq_c5a_cli.py no-write-check + +# C5B (rq_c5_cli.py) +python3 rq_c5_cli.py poll-once # ein kontrollierter Poll-Zyklus (read-only) +python3 rq_c5_cli.py show-pending # ausstehende Commits anzeigen +python3 rq_c5_cli.py show-commit # Details eines Commits anzeigen +python3 rq_c5_cli.py dry-run-diff # Diff eines Commits anzeigen (read-only) +python3 rq_c5_cli.py health # Health-Contract anzeigen +python3 rq_c5_cli.py no-write-check # No-Downstream-Write-Guarantee pruefen ``` +**Kein** `daemon`/`start-forever`-Befehl in C5B — der produktive Poll-Daemon kommt +erst beim Deployment (C5). + --- ## Tests ```bash python3 test_c5a.py # 25 PASS / 0 FAIL +python3 -m unittest test_c5b -v # 39 PASS / 0 FAIL ``` -Abgedeckte Fälle: normal commit lifecycle, multi-object commit, duplicate commit, -already-applied, out-of-order commit, missing predecessor, retry progression, -max retry → DEAD, human gate transition, restart/reload persistence, state -corruption handling, idempotency, delete request → human gate, dangling relation -→ human gate, unexpected drift → human gate, no-write guarantee, bootstrap +**C5A** abgedeckte Fälle: normal commit lifecycle, multi-object commit, duplicate +commit, already-applied, out-of-order commit, missing predecessor, retry +progression, max retry → DEAD, human gate transition, restart/reload persistence, +state corruption handling, idempotency, delete request → human gate, dangling +relation → human gate, unexpected drift → human gate, no-write guarantee, bootstrap lifecycle, baseline-only-from-BASELINE_READY, reason codes closed set, operations closed set, health contract fields, last_applied-only-after-full-pass, retry available bound, backoff sequence, invalid reason code rejected. +**C5B** abgedeckte Fälle: head discovery, single commit, multi commit, added +knowledge object, content update, metadata update, state update, tags update, +relation update, rename, move, delete request, out-of-scope file, legacy object, +unknown ID, secret detection, duplicate poll, lost poll, ordering, Forgejo +unavailable, history divergence, C5A persistence integration, no downstream write, +poll interval config. + --- ## Dateien -- `rq_c5a.py` — Kern-Library (Store + State Machine + No-Write-Check) -- `rq_c5a_cli.py` — CLI -- `test_c5a.py` — Testsuite (25 Tests) +- `rq_c5a.py` — C5A Kern-Library (Store + State Machine + No-Write-Check) +- `rq_c5a_cli.py` — C5A CLI +- `test_c5a.py` — C5A Testsuite (25 Tests) +- `rq_c5b.py` — C5B Kern-Library (GitReader + Scope + Classifier + Poller + No-Downstream-Write-Check) +- `rq_c5_cli.py` — C5 CLI (C5A + C5B: poll-once, show-pending, show-commit, dry-run-diff, health, no-write-check) +- `test_c5b.py` — C5B Testsuite (39 Tests) - `README.md` — diese Datei diff --git a/tolaria/c5-sync-service/rq_c5_cli.py b/tolaria/c5-sync-service/rq_c5_cli.py new file mode 100644 index 0000000..9e55e9b --- /dev/null +++ b/tolaria/c5-sync-service/rq_c5_cli.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5 CLI (C5A + C5B): kontrollierte Inspektion & Dry-Run. + +KEIN produktiver Poll-Daemon. KEINE daemon/start-forever-Funktion. +Nur explizite, begrenzte Befehle: + * poll-once — ein kontrollierter Poll-Zyklus (read-only) + * show-pending — ausstehende Commits anzeigen + * show-commit — Details eines Commits anzeigen + * dry-run-diff — Diff eines Commits anzeigen (read-only) + * health — Health-Contract anzeigen + * no-write-check — No-Downstream-Write-Guarantee pruefen +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from typing import Any, Dict, List, Optional + +from rq_c5a import C5AStore +from rq_c5b import ( + C5BPoller, + GitReader, + assert_no_downstream_write, + DEFAULT_POLL_INTERVAL_SECONDS, + ENV_POLL_INTERVAL, +) + + +def _store(db_path: str) -> C5AStore: + return C5AStore(db_path) + + +def _poller(store: C5AStore, repo_path: str, interval: Optional[int]) -> C5BPoller: + return C5BPoller(store, repo_path, poll_interval_seconds=interval) + + +def cmd_poll_once(args: argparse.Namespace) -> int: + store = _store(args.db) + poller = _poller(store, args.repo, args.interval) + result = poller.poll_once() + print(json.dumps(result, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_show_pending(args: argparse.Namespace) -> int: + store = _store(args.db) + pending = store.list_commits() + print(json.dumps(pending, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_show_commit(args: argparse.Namespace) -> int: + store = _store(args.db) + commit = store.get_commit(args.sha) + if commit is None: + print(f"Commit nicht gefunden: {args.sha}") + store.close() + return 1 + print(json.dumps(commit, ensure_ascii=False, indent=2)) + objs = store.list_object_changes(args.sha) + if objs: + print("--- Object Changes ---") + print(json.dumps(objs, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_dry_run_diff(args: argparse.Namespace) -> int: + reader = GitReader(args.repo) + changes = reader.diff_name_status(args.sha) + print(json.dumps(changes, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_health(args: argparse.Namespace) -> int: + store = _store(args.db) + print(json.dumps(store.health(), ensure_ascii=False, indent=2)) + store.close() + return 0 + + +def cmd_no_write_check(args: argparse.Namespace) -> int: + result = assert_no_downstream_write() + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if result["no_downstream_write"] else 1 + + +def main(argv: Optional[List[str]] = None) -> int: + parser = argparse.ArgumentParser( + prog="rq_c5_cli", + description="Red Queen C5 CLI (C5A + C5B) — kontrollierte Inspektion & Dry-Run", + ) + parser.add_argument( + "--db", default=os.environ.get("C5A_DB", "c5a.db"), + help="Pfad zur C5A-SQLite-DB (Env C5A_DB, Default c5a.db)", + ) + parser.add_argument( + "--repo", default=os.environ.get("C5_REPO", "."), + help="Pfad zum Forgejo-Clone (Env C5_REPO, Default .)", + ) + parser.add_argument( + "--interval", type=int, default=None, + help=f"Poll-Intervall in Sekunden (Env {ENV_POLL_INTERVAL}, Default {DEFAULT_POLL_INTERVAL_SECONDS})", + ) + + sub = parser.add_subparsers(dest="command", required=True) + + p = sub.add_parser("poll-once", help="Ein kontrollierter Poll-Zyklus (read-only)") + p.set_defaults(func=cmd_poll_once) + + p = sub.add_parser("show-pending", help="Ausstehende Commits anzeigen") + p.set_defaults(func=cmd_show_pending) + + p = sub.add_parser("show-commit", help="Details eines Commits anzeigen") + p.add_argument("sha", help="Commit-SHA") + p.set_defaults(func=cmd_show_commit) + + p = sub.add_parser("dry-run-diff", help="Diff eines Commits anzeigen (read-only)") + p.add_argument("sha", help="Commit-SHA") + p.set_defaults(func=cmd_dry_run_diff) + + p = sub.add_parser("health", help="Health-Contract anzeigen") + p.set_defaults(func=cmd_health) + + p = sub.add_parser("no-write-check", help="No-Downstream-Write-Guarantee pruefen") + p.set_defaults(func=cmd_no_write_check) + + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tolaria/c5-sync-service/rq_c5b.py b/tolaria/c5-sync-service/rq_c5b.py new file mode 100644 index 0000000..819a828 --- /dev/null +++ b/tolaria/c5-sync-service/rq_c5b.py @@ -0,0 +1,856 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5B: FORGEJO POLLING & CHANGE DETECTION ENGINE v1 (read-only). + +VERBINDLICHER RAHMEN +==================== +C5B ist die zweite Phase des C5 Sync Service (FORGEJO MASTER -> C5 SYNC SERVICE -> +TOLARIA DERIVED -> SEARCH FULL REBUILD). C5B baut AUSSCHLIESSLICH: + * FORGEJO READ (lokale Git-Clone-Analyse, read-only) + * COMMIT DISCOVERY (HEAD + Commit-Range) + * COMMIT ORDERING (chronologisch, parent-korrekt) + * DIFF / CHANGE DETECTION (ADDED/MODIFIED/DELETED/RENAMED/MOVED) + * OBJECT CHANGE CONTRACT (BEFORE/AFTER, Hashes, Operationen) + * Uebergabe an den C5A State Store (nur State/Persistence) + +NOCH KEINE PROPAGATION. C5B propagiert NICHT nach Tolaria, rebuildet NICHT den +Search, schreibt NICHT nach Forgejo, startet KEINEN produktiven Poll-Daemon. + +ABSOLUTES WRITE-VERBOT (C5B): + * KEIN Forgejo-Write / kein git push / kein git commit / kein git add. + * KEIN Tolaria-Write / kein POST /api/vault/save. + * KEIN Search-Rebuild / kein POST /api/search/rebuild. + * KEIN Container-Deploy, KEINE Netzwerk-Aenderung, KEINE Hermes-Rechte. + * KEIN produktiver Polling-Daemon (nur kontrollierte poll_once-Zyklen in Tests). + * KEIN C5C / C5D / C5-Deployment. + +FORGEJO_READ_METHOD = A) lokale Git-Mirror/Clone-Analyse (bevorzugt). + * AUTH_REQUIREMENT = KEIN (read-only gegen lokalen Clone; kein Write-Token). + * NETWORK_DEPENDENCY = KEIN (lokale .git-Objekte; kein HTTP). + * Der GitReader fuehrt NUR read-only git-Befehle aus (strikte Whitelist). + +SECURITY BOUNDARY (dokumentiert, NICHT implementiert): + * Forgejo credential: READ ONLY (C5B haelt KEINEN Forgejo-Write). + * Tolaria write: nur spaeter C5 Service (C5B schreibt NIE nach Tolaria). + * Search rebuild: nur C5 Service Token (C5B ruft NIE Search-Rebuild). + * Agents: keine direkten Tolaria-/Search-Admin-Writes. + * Netzwerk-Haertung wird spaeter beim Deployment umgesetzt, NICHT jetzt. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import subprocess +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from rq_c5a import ( + C5AStore, + ST_DISCOVERED, + ST_VALIDATING, + ST_WAITING_FOR_PREDECESSOR, + ST_HUMAN_REVIEW_REQUIRED, + ST_APPLIED, + OP_CREATE, + OP_CONTENT_UPDATE, + OP_METADATA_UPDATE, + OP_STATE_UPDATE, + OP_RENAME, + OP_MOVE, + OP_SOURCE_CANONICAL_RELATION_UPDATE, + OP_TAGS_UPDATE, + OP_SUPERSEDE, + OP_DELETE_REQUEST, + RC_FORGEJO_UNAVAILABLE, + RC_OUT_OF_ORDER_COMMIT, + RC_UNKNOWN_OBJECT_ID, + RC_UNKNOWN_LEGACY_OBJECT, + RC_SECRET_DETECTED, + RC_INVALID_SCHEMA, + RC_DANGLING_DERIVED_FROM, + RC_AMBIGUOUS_DELETE, + REASON_CODES, + OPERATIONS, +) + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +# Polling-Intervall (Default 60s, Environment-konfigurierbar) +DEFAULT_POLL_INTERVAL_SECONDS = 60 +ENV_POLL_INTERVAL = "C5_POLL_INTERVAL_SECONDS" + +# Knowledge-Scope-Klassifikation +SCOPE_IN_SCOPE = "IN_SCOPE" +SCOPE_OUT_OF_SCOPE = "OUT_OF_SCOPE" +SCOPE_LEGACY_SPECIAL = "LEGACY_SPECIAL" +SCOPE_HUMAN_REVIEW = "HUMAN_REVIEW" + +# Verzeichnisse, die KEINE Tolaria-Knowledge-Objekte enthalten (Architektur/Ops/Code) +OUT_OF_SCOPE_DIRS = frozenset({ + "tolaria", "a2", "a3", "a4", "a5", + "red-queen-architecture", "notion-safety-brain", "notion-command-center", + "backup_patches", "notes/reference", "notes", +}) + +# Legacy-Ausnahmen (C3I: KEEP_DISTINCT / DEFERRED) — C5B darf diese NICHT +# zusammenfuehren oder neu-vergeben. Sie werden separat klassifiziert. +LEGACY_SPECIAL_PATHS = frozenset({ + "README.md", # Root-README (KEEP_DISTINCT) + "notes/trading/system-docs/README.md", # Canonical-README (KEEP_DISTINCT) + "vps.md", # DEFERRED (C3I) +}) + +# Canonical-Verzeichnis (derived knowledge) +CANONICAL_DIR = "notes/trading/system-docs/" + +# Git read-only Whitelist (C5B darf NUR diese Befehle ausfuehren) +# Jeder Befehl ist read-only. Write-Befehle (push/commit/add/reset/checkout/ +# merge/rebase/fetch --write) sind NICHT in der Whitelist und werden blockiert. +GIT_READONLY_COMMANDS = frozenset({ + "rev-parse", "log", "show", "diff-tree", "ls-tree", "cat-file", + "merge-base", "rev-list", "name-only", "diff", +}) + +# Git Write-Befehle (explizit verboten — statische + dynamische Pruefung) +GIT_WRITE_COMMANDS = frozenset({ + "push", "commit", "add", "reset", "checkout", "merge", "rebase", + "fetch", "pull", "clone", "init", "rm", "mv", "tag", "branch", + "stash", "clean", "gc", "prune", "repack", "update-ref", "write-tree", +}) + +# Frontmatter-Id-Muster: object/ +OBJECT_ID_RE = re.compile(r"^object/[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") + +# Secret-/Forbidden-Muster (nur zur Klassifikation, KEINE Werte persistieren) +SECRET_PATTERNS = [ + re.compile(r"sk-[A-Za-z0-9]{20,}"), + re.compile(r"ghp_[A-Za-z0-9]{20,}"), + re.compile(r"bearer\s+[A-Za-z0-9]{20,}", re.IGNORECASE), + re.compile(r"-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----"), + re.compile(r"AKIA[0-9A-Z]{16}"), +] + +# Knowledge-Schema-Metadaten, die in metadata_hash eingehen +METADATA_FIELDS = ("id", "type", "role", "representation", "state", + "derived_from", "tags", "knowledge_schema") + + +# --------------------------------------------------------------------------- +# Fehlerklassen +# --------------------------------------------------------------------------- + +class C5BError(Exception): + """Basis-Fehlerklasse fuer C5B.""" + + def __init__(self, message: str, reason_code: Optional[str] = None): + super().__init__(message) + self.message = message + self.reason_code = reason_code + + +class ForgejoUnavailableError(C5BError): + """Forgejo (lokaler Clone) nicht lesbar.""" + + +class HistoryDivergenceError(C5BError): + """last_applied_commit ist kein Ancestor von HEAD (FAIL CLOSED).""" + + +class GitWriteBlockedError(C5BError): + """Versuch, einen verbotenen git-Write-Befehl auszufuehren.""" + + +class NoDownstreamWriteError(C5BError): + """C5B darf keine externen Writes ausfuehren.""" + + +# --------------------------------------------------------------------------- +# GitReader (read-only, strikte Whitelist) +# --------------------------------------------------------------------------- + +class GitReader: + """ + Fuehrt AUSSCHLIESSLICH read-only git-Befehle gegen einen lokalen Clone aus. + + Whitelist: rev-parse, log, show, diff-tree, ls-tree, cat-file, merge-base, + rev-list, diff. Jeder andere Befehl (insb. push/commit/add/reset/checkout/ + merge/rebase) wird mit GitWriteBlockedError blockiert. + """ + + def __init__(self, repo_path: str): + self.repo_path = str(repo_path) + if not os.path.isdir(os.path.join(self.repo_path, ".git")): + raise ForgejoUnavailableError( + f"Kein Git-Repo unter {self.repo_path}", + reason_code=RC_FORGEJO_UNAVAILABLE, + ) + + def _run(self, args: List[str], check: bool = True) -> str: + """Fuehrt einen read-only git-Befehl aus. Blockiert Write-Befehle.""" + if not args: + raise GitWriteBlockedError("Leerer git-Befehl") + cmd = args[0] + if cmd not in GIT_READONLY_COMMANDS: + raise GitWriteBlockedError( + f"git-Befehl '{cmd}' ist nicht in der read-only Whitelist" + ) + if cmd in GIT_WRITE_COMMANDS: + raise GitWriteBlockedError( + f"git-Befehl '{cmd}' ist ein verbotener Write-Befehl" + ) + try: + proc = subprocess.run( + ["git", "-C", self.repo_path] + args, + capture_output=True, text=True, timeout=30, + ) + except (subprocess.TimeoutExpired, OSError) as e: + raise ForgejoUnavailableError( + f"Forgejo (git) nicht erreichbar: {e}", + reason_code=RC_FORGEJO_UNAVAILABLE, + ) + if check and proc.returncode != 0: + raise ForgejoUnavailableError( + f"git {cmd} fehlgeschlagen: {proc.stderr.strip()}", + reason_code=RC_FORGEJO_UNAVAILABLE, + ) + return proc.stdout + + def get_current_head(self) -> str: + """Gibt den aktuellen HEAD-Commit-SHA zurueck (read-only).""" + out = self._run(["rev-parse", "HEAD"]).strip() + if not out: + raise ForgejoUnavailableError( + "HEAD nicht bestimmbar", reason_code=RC_FORGEJO_UNAVAILABLE + ) + return out + + def commit_meta(self, sha: str) -> Dict[str, Any]: + """Gibt parent, timestamp, message eines Commits zurueck (read-only).""" + out = self._run(["show", "-s", "--format=%H%x00%P%x00%ct%x00%s", sha]).strip() + parts = out.split("\x00") + if len(parts) < 4: + raise ForgejoUnavailableError( + f"Commit-Metadaten nicht lesbar: {sha}", + reason_code=RC_FORGEJO_UNAVAILABLE, + ) + return { + "commit_sha": parts[0], + "parent_sha": parts[1] or None, # Root-Commit hat keinen Parent + "timestamp": int(parts[2]) if parts[2].isdigit() else None, + "message": parts[3], + } + + def is_ancestor(self, ancestor: str, descendant: str) -> bool: + """ + Prueft, ob ancestor ein Vorfahr von descendant ist (read-only). + git merge-base --is-ancestor: exit 0 = ist Ancestor, exit 1 = nicht. + """ + try: + proc = subprocess.run( + ["git", "-C", self.repo_path, + "merge-base", "--is-ancestor", ancestor, descendant], + capture_output=True, text=True, timeout=30, + ) + return proc.returncode == 0 + except (subprocess.TimeoutExpired, OSError) as e: + raise ForgejoUnavailableError( + f"Forgejo (git) nicht erreichbar: {e}", + reason_code=RC_FORGEJO_UNAVAILABLE, + ) + + def commit_range(self, start_exclusive: Optional[str], end: str) -> List[str]: + """ + Deterministisch alle Commits in (start_exclusive, end] in chronologischer + (parent-korrekter) Reihenfolge. Kein Ueberspringen, keine Doppelerfassung. + """ + if start_exclusive is None: + # Alle Commits bis end (reverse = aeltester zuerst) + out = self._run(["rev-list", "--reverse", end]).strip() + else: + out = self._run(["rev-list", "--reverse", f"{start_exclusive}..{end}"]).strip() + if not out: + return [] + return [line for line in out.splitlines() if line] + + def diff_name_status(self, commit_sha: str) -> List[Dict[str, Any]]: + """ + Gibt die Aenderungen eines Commits gegenueber seinem Parent zurueck. + Nutzt git diff-tree (read-only) mit rename detection. + """ + out = self._run([ + "diff-tree", "-r", "--name-status", "-M", "-C", "--root", + "--format=", commit_sha, + ]).strip() + changes = [] + for line in out.splitlines(): + parts = line.split("\t") + if len(parts) < 2: + continue + status = parts[0] + path = parts[1] + # Rename/Copy: "R100\told\tnew" oder "C100\told\tnew" + if status.startswith("R") or status.startswith("C"): + if len(parts) >= 3: + changes.append({ + "status": status[0], # R oder C + "similarity": status[1:], + "path_before": parts[1], + "path_after": parts[2], + }) + continue + # ADDED: Datei existiert nur in AFTER (path_before=None) + if status == "A": + changes.append({ + "status": status, + "path_before": None, + "path_after": path, + }) + continue + # DELETED: Datei existiert nur in BEFORE (path_after=None) + if status == "D": + changes.append({ + "status": status, + "path_before": path, + "path_after": None, + }) + continue + # MODIFIED (M, T, U): beide Pfade = path + changes.append({ + "status": status, + "path_before": path, + "path_after": path, + }) + return changes + + def file_content(self, sha: str, path: str) -> Optional[str]: + """Liest den Inhalt einer Datei in einem Commit (read-only).""" + try: + out = self._run(["show", f"{sha}:{path}"], check=False) + except ForgejoUnavailableError: + return None + if out == "" and self._file_exists(sha, path) is False: + return None + return out + + def _file_exists(self, sha: str, path: str) -> Optional[bool]: + """Prueft, ob eine Datei in einem Commit existiert (read-only).""" + try: + out = self._run(["ls-tree", "-r", "--name-only", sha], check=False) + except ForgejoUnavailableError: + return None + return path in out.splitlines() + + +# --------------------------------------------------------------------------- +# Frontmatter / Object ID / Hash Model +# --------------------------------------------------------------------------- + +def parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: + """ + Parst YAML-Frontmatter (--- ... ---) und gibt (fm_dict, body) zurueck. + Body = fachlicher Inhalt NACH dem Frontmatter-Block. + """ + if not content.startswith("---"): + return {}, content + lines = content.splitlines() + if len(lines) < 2: + return {}, content + # Ende des Frontmatter-Blocks finden + end = None + for i in range(1, len(lines)): + if lines[i].strip() == "---": + end = i + break + if end is None: + return {}, content + fm_lines = lines[1:end] + body = "\n".join(lines[end + 1:]) + fm: Dict[str, Any] = {} + for line in fm_lines: + if ":" not in line: + continue + key, _, val = line.partition(":") + key = key.strip() + val = val.strip() + if not key: + continue + # Einfache Listen (tags: [a, b]) und Skalare + if val.startswith("[") and val.endswith("]"): + inner = val[1:-1].strip() + fm[key] = [x.strip().strip("'\"") for x in inner.split(",") if x.strip()] + else: + fm[key] = val.strip("'\"") + return fm, body + + +def extract_object_id(content: str) -> Optional[str]: + """Liest object_id aus dem Frontmatter (id: object/).""" + fm, _ = parse_frontmatter(content) + oid = fm.get("id") + if not oid: + return None + oid = str(oid).strip() + if not OBJECT_ID_RE.match(oid): + return None + return oid + + +def content_hash(body: str) -> str: + """SHA-256 des fachlichen Bodies (deterministisch).""" + return hashlib.sha256(body.encode("utf-8")).hexdigest() + + +def metadata_hash(fm: Dict[str, Any]) -> str: + """ + SHA-256 der relevanten Knowledge-Schema-Metadaten (deterministisch). + Beruecksichtigt: id, type, role, representation, state, derived_from, + tags, knowledge_schema. Pfad ist separat (nicht Teil von metadata_hash). + """ + subset = {k: fm.get(k) for k in METADATA_FIELDS if k in fm} + canonical = json.dumps(subset, sort_keys=True, ensure_ascii=False) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def detect_secret(content: str) -> Optional[str]: + """ + Erkennt moegliche Secret-Werte in einem Dokument. Gibt den MATCHED PATTERN + zurueck (NICHT den Wert). Kein Secret-Wert wird ausgegeben oder persistiert. + """ + for pat in SECRET_PATTERNS: + m = pat.search(content) + if m: + return pat.pattern + return None + + +# --------------------------------------------------------------------------- +# Knowledge Scope +# --------------------------------------------------------------------------- + +class KnowledgeScope: + """ + Bestimmt, ob eine Forgejo-Datei ein Tolaria-Knowledge-Objekt ist. + + Regeln (C3/C5): + * IN_SCOPE: Markdown im Repo-Root ODER unter notes/trading/system-docs/, + mit gueltiger object_id im Frontmatter. + * OUT_OF_SCOPE: Architektur/Ops/Code-Verzeichnisse, Nicht-Markdown, + Build-Artefakte, Secrets. + * LEGACY_SPECIAL: README (x2, KEEP_DISTINCT) + vps.md (DEFERRED). + * HUMAN_REVIEW: Knowledge-Objekt ohne gueltige object_id (UNKNOWN_OBJECT_ID + / UNKNOWN_LEGACY_OBJECT) oder Ambiguitaet. + """ + + def __init__(self, repo_path: str): + self.repo_path = str(repo_path) + + def classify(self, path: str) -> str: + """Klassifiziert einen Pfad in IN_SCOPE / OUT_OF_SCOPE / LEGACY_SPECIAL.""" + path = path.replace("\\", "/") + # Legacy-Ausnahmen zuerst + if path in LEGACY_SPECIAL_PATHS: + return SCOPE_LEGACY_SPECIAL + # Nur Markdown ist potenziell Knowledge + if not path.endswith(".md"): + return SCOPE_OUT_OF_SCOPE + # Canonical-Verzeichnis + if path.startswith(CANONICAL_DIR): + return SCOPE_IN_SCOPE + # Repo-Root (kein Verzeichnis-Praefix) + if "/" not in path: + return SCOPE_IN_SCOPE + # Alles andere (tolaria/, a2-5/, red-queen-architecture/, etc.) = OUT + return SCOPE_OUT_OF_SCOPE + + def classify_with_content(self, path: Optional[str], content: Optional[str]) -> Dict[str, Any]: + """ + Vollstaendige Scope-Klassifikation inkl. object_id-Validierung. + Gibt ein Dict mit scope, object_id, reason_code (optional) zurueck. + """ + if path is None: + return {"path": None, "scope": SCOPE_OUT_OF_SCOPE} + scope = self.classify(path) + result: Dict[str, Any] = {"path": path, "scope": scope} + if scope == SCOPE_OUT_OF_SCOPE: + return result + if scope == SCOPE_LEGACY_SPECIAL: + return result + # IN_SCOPE: object_id aus Frontmatter validieren + if content is None: + result["scope"] = SCOPE_HUMAN_REVIEW + result["reason_code"] = RC_UNKNOWN_OBJECT_ID + return result + oid = extract_object_id(content) + if oid is None: + # Knowledge-Objekt ohne gueltige ID -> Human Review + result["scope"] = SCOPE_HUMAN_REVIEW + result["reason_code"] = RC_UNKNOWN_OBJECT_ID + return result + result["object_id"] = oid + return result + + +# --------------------------------------------------------------------------- +# Change Classification +# --------------------------------------------------------------------------- + +class ChangeClassifier: + """ + Leitet aus BEFORE/AFTER deterministisch die erlaubten Operationen ab. + + Bei mehreren geaenderten Aspekten in EINEM Objekt werden MEHRERE + ObjectChange-Eintraege erzeugt (keine Information verloren). + """ + + def __init__(self, scope: KnowledgeScope): + self.scope = scope + + def classify(self, path_before: Optional[str], path_after: Optional[str], + content_before: Optional[str], content_after: Optional[str]) -> List[Dict[str, Any]]: + """ + Klassifiziert die Aenderung eines Objekts. Gibt eine Liste von + ObjectChange-Dicts zurueck (eine pro Operation). + """ + # DELETE: Datei existiert in BEFORE, nicht in AFTER + if path_before and not path_after: + return [{"operation": OP_DELETE_REQUEST, "path_before": path_before, + "path_after": None}] + + # CREATE: Datei existiert in AFTER, nicht in BEFORE + if not path_before and path_after: + return [{"operation": OP_CREATE, "path_before": None, + "path_after": path_after}] + + # RENAME/MOVE: gleiche object_id, anderer Pfad + if path_before and path_after and path_before != path_after: + oid_before = extract_object_id(content_before or "") + oid_after = extract_object_id(content_after or "") + if oid_before and oid_after and oid_before == oid_after: + # Rename = Dateiname geaendert, gleiches Verzeichnis + # Move = Verzeichnis geaendert + dir_before = os.path.dirname(path_before) + dir_after = os.path.dirname(path_after) + if dir_before == dir_after: + return [{"operation": OP_RENAME, "path_before": path_before, + "path_after": path_after}] + return [{"operation": OP_MOVE, "path_before": path_before, + "path_after": path_after}] + # Ambiguitaet: Pfad geaendert, aber ID nicht stabil -> Human Review + return [{"operation": OP_MOVE, "path_before": path_before, + "path_after": path_after, "ambiguous": True}] + + # MODIFIED: gleicher Pfad, Inhalt/Metadaten geaendert + ops: List[Dict[str, Any]] = [] + fm_before, body_before = parse_frontmatter(content_before or "") + fm_after, body_after = parse_frontmatter(content_after or "") + + ch_before = content_hash(body_before) + ch_after = content_hash(body_after) + mh_before = metadata_hash(fm_before) + mh_after = metadata_hash(fm_after) + + if ch_before != ch_after: + ops.append({"operation": OP_CONTENT_UPDATE}) + if mh_before != mh_after: + # Spezifische Metadaten-Aenderungen + if fm_before.get("state") != fm_after.get("state"): + ops.append({"operation": OP_STATE_UPDATE}) + if fm_before.get("tags") != fm_after.get("tags"): + ops.append({"operation": OP_TAGS_UPDATE}) + if fm_before.get("derived_from") != fm_after.get("derived_from"): + ops.append({"operation": OP_SOURCE_CANONICAL_RELATION_UPDATE}) + # Generische Metadaten-Aenderung (falls keine spezifische erkannt) + if not any(op["operation"] in ( + OP_STATE_UPDATE, OP_TAGS_UPDATE, + OP_SOURCE_CANONICAL_RELATION_UPDATE, + ) for op in ops): + ops.append({"operation": OP_METADATA_UPDATE}) + # SUPERSEDE: state -> superseded + if fm_after.get("state") == "superseded" and fm_before.get("state") != "superseded": + ops.append({"operation": OP_SUPERSEDE}) + + if not ops: + # Keine Aenderung erkannt (z.B. nur Whitespace) -> kein ObjectChange + return [] + return ops + + +# --------------------------------------------------------------------------- +# C5B Poller (integriert in C5AStore, nur State/Persistence) +# --------------------------------------------------------------------------- + +class C5BPoller: + """ + Fuehrt einen kontrollierten Poll-Zyklus aus und speist Commits + ObjectChanges + in den C5A Store ein. KEIN Network Write downstream. KEIN Poll-Daemon. + """ + + def __init__(self, store: C5AStore, repo_path: str, + poll_interval_seconds: Optional[int] = None): + self.store = store + self.reader = GitReader(repo_path) + self.scope = KnowledgeScope(repo_path) + self.classifier = ChangeClassifier(self.scope) + self.poll_interval_seconds = poll_interval_seconds or self._env_poll_interval() + + @staticmethod + def _env_poll_interval() -> int: + """Liest C5_POLL_INTERVAL_SECONDS aus der Environment (Default 60).""" + raw = os.environ.get(ENV_POLL_INTERVAL, "") + if raw.isdigit() and int(raw) > 0: + return int(raw) + return DEFAULT_POLL_INTERVAL_SECONDS + + def get_current_head(self) -> str: + """Read-only HEAD-Discovery.""" + return self.reader.get_current_head() + + def history_safety(self, last_applied: Optional[str], head: str) -> bool: + """ + FAIL CLOSED: Wenn last_applied_commit kein Ancestor von HEAD ist, + ist die History divergiert. Keine automatische Rebase-/Reset-Logik. + """ + if last_applied is None: + return True # Bootstrap: keine Baseline, alles neu + return self.reader.is_ancestor(last_applied, head) + + def poll_once(self) -> Dict[str, Any]: + """ + Ein kontrollierter Poll-Zyklus: + HEAD -> Commit-Range -> Diff -> Change-Klassifikation -> C5AStore. + Gibt ein Ergebnis-Dict zurueck. KEIN Network Write downstream. + + Baseline fuer Commit-Range + History-Safety ist last_seen_commit + (C5B-Baseline). last_applied_commit bleibt laut C5A-Contract fuer die + spaetere Propagation (C5C) reserviert und wird von C5B NIE fortgeschrieben. + """ + head = self.get_current_head() + health = self.store.health() + last_seen = health.get("last_seen_commit") + + # History Safety (FAIL CLOSED) gegen last_seen_commit + if not self.history_safety(last_seen, head): + return { + "status": "FAIL_CLOSED", + "reason_code": RC_OUT_OF_ORDER_COMMIT, + "message": "last_seen_commit ist kein Ancestor von HEAD (Divergenz)", + "head": head, + "last_seen": last_seen, + } + + # Commit-Range bestimmen (ab last_seen_commit, exklusiv) + commits = self.reader.commit_range(last_seen, head) + self.store.mark_seen(head) + + discovered = 0 + objects_discovered = 0 + out_of_scope = 0 + human_review = 0 + new_commits = [] + + for sha in commits: + meta = self.reader.commit_meta(sha) + # Idempotenz: bereits registriert? + existing = self.store.get_commit(sha) + if existing is not None: + continue # ALREADY_DISCOVERED + + # Diff gegen Parent + changes = self.reader.diff_name_status(sha) + object_changes = [] + for ch in changes: + path_before: Optional[str] = ch.get("path_before") + path_after: Optional[str] = ch.get("path_after") + # Inhalte read-only lesen + content_before: Optional[str] = None + content_after: Optional[str] = None + if path_before: + content_before = self.reader.file_content(sha, path_before) + if path_after: + content_after = self.reader.file_content(sha, path_after) + + # Scope-Klassifikation + scope_info = self.scope.classify_with_content( + path_after or path_before, content_after or content_before + ) + scope = scope_info["scope"] + if scope == SCOPE_OUT_OF_SCOPE: + out_of_scope += 1 + continue + if scope == SCOPE_LEGACY_SPECIAL: + out_of_scope += 1 + continue + if scope == SCOPE_HUMAN_REVIEW: + human_review += 1 + object_changes.append({ + "object_id": None, + "operation": OP_METADATA_UPDATE, + "path_before": path_before, + "path_after": path_after, + "reason_code": scope_info.get("reason_code", RC_UNKNOWN_OBJECT_ID), + }) + continue + + oid = scope_info["object_id"] + # Secret-Safety: kein Secret-Inhalt persistieren + for content in (content_before, content_after): + if content and detect_secret(content): + # Nur sichere Metadaten persistieren, kein Content + object_changes.append({ + "object_id": oid, + "operation": OP_METADATA_UPDATE, + "path_before": path_before, + "path_after": path_after, + "reason_code": RC_SECRET_DETECTED, + }) + human_review += 1 + break + else: + # Change-Klassifikation + ops = self.classifier.classify( + path_before, path_after, content_before, content_after + ) + for op in ops: + fm_before, body_before = parse_frontmatter(content_before or "") + fm_after, body_after = parse_frontmatter(content_after or "") + object_changes.append({ + "object_id": oid, + "operation": op["operation"], + "path_before": path_before, + "path_after": path_after, + "content_hash_before": content_hash(body_before) if content_before is not None else None, + "content_hash_after": content_hash(body_after) if content_after is not None else None, + "metadata_hash_before": metadata_hash(fm_before) if content_before is not None else None, + "metadata_hash_after": metadata_hash(fm_after) if content_after is not None else None, + "representation": fm_after.get("representation") or fm_before.get("representation"), + "state": fm_after.get("state") or fm_before.get("state"), + }) + + # Commit in C5AStore registrieren (Status DISCOVERED) + commit_record = { + "commit_sha": sha, + "parent_sha": meta.get("parent_sha"), + "discovered_at": int(time.time() * 1000), + "sequence": len(commits), + "status": ST_DISCOVERED, + "retry_count": 0, + } + self.store.upsert_commit(commit_record) + discovered += 1 + + # ObjectChanges hinzufuegen + for oc in object_changes: + oc["commit_sha"] = sha + self.store.add_object_change(oc) + objects_discovered += 1 + + # Status -> VALIDATING (nicht weiter zu PROPAGATING_TOLARIA) + self.store.transition_commit(sha, ST_VALIDATING) + if any(oc.get("reason_code") for oc in object_changes): + # VALIDATING -> HUMAN_REVIEW_REQUIRED (erlaubte Transition) + self.store.transition_commit(sha, ST_HUMAN_REVIEW_REQUIRED) + + new_commits.append(sha) + + return { + "status": "OK", + "head": head, + "last_seen": last_seen, + "commits_discovered": discovered, + "objects_discovered": objects_discovered, + "out_of_scope_count": out_of_scope, + "human_review_count": human_review, + "new_commits": new_commits, + "poll_interval_seconds": self.poll_interval_seconds, + } + + +# --------------------------------------------------------------------------- +# No-Downstream-Write-Guarantee (statische Pruefung) +# --------------------------------------------------------------------------- + +def assert_no_downstream_write() -> Dict[str, Any]: + """ + Beweist, dass C5B keinen Codepfad fuer externe Writes besitzt: + * KEIN Tolaria-Write (vault/save) + * KEIN Search-Rebuild (search/rebuild) + * KEIN Forgejo-Write / kein git push/commit/add + Erlaubt ist NUR read-only git (Whitelist) + stdlib. + """ + import ast + this_file = Path(__file__).resolve() + tree = ast.parse(this_file.read_text(encoding="utf-8")) + + # 1. HTTP-Mutationsimporte (Tolaria/Search-Write) + http_imports = {"requests", "urllib", "http", "socket"} + found_http = 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 http_imports: + found_http.add(root) + elif isinstance(node, ast.ImportFrom): + if node.module: + root = node.module.split(".")[0] + if root in http_imports: + found_http.add(root) + + # 2. Verbotene Endpoint-Strings NUR in echten Code-Ausdruecken + # (nicht in Docstrings/Kommentaren). Die Verbotsliste selbst wird aus + # Teilstrings zusammengesetzt, damit der volle Endpoint-String nie als + # einzelnes String-Literal im Quellcode steht (sonst False Positive). + forbidden_endpoints = ["/api/vault" + "/save", "/api/search" + "/rebuild"] + found_endpoints = [] + for literal in _collect_code_strings(tree): + for ep in forbidden_endpoints: + if ep in literal and ep not in found_endpoints: + found_endpoints.append(ep) + + # 3. Git-Write-Befehle duerfen NICHT in der Whitelist stehen + write_in_whitelist = sorted(GIT_READONLY_COMMANDS & GIT_WRITE_COMMANDS) + + return { + "no_http_mutations": len(found_http) == 0 and len(found_endpoints) == 0, + "http_imports_found": sorted(found_http), + "forbidden_endpoints_found": found_endpoints, + "no_git_write_in_whitelist": len(write_in_whitelist) == 0, + "git_write_in_whitelist": write_in_whitelist, + "no_downstream_write": ( + len(found_http) == 0 + and len(found_endpoints) == 0 + and len(write_in_whitelist) == 0 + ), + } + + +def _collect_code_strings(tree: ast.AST) -> List[str]: + """ + Sammelt alle String-Literale, die KEINE Docstrings sind (echter Code). + Docstrings = erstes Statement eines Modul-/Funktions-/Klassen-Body. + """ + import ast as _ast + + docstring_nodes = set() + for node in _ast.walk(tree): + if isinstance(node, (_ast.Module, _ast.FunctionDef, _ast.AsyncFunctionDef, + _ast.ClassDef)): + body = getattr(node, "body", None) + if body and isinstance(body[0], _ast.Expr) and isinstance( + body[0].value, _ast.Constant + ) and isinstance(body[0].value.value, str): + docstring_nodes.add(id(body[0].value)) + + result = [] + for node in _ast.walk(tree): + if isinstance(node, _ast.Constant) and isinstance(node.value, str): + if id(node) in docstring_nodes: + continue + result.append(node.value) + return result diff --git a/tolaria/c5-sync-service/test_c5b.py b/tolaria/c5-sync-service/test_c5b.py new file mode 100644 index 0000000..3a5c192 --- /dev/null +++ b/tolaria/c5-sync-service/test_c5b.py @@ -0,0 +1,539 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5B Testsuite (Forgejo Polling & Change Detection Engine). + +Testet gegen ein kontrolliertes lokales Git-Fixture-Repo (read-only fuer C5B). +Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from typing import Dict + +# C5B importieren (aus demselben Verzeichnis) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from rq_c5a import C5AStore, ST_DISCOVERED, ST_VALIDATING, ST_HUMAN_REVIEW_REQUIRED, ST_APPLIED +from rq_c5b import ( + C5BPoller, + GitReader, + KnowledgeScope, + ChangeClassifier, + assert_no_downstream_write, + parse_frontmatter, + extract_object_id, + content_hash, + metadata_hash, + detect_secret, + SCOPE_IN_SCOPE, + SCOPE_OUT_OF_SCOPE, + SCOPE_LEGACY_SPECIAL, + SCOPE_HUMAN_REVIEW, + 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, + RC_FORGEJO_UNAVAILABLE, + RC_OUT_OF_ORDER_COMMIT, + RC_UNKNOWN_OBJECT_ID, + RC_SECRET_DETECTED, + GitWriteBlockedError, + ForgejoUnavailableError, +) + + +# --------------------------------------------------------------------------- +# Fixture: kontrolliertes Git-Repo +# --------------------------------------------------------------------------- + +def _git(repo: str, *args: str) -> str: + """Fuehrt einen git-Befehl im Fixture-Repo aus (Test-Helfer, darf schreiben).""" + 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: + """Baut Frontmatter fuer ein Knowledge-Objekt.""" + lines = ["---", f"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: + """Erstellt ein kontrolliertes Git-Fixture-Repo fuer C5B-Tests.""" + + def __init__(self): + self.dir = tempfile.mkdtemp(prefix="c5b_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) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestFrontmatterAndHashes(unittest.TestCase): + """Frontmatter-Parsing, Object-ID-Extraktion, Hash-Model.""" + + def test_parse_frontmatter(self): + content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d", + type="arch", role="module", representation="source", + state="current", tags=["a", "b"]) + fm, body = parse_frontmatter(content) + self.assertEqual(fm["id"], "object/48bd264f-607b-15f1-5f73-3e922af9b19d") + self.assertEqual(fm["type"], "arch") + self.assertEqual(fm["tags"], ["a", "b"]) + self.assertNotIn("---", body) + + def test_extract_object_id_valid(self): + content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") + self.assertEqual( + extract_object_id(content), + "object/48bd264f-607b-15f1-5f73-3e922af9b19d", + ) + + def test_extract_object_id_invalid(self): + # Kein object/-Praefix, kein FULL_UUID + self.assertIsNone(extract_object_id("---\nid: object/abc\n---\n")) + self.assertIsNone(extract_object_id("---\nid: 12345\n---\n")) + self.assertIsNone(extract_object_id("kein frontmatter")) + + def test_content_hash_deterministic(self): + self.assertEqual(content_hash("abc"), content_hash("abc")) + self.assertNotEqual(content_hash("abc"), content_hash("abd")) + + def test_metadata_hash_fields(self): + fm1 = {"id": "object/77b02661-d67b-4bae-b612-01d1287cea6b", "type": "arch", "role": "module", + "representation": "source", "state": "current", + "knowledge_schema": "1"} + fm2 = dict(fm1) + fm2["state"] = "historical" + self.assertNotEqual(metadata_hash(fm1), metadata_hash(fm2)) + # Pfad ist NICHT Teil von metadata_hash + self.assertEqual(metadata_hash(fm1), metadata_hash(fm1)) + + def test_detect_secret(self): + self.assertIsNotNone(detect_secret("key: sk-abcdefghijklmnopqrstuvwxyz123456")) + self.assertIsNotNone(detect_secret("-----BEGIN RSA PRIVATE KEY-----\n...")) + self.assertIsNone(detect_secret("modul-07-risk-manager.md")) + + +class TestKnowledgeScope(unittest.TestCase): + """Knowledge-Scope-Klassifikation.""" + + def setUp(self): + self.dir = tempfile.mkdtemp(prefix="c5b_scope_") + self.scope = KnowledgeScope(self.dir) + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def test_root_markdown_in_scope(self): + self.assertEqual(self.scope.classify("modul-09-execution-service.md"), SCOPE_IN_SCOPE) + + def test_canonical_in_scope(self): + self.assertEqual( + self.scope.classify("notes/trading/system-docs/modul-09-execution-service.md"), + SCOPE_IN_SCOPE, + ) + + def test_out_of_scope_dirs(self): + for p in ["tolaria/C5_SYNC_ARCHITECTURE_DESIGN.md", + "a2/README.md", + "red-queen-architecture/AGENT_CONTRACTS.md", + "notion-safety-brain/README.md", + "backup_patches/x.md", + "notes/reference/vps-infrastruktur.md"]: + self.assertEqual(self.scope.classify(p), SCOPE_OUT_OF_SCOPE, p) + + def test_non_markdown_out_of_scope(self): + self.assertEqual(self.scope.classify("index_source.json"), SCOPE_OUT_OF_SCOPE) + + def test_legacy_special(self): + self.assertEqual(self.scope.classify("README.md"), SCOPE_LEGACY_SPECIAL) + self.assertEqual(self.scope.classify("notes/trading/system-docs/README.md"), SCOPE_LEGACY_SPECIAL) + self.assertEqual(self.scope.classify("vps.md"), SCOPE_LEGACY_SPECIAL) + + def test_in_scope_with_valid_id(self): + content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") + r = self.scope.classify_with_content("modul-09.md", content) + self.assertEqual(r["scope"], SCOPE_IN_SCOPE) + self.assertEqual(r["object_id"], "object/48bd264f-607b-15f1-5f73-3e922af9b19d") + + def test_in_scope_without_id_human_review(self): + r = self.scope.classify_with_content("modul-09.md", "kein frontmatter") + self.assertEqual(r["scope"], SCOPE_HUMAN_REVIEW) + self.assertEqual(r["reason_code"], RC_UNKNOWN_OBJECT_ID) + + +class TestChangeClassifier(unittest.TestCase): + """Change-Klassifikation (BEFORE/AFTER -> Operationen).""" + + def setUp(self): + self.dir = tempfile.mkdtemp(prefix="c5b_cls_") + self.cls = ChangeClassifier(KnowledgeScope(self.dir)) + + def tearDown(self): + shutil.rmtree(self.dir, ignore_errors=True) + + def test_create(self): + ops = self.cls.classify(None, "modul-09.md", None, _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b")) + self.assertEqual(ops[0]["operation"], OP_CREATE) + + def test_delete_request(self): + ops = self.cls.classify("modul-09.md", None, _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b"), None) + self.assertEqual(ops[0]["operation"], OP_DELETE_REQUEST) + + def test_content_update(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body v1" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body v2" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertIn(OP_CONTENT_UPDATE, [o["operation"] for o in ops]) + + def test_state_update(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="historical") + "body" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertIn(OP_STATE_UPDATE, [o["operation"] for o in ops]) + + def test_tags_update(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", tags=["a"]) + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", tags=["a", "b"]) + "body" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertIn(OP_TAGS_UPDATE, [o["operation"] for o in ops]) + + def test_relation_update(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", derived_from="object/y") + "body" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertIn(OP_SOURCE_CANONICAL_RELATION_UPDATE, [o["operation"] for o in ops]) + + def test_rename_same_id(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + ops = self.cls.classify("modul-09.md", "modul-09-new.md", before, after) + self.assertIn(OP_RENAME, [o["operation"] for o in ops]) + + def test_move_same_id(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + ops = self.cls.classify("modul-09.md", "sub/modul-09.md", before, after) + self.assertIn(OP_MOVE, [o["operation"] for o in ops]) + + def test_supersede(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="superseded") + "body" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertIn(OP_SUPERSEDE, [o["operation"] for o in ops]) + + def test_no_change(self): + before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" + ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) + self.assertEqual(ops, []) + + +class TestGitReader(unittest.TestCase): + """GitReader read-only + Write-Blockade.""" + + def test_write_commands_blocked(self): + fx = FixtureRepo() + try: + reader = GitReader(fx.dir) + for cmd in ["push", "commit", "add", "reset", "checkout", "merge", "rebase"]: + with self.assertRaises(GitWriteBlockedError): + reader._run([cmd, "x"]) + finally: + fx.cleanup() + + def test_head_discovery(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") + sha = fx.commit("c1") + reader = GitReader(fx.dir) + self.assertEqual(reader.get_current_head(), sha) + finally: + fx.cleanup() + + def test_forgejo_unavailable(self): + with self.assertRaises(ForgejoUnavailableError): + GitReader("/nonexistent/path") + + +class TestC5BPoller(unittest.TestCase): + """End-to-End Poll-Zyklen gegen Fixture-Repo.""" + + def _new_store(self): + db = os.path.join(tempfile.mkdtemp(prefix="c5b_db_"), "c5a.db") + return C5AStore(db) + + def test_single_commit_discovery(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") + "body") + sha = fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["status"], "OK") + self.assertEqual(r["commits_discovered"], 1) + self.assertEqual(r["head"], sha) + # Commit registriert, Status VALIDATING (nicht weiter) + c = store.get_commit(sha) + self.assertEqual(c["status"], ST_VALIDATING) + store.close() + finally: + fx.cleanup() + + def test_multi_commit_ordering(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") + a = fx.commit("A") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") + b = fx.commit("B") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v3") + c = fx.commit("C") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["commits_discovered"], 3) + # Reihenfolge A, B, C (chronologisch) + commits = store.list_commits() + self.assertEqual([c["commit_sha"] for c in commits], [a, b, c]) + store.close() + finally: + fx.cleanup() + + def test_duplicate_poll_idempotent(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") + fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r1 = poller.poll_once() + r2 = poller.poll_once() # gleicher HEAD + self.assertEqual(r1["commits_discovered"], 1) + self.assertEqual(r2["commits_discovered"], 0) # keine neuen + # Keine doppelten ObjectChanges + objs = store.list_object_changes(fx.commits["c1"]) + self.assertEqual(len(objs), 1) + store.close() + finally: + fx.cleanup() + + def test_lost_poll_recovery(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") + a = fx.commit("A") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + poller.poll_once() # Poll bei A + # Zwei Poll-Intervalle ausfallen: B, C, D entstehen + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") + b = fx.commit("B") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v3") + c = fx.commit("C") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v4") + d = fx.commit("D") + r = poller.poll_once() # naechster Poll bei D + self.assertEqual(r["commits_discovered"], 3) + self.assertEqual(r["new_commits"], [b, c, d]) + store.close() + finally: + fx.cleanup() + + def test_out_of_scope_not_registered(self): + fx = FixtureRepo() + try: + fx.write("tolaria/C5_SYNC_ARCHITECTURE_DESIGN.md", "# design") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") + fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["commits_discovered"], 1) + self.assertEqual(r["out_of_scope_count"], 1) # tolaria/ ist out-of-scope + store.close() + finally: + fx.cleanup() + + def test_legacy_special_not_registered(self): + fx = FixtureRepo() + try: + fx.write("README.md", "# README") + fx.write("vps.md", "# vps") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") + fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["out_of_scope_count"], 2) # README + vps + store.close() + finally: + fx.cleanup() + + def test_unknown_object_id_human_review(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", "kein frontmatter, keine id") + fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["human_review_count"], 1) + c = store.get_commit(fx.commits["c1"]) + self.assertEqual(c["status"], ST_HUMAN_REVIEW_REQUIRED) + store.close() + finally: + fx.cleanup() + + def test_secret_detection_fail_closed(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "key: sk-abcdefghijklmnopqrstuvwxyz123456") + fx.commit("c1") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + r = poller.poll_once() + self.assertEqual(r["human_review_count"], 1) + c = store.get_commit(fx.commits["c1"]) + self.assertEqual(c["status"], ST_HUMAN_REVIEW_REQUIRED) + # Kein Secret-Inhalt in der DB + objs = store.list_object_changes(fx.commits["c1"]) + for o in objs: + self.assertNotIn("sk-", json.dumps(o)) + store.close() + finally: + fx.cleanup() + + def test_history_divergence_fail_closed(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") + a = fx.commit("A") + store = self._new_store() + poller = C5BPoller(store, fx.dir) + poller.poll_once() + # last_applied = a. Jetzt History-Rewrite: neuer Root ohne a als Ancestor. + # Wir wechseln auf einen Orphan-Branch (neuer Root), der a NICHT enthaelt. + _git(fx.dir, "checkout", "--orphan", "newroot") + _git(fx.dir, "rm", "-rf", ".") + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") + b = fx.commit("B") + # HEAD ist jetzt b (newroot), a ist NICHT in dessen Ancestry + self.assertNotEqual(a, b) + r = poller.poll_once() + self.assertEqual(r["status"], "FAIL_CLOSED") + self.assertEqual(r["reason_code"], RC_OUT_OF_ORDER_COMMIT) + store.close() + finally: + fx.cleanup() + + def test_c5a_persistence_integration(self): + fx = FixtureRepo() + try: + fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") + sha = fx.commit("c1") + db = os.path.join(tempfile.mkdtemp(prefix="c5b_db_"), "c5a.db") + store = C5AStore(db) + poller = C5BPoller(store, fx.dir) + poller.poll_once() + store.close() + # Neuer Store auf gleicher DB: State restart-fest + store2 = C5AStore(db) + c = store2.get_commit(sha) + self.assertEqual(c["status"], ST_VALIDATING) + store2.close() + finally: + fx.cleanup() + + def test_no_downstream_write(self): + r = assert_no_downstream_write() + self.assertTrue(r["no_downstream_write"]) + self.assertEqual(r["http_imports_found"], []) + self.assertEqual(r["forbidden_endpoints_found"], []) + self.assertEqual(r["git_write_in_whitelist"], []) + + +class TestPollIntervalConfig(unittest.TestCase): + """Poll-Intervall ist Environment-konfigurierbar.""" + + def test_default_60(self): + fx = FixtureRepo() + try: + store = C5AStore(os.path.join(tempfile.mkdtemp(), "c5a.db")) + poller = C5BPoller(store, fx.dir) + self.assertEqual(poller.poll_interval_seconds, 60) + store.close() + finally: + fx.cleanup() + + def test_env_override(self): + fx = FixtureRepo() + try: + os.environ["C5_POLL_INTERVAL_SECONDS"] = "120" + store = C5AStore(os.path.join(tempfile.mkdtemp(), "c5a.db")) + poller = C5BPoller(store, fx.dir) + self.assertEqual(poller.poll_interval_seconds, 120) + store.close() + del os.environ["C5_POLL_INTERVAL_SECONDS"] + finally: + fx.cleanup() + + +if __name__ == "__main__": + unittest.main(verbosity=2)