trading-system-docs/tolaria/c5-sync-service/rq_c5d.py

512 lines
21 KiB
Python

#!/usr/bin/env python3
"""
Red Queen — C5D: SEARCH INTEGRATION + COMMIT COMPLETION v1 (deterministische, inaktive Library).
C5D implementiert ausschliesslich:
UPDATING_SEARCH -> SEARCH FULL REBUILD -> VERIFYING_SEARCH
-> SEARCH HEALTH/INTEGRITY READ-BACK -> APPLIED -> last_applied_commit fortschreiben
VERBINDLICHE REIHENFOLGE (ORDERING INVARIANT):
Forgejo Commit -> Tolaria propagiert (C5C) -> Tolaria Read-Back DRIFT=0
-> Search Full Rebuild -> Search Health/Integrity PASS -> Commit APPLIED
-> last_applied_commit fortschreiben
INVARIANTE: Search darf NIEMALS vor erfolgreicher Tolaria-Verifikation aktualisiert
werden. APPLIED und last_applied_commit duerfen ausschliesslich gesetzt werden,
wenn Tolaria UND Search vollstaendig erfolgreich und verifiziert sind.
C5D startet NUR aus ST_UPDATING_SEARCH (Tolaria bereits verifiziert) oder
ST_RETRY_PENDING (Search-Retry-Replay). C5D ruft NIE Tolaria auf -> kein
Tolaria-Doppel-Write beim Search-Retry.
ARCHITEKTURREGEL (verbindlich):
* Forgejo bleibt MASTER / Source of Truth.
* Tolaria ist ausschliesslich DERIVED.
* C5D darf: Forgejo lesen, Search kontrolliert rebuilden (mit Token).
* C5D darf NIEMALS: Tolaria -> Forgejo schreiben, Forgejo veraendern,
Tolaria schreiben, IDs neu vergeben, Knowledge Content umformulieren.
NO-TOLARIA-WRITE-GUARANTEE: C5D ruft NICHT /api/vault/save auf.
NO-MASTER-WRITE-GUARANTEE: C5D fuehrt KEIN git push / Forgejo-Write aus.
NO-PRODUCTION-ACTIVATION: C5D fuehrt KEINE produktive Propagation / Search-Rebuild
aus (nur Tests gegen Fake/Mock + read-only Dry-Run).
FAIL-CLOSED: Bei Search-Ausfall, Timeout, Auth-Fehler, ungueltiger Antwort,
Rebuild-Fehler oder fehlerhafter Integrity -> Commit NICHT APPLIED,
last_applied_commit NICHT veraendert, Fehler klassifiziert, Retry/Human-Gate
gemaess C5A-Contract.
"""
from __future__ import annotations
import json
import os
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
# C5A / C5C wiederverwenden (keine konkurrierende State Machine)
from rq_c5a import (
C5AStore,
ST_UPDATING_SEARCH,
ST_VERIFYING_SEARCH,
ST_APPLIED,
ST_RETRY_PENDING,
ST_DEAD,
ST_HUMAN_REVIEW_REQUIRED,
IDEM_ALREADY_APPLIED,
IDEM_RETRY_SAFE,
RC_SEARCH_REBUILD_FAILURE,
RC_AUTH_FAILURE,
RC_UNKNOWN_OBJECT_ID,
RC_UNEXPECTED_TOLARIA_DRIFT,
DEFAULT_MAX_RETRIES,
DEFAULT_BACKOFF_SECONDS,
)
from rq_c5c import C5CError
# ---------------------------------------------------------------------------
# Konstanten
# ---------------------------------------------------------------------------
# Search-API-Basis (Default; per Env C5_SEARCH_BASE ueberschreibbar)
DEFAULT_SEARCH_BASE = "http://127.0.0.1:8325/api/search"
ENV_SEARCH_BASE = "C5_SEARCH_BASE"
ENV_SEARCH_TOKEN = "TOLARIA_SEARCH_REBUILD_TOKEN"
# Search-Rebuild-Ergebnis
REBUILD_OK = "ok"
# Integrity-Verifikations-Ergebnisse
INTEGRITY_OK = "INTEGRITY_OK"
INTEGRITY_FAIL = "INTEGRITY_FAIL"
class SearchError(C5CError):
"""Basis-Fehler fuer C5D Search-Integration."""
class SearchUnavailableError(SearchError):
"""Search nicht erreichbar / Timeout / transienter HTTP-Fehler (retrybar)."""
class SearchAuthError(SearchError):
"""Search-Rebuild Auth-Fehler (nicht retrybar, Human Gate)."""
class SearchRebuildError(SearchError):
"""Search-Rebuild fehlgeschlagen (nicht retrybar, Human Gate)."""
class SearchIntegrityError(SearchError):
"""Search Health/Integrity nicht PASS (nicht retrybar, Human Gate)."""
class SearchMalformedResponseError(SearchError):
"""Search-Antwort ungueltig / malformed (nicht retrybar, Human Gate)."""
# ---------------------------------------------------------------------------
# Search Client (isoliert: rebuild / health)
# ---------------------------------------------------------------------------
class SearchClient:
"""
Isolierter Search-Client. Nur C5D-Writer darf rebuild() verwenden.
rebuild() — POST /api/search/rebuild (Bearer-Token, kontrolliert)
health() — GET /api/search/health (Read-Back, oeffentlich)
Keine generische Agent-Write-Funktion. Kein Tolaria-Write.
"""
def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None,
timeout: float = 15.0):
self.base_url = (base_url or os.environ.get(ENV_SEARCH_BASE)
or DEFAULT_SEARCH_BASE).rstrip("/")
self.token = token if token is not None else os.environ.get(ENV_SEARCH_TOKEN, "")
self.timeout = timeout
# -- HTTP-Helfer --------------------------------------------------------
def _post(self, endpoint: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
data = json.dumps(payload or {}).encode("utf-8")
headers = {"Content-Type": "application/json"}
if self.token:
headers["Authorization"] = "Bearer " + self.token
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else {}
except urllib.error.HTTPError as e:
if e.code in (401, 403):
raise SearchAuthError(
f"Search Auth-Fehler HTTP {e.code} auf {endpoint}",
RC_AUTH_FAILURE)
if e.code >= 500:
raise SearchUnavailableError(
f"Search HTTP {e.code} auf {endpoint}", RC_SEARCH_REBUILD_FAILURE)
raise SearchRebuildError(
f"Search HTTP {e.code} auf {endpoint}: {e.read().decode('utf-8', 'replace')[:200]}",
RC_SEARCH_REBUILD_FAILURE)
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise SearchUnavailableError(
f"Search nicht erreichbar ({endpoint}): {e}", RC_SEARCH_REBUILD_FAILURE)
def _get(self, endpoint: str) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
req = urllib.request.Request(url, method="GET")
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else {}
except urllib.error.HTTPError as e:
if e.code >= 500:
raise SearchUnavailableError(
f"Search HTTP {e.code} auf {endpoint}", RC_SEARCH_REBUILD_FAILURE)
raise SearchRebuildError(
f"Search HTTP {e.code} auf {endpoint}: {e.read().decode('utf-8', 'replace')[:200]}",
RC_SEARCH_REBUILD_FAILURE)
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise SearchUnavailableError(
f"Search nicht erreichbar ({endpoint}): {e}", RC_SEARCH_REBUILD_FAILURE)
# -- Rebuild (NUR C5D-Writer) -------------------------------------------
def rebuild(self) -> Dict[str, Any]:
"""Fuehrt einen Search Full Rebuild aus (POST /api/search/rebuild)."""
resp = self._post("rebuild")
if not isinstance(resp, dict):
raise SearchMalformedResponseError(
f"Search-Rebuild-Antwort ungueltig: {type(resp).__name__}",
RC_SEARCH_REBUILD_FAILURE)
if resp.get("status") != REBUILD_OK:
raise SearchRebuildError(
f"Search-Rebuild fehlgeschlagen: {resp.get('error', resp)}",
RC_SEARCH_REBUILD_FAILURE)
return resp
# -- Health (Read-Back) -------------------------------------------------
def health(self) -> Dict[str, Any]:
"""Liest Search Health/Integrity (GET /api/search/health)."""
resp = self._get("health")
if not isinstance(resp, dict):
raise SearchMalformedResponseError(
f"Search-Health-Antwort ungueltig: {type(resp).__name__}",
RC_SEARCH_REBUILD_FAILURE)
return resp
# ---------------------------------------------------------------------------
# Integrity-Verifikation (Read-Back, §C5D)
# ---------------------------------------------------------------------------
def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]:
"""
Verifiziert Search Health/Integrity per Read-Back.
Ein HTTP 200 allein reicht NICHT als Erfolg. Health muss PASS sein:
* integrity_ok == True
* index_built == True
* object_count > 0 (unerwarteter/falscher Indexzustand -> FAIL)
* failed_objects leer (optional, aber fail-closed wenn nicht leer)
Rueckgabe: {"ok": bool, "checks": {...}, "reason": optional}
"""
if not isinstance(health, dict):
raise SearchMalformedResponseError(
f"Search-Health-Antwort ungueltig: {type(health).__name__}",
RC_SEARCH_REBUILD_FAILURE)
checks = {
"integrity_ok": bool(health.get("integrity_ok")),
"index_built": bool(health.get("index_built")),
"object_count_positive": int(health.get("object_count", 0)) > 0,
"failed_objects_empty": not bool(health.get("failed_objects")),
}
ok = all(checks.values())
reason = None
if not ok:
failed = [k for k, v in checks.items() if not v]
reason = "Search-Health nicht PASS: " + ", ".join(failed)
return {"ok": ok, "checks": checks, "reason": reason}
# ---------------------------------------------------------------------------
# C5D Engine (Search Integration + Commit Completion)
# ---------------------------------------------------------------------------
class C5DEngine:
"""
Vervollstaendigt einen Commit nach erfolgreicher Tolaria-Propagation (C5C).
Startet NUR aus ST_UPDATING_SEARCH (Tolaria verifiziert, READY_FOR_SEARCH)
oder ST_RETRY_PENDING (Search-Retry-Replay). Fuehrt Search-Rebuild aus,
verifiziert Health/Integrity per Read-Back, und setzt bei vollem PASS
ST_APPLIED + last_applied_commit.
C5D ruft NIE Tolaria auf -> kein Tolaria-Doppel-Write beim Search-Retry.
"""
def __init__(
self,
store: C5AStore,
search: Optional[SearchClient] = None,
max_retries: int = DEFAULT_MAX_RETRIES,
backoff_seconds: Optional[List[int]] = None,
):
self.store = store
self.search = search or SearchClient()
self.max_retries = max_retries
self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS
# -- Idempotenz ---------------------------------------------------------
def _commit_already_applied(self, commit_sha: str) -> bool:
"""Commit bereits vollstaendig APPLIED -> kein Downstream-Write."""
return self.store.commit_status(commit_sha) == ST_APPLIED
# -- Retry-Klassifikation (§17) -----------------------------------------
def _is_retryable(self, err: SearchError) -> bool:
"""Nur technische, retrybare Fehler nutzen das Retry-Modell."""
return isinstance(err, SearchUnavailableError)
def _handle_failure(self, commit_sha: str, err: SearchError) -> Dict[str, Any]:
"""Behandelt einen Search-Fehler: Retry, DEAD oder Human Gate."""
cur = self.store.commit_status(commit_sha)
if self._is_retryable(err):
retry = self.store.increment_retry(commit_sha)
# Sicherstellen, dass wir in RETRY_PENDING sind (erlaubt von
# UPDATING_SEARCH und VERIFYING_SEARCH), bevor wir ggf. zu DEAD wechseln.
if cur in (ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH):
self.store.transition_commit(commit_sha, ST_RETRY_PENDING)
if retry >= self.max_retries:
self.store.set_commit_error(commit_sha, err.reason_code or RC_SEARCH_REBUILD_FAILURE, err.message)
self.store.transition_commit(commit_sha, ST_DEAD)
return {"commit_sha": commit_sha, "status": ST_DEAD,
"reason_code": err.reason_code, "retry_count": retry}
self.store.set_commit_error(commit_sha, err.reason_code or RC_SEARCH_REBUILD_FAILURE, err.message)
return {"commit_sha": commit_sha, "status": ST_RETRY_PENDING,
"reason_code": err.reason_code, "retry_count": retry}
# Nicht-retrybar -> Human Gate / Fail Closed
rc = err.reason_code or RC_SEARCH_REBUILD_FAILURE
self.store.set_commit_error(commit_sha, rc, err.message)
self.store.transition_commit(commit_sha, ST_HUMAN_REVIEW_REQUIRED)
return {"commit_sha": commit_sha, "status": ST_HUMAN_REVIEW_REQUIRED,
"reason_code": rc}
# -- Commit anwenden ----------------------------------------------------
def apply_commit(self, commit_sha: str) -> Dict[str, Any]:
"""
Vervollstaendigt einen Commit: Search-Rebuild -> Health/Integrity -> APPLIED.
VERBINDLICHE REIHENFOLGE:
UPDATING_SEARCH -> (Search Rebuild) -> VERIFYING_SEARCH
-> (Health/Integrity Read-Back) -> APPLIED -> last_applied_commit
FAIL-CLOSED: Bei jedem Search-Fehler bleibt der Commit NICHT APPLIED und
last_applied_commit unveraendert.
"""
commit = self.store.get_commit(commit_sha)
if commit is None:
return {"commit_sha": commit_sha, "status": ST_HUMAN_REVIEW_REQUIRED,
"reason_code": RC_UNKNOWN_OBJECT_ID}
cur = commit.get("status")
# Idempotenz: bereits APPLIED -> kein Downstream-Write
if cur == ST_APPLIED:
return {"commit_sha": commit_sha, "status": ST_APPLIED,
"idempotency": IDEM_ALREADY_APPLIED}
# Nur aus UPDATING_SEARCH oder RETRY_PENDING (Search-Retry-Replay) starten
if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING):
return {"commit_sha": commit_sha, "status": cur,
"message": "Commit nicht im Search-Schritt (Tolaria nicht verifiziert)"}
# Search-Retry-Replay: RETRY_PENDING -> UPDATING_SEARCH (kein Tolaria-Write)
if cur == ST_RETRY_PENDING:
self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH)
# 1) Search Full Rebuild (UPDATING_SEARCH)
try:
rebuild = self.search.rebuild()
except SearchError as e:
return self._handle_failure(commit_sha, e)
# Defense in depth: HTTP 200 allein reicht NICHT. Die Engine validiert
# den Rebuild-Response selbst (status == "ok"), unabhaengig vom Client.
if not isinstance(rebuild, dict) or rebuild.get("status") != REBUILD_OK:
return self._handle_failure(
commit_sha,
SearchRebuildError(
f"Search-Rebuild-Response ungueltig: {rebuild!r}",
RC_SEARCH_REBUILD_FAILURE))
# 2) Rebuild OK -> VERIFYING_SEARCH
self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH)
# 3) Health/Integrity Read-Back (VERIFYING_SEARCH)
try:
health = self.search.health()
integrity = verify_integrity(health)
except SearchError as e:
return self._handle_failure(commit_sha, e)
if not integrity["ok"]:
return self._handle_failure(
commit_sha,
SearchIntegrityError(integrity["reason"] or "Search-Integrity nicht PASS",
RC_SEARCH_REBUILD_FAILURE))
# 4) Vollstaendiger PASS -> APPLIED + last_applied_commit fortschreiben
self.store.transition_commit(commit_sha, ST_APPLIED)
self.store.mark_applied(commit_sha)
return {"commit_sha": commit_sha, "status": ST_APPLIED,
"idempotency": IDEM_RETRY_SAFE, "integrity": integrity,
"rebuild": rebuild}
# ---------------------------------------------------------------------------
# C5D Dry-Run (read-only) — zeigt, welche Commits fuer Search bereit sind
# ---------------------------------------------------------------------------
class C5DDryRun:
"""
Read-only Dry-Run: identifiziert Commits in UPDATING_SEARCH (READY_FOR_SEARCH)
und RETRY_PENDING (Search-Retry-Replay). Fuehrt KEINEN Search-Rebuild aus.
"""
def __init__(self, store: C5AStore):
self.store = store
def plan(self) -> Dict[str, Any]:
"""Erzeugt den C5D-Plan (read-only, KEIN Search-Rebuild)."""
ready = []
retry = []
for c in self.store.list_commits():
status = c.get("status")
if status == ST_UPDATING_SEARCH:
ready.append(c.get("commit_sha"))
elif status == ST_RETRY_PENDING:
retry.append(c.get("commit_sha"))
return {
"ready_for_search": ready,
"search_retry_pending": retry,
"note": "read-only Dry-Run: KEIN produktiver Search-Rebuild",
}
# ---------------------------------------------------------------------------
# Guarantees (statische Pruefung)
# ---------------------------------------------------------------------------
def _collect_docstrings(tree: Any) -> set:
"""Sammelt alle Docstring-Strings (Modul-/Funktions-/Klassen-Docstrings)."""
import ast
docstrings = set()
for node in ast.walk(tree):
if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef,
ast.ClassDef)):
body = node.body
if body and isinstance(body[0], ast.Expr):
val = body[0].value
if isinstance(val, ast.Constant) and isinstance(val.value, str):
docstrings.add(val.value)
return docstrings
def assert_no_tolaria_write() -> Dict[str, Any]:
"""
Beweist statisch, dass C5D NICHT nach Tolaria schreibt (kein /api/vault/save).
Prueft, dass keine Tolaria-Write-Endpoint-Strings in echten Code-Ausdruecken
(nicht Docstrings/Kommentaren) vorkommen.
"""
import ast
this_file = Path(__file__).resolve()
tree = ast.parse(this_file.read_text(encoding="utf-8"))
docstrings = _collect_docstrings(tree)
# Fragmentierte banned-Strings (Selbstreferenz vermeiden)
banned = ["/api/" + "vault/" + "save", "vault/" + "save", "api/" + "vault"]
found = []
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if node.value in docstrings:
continue
for b in banned:
if b in node.value:
found.append(node.value)
return {
"no_tolaria_write": len(found) == 0,
"tolaria_write_endpoints_found": sorted(set(found)),
}
def assert_no_master_write() -> Dict[str, Any]:
"""
Beweist statisch, dass C5D KEIN git push / Forgejo-Write ausfuehrt.
"""
import ast
this_file = Path(__file__).resolve()
tree = ast.parse(this_file.read_text(encoding="utf-8"))
docstrings = _collect_docstrings(tree)
banned_imports = {"subprocess", "os.system", "git"}
found_imports = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
root = alias.name.split(".")[0]
if root in banned_imports:
found_imports.add(root)
elif isinstance(node, ast.ImportFrom):
if node.module:
root = node.module.split(".")[0]
if root in banned_imports:
found_imports.add(root)
banned_calls = ["git " + "pus" + "h", "git " + "comm" + "it", "git " + "ad" + "d", "pus" + "h"]
found_calls = []
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if node.value in docstrings:
continue
for b in banned_calls:
if b in node.value:
found_calls.append(node.value)
return {
"no_master_write": len(found_imports) == 0 and len(found_calls) == 0,
"banned_imports_found": sorted(found_imports),
"write_commands_found": sorted(set(found_calls)),
}
def assert_no_production_activation() -> Dict[str, Any]:
"""
Beweist statisch, dass C5D KEINE produktive Aktivierung enthaelt
(kein Daemon/Deploy/Canary/Thread/Cron/Self-Schedule).
"""
import ast
this_file = Path(__file__).resolve()
tree = ast.parse(this_file.read_text(encoding="utf-8"))
docstrings = _collect_docstrings(tree)
banned = ["daem" + "on", "serve_" + "forever", "Threading" + "HTTPServer",
"cro" + "n", "sched" + "ule", "depl" + "oy", "can" + "ary",
"threading." + "Thread", "while " + "True"]
found = []
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if node.value in docstrings:
continue
for b in banned:
if b in node.value:
found.append(node.value)
return {
"no_production_activation": len(found) == 0,
"activation_patterns_found": sorted(set(found)),
}