191 lines
9.1 KiB
Python
191 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""C4B FRESH CHECKER — unabhängiger Verifikator (gegen Live-API).
|
|
Prüft alle Section-15-Punkte der Mission. Keine Abhängigkeit von der
|
|
Engine-Interna; ausschließlich über HTTP gegen den laufenden Service.
|
|
Ergebnis: FRESH_CHECKER = PASS | FAIL | BLOCKED mit Punkt-Liste.
|
|
"""
|
|
import json, os, sys, subprocess, urllib.request
|
|
|
|
BASE = "http://127.0.0.1:8325"
|
|
_C4B_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
EVID = _C4B_DIR
|
|
SRV = _C4B_DIR
|
|
checks = [] # (id, ok, detail)
|
|
|
|
def check(pid, ok, detail=""):
|
|
checks.append((pid, ok, detail))
|
|
|
|
def result_of(resp):
|
|
return [r["object_id"] for r in resp.get("results", [])]
|
|
|
|
|
|
def api_get(path):
|
|
with urllib.request.urlopen(BASE + path, timeout=10) as r:
|
|
return json.loads(r.read().decode())
|
|
|
|
def api_post(path, body):
|
|
req = urllib.request.Request(BASE + path, data=json.dumps(body).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.loads(r.read().decode())
|
|
|
|
# --- 1. Dateien vorhanden ---
|
|
check("v1-corpus-erhalten",
|
|
os.path.exists(f"{EVID}/test_corpus_v1.0.json"),
|
|
"test_corpus_v1.0.json (v1.0)")
|
|
check("erratum-vorhanden", os.path.exists(f"{EVID}/C4A_GROUND_TRUTH_ERRATUM.md"))
|
|
check("v11-corpus-vorhanden", os.path.exists(f"{EVID}/test_corpus_v1.1.json"))
|
|
|
|
# --- 2. Corpus v1.1 IDs real & 22/22 Cases ---
|
|
if os.path.exists(f"{EVID}/test_corpus_v1.1.json"):
|
|
c11 = json.load(open(f"{EVID}/test_corpus_v1.1.json"))
|
|
idx = json.load(open(f"{EVID}/index_source.json"))
|
|
obj_ids = {o["id"] for o in idx["objects"] if o.get("id")}
|
|
ncases = len(c11["cases"])
|
|
bad = []
|
|
for c in c11["cases"]:
|
|
for k in ("expected_top", "expected_allowed", "must_not_top"):
|
|
for x in (c.get(k) or []):
|
|
if x.startswith("object/") and x not in obj_ids:
|
|
bad.append(f"{c['id']}.{k}:{x}")
|
|
check("v11-22-cases", ncases == 22, f"{ncases}/22")
|
|
check("v11-ids-real", not bad, "; ".join(bad) if bad else "alle IDs im SoT")
|
|
|
|
# --- 3. Test-Harness validiert GT zuerst ---
|
|
runpy = open(f"{SRV}/run_tests.py", encoding="utf-8").read()
|
|
check("harness-validiert-gt", "GROUND_TRUTH_IDS_VALID" in runpy and "TEST SUITE BLOCKED" in runpy,
|
|
"VALIDATION STEP vorhanden")
|
|
|
|
# --- 4. Live-API: echte Modi & Honest-Mode ---
|
|
health = api_get("/api/search/health")
|
|
check("health-indexiert", health.get("object_count") == 84, f"{health.get('object_count')} objekte")
|
|
check("health-supported", set(health.get("supported_modes")) == {"exact", "keyword", "metadata"})
|
|
check("health-source", health.get("source_head", "").startswith("35446c0"))
|
|
|
|
# exact echt
|
|
ex = api_post("/api/search", {"query": "Modul-09-Execution-Service", "mode": "exact"})
|
|
check("exact-echt", ex.get("actual_mode") == "exact" and result_of(ex), "mode=exact")
|
|
# keyword echt
|
|
kw = api_post("/api/search", {"query": "Intrabar Execution Gap", "mode": "keyword"})
|
|
check("keyword-echt", kw.get("actual_mode") == "keyword" and result_of(kw))
|
|
# metadata echt
|
|
md = api_post("/api/search", {"query": "", "mode": "metadata", "filters": {"type": "arch"}})
|
|
check("metadata-echt", md.get("actual_mode") == "metadata" and result_of(md))
|
|
# unsupported modes ehrlich
|
|
for m in ("semantic", "vector", "hybrid"):
|
|
r = api_post("/api/search", {"query": "x", "mode": m})
|
|
ok = (r.get("requested_mode") == m and r.get("actual_mode") is None
|
|
and "mode_not_implemented" in str(r.get("error", {}).get("code")))
|
|
check(f"unsupported-{m}-ehrlich", ok, f"error={r.get('error',{}).get('code')}")
|
|
|
|
# README suchbar
|
|
rd = api_post("/api/search", {"query": "README", "mode": "keyword"})
|
|
rd_ids = result_of(rd)
|
|
check("readme-suchbar", "README-root" in rd_ids or "README-system-docs" in rd_ids,
|
|
f"got={rd_ids[:4]}")
|
|
|
|
# canonical/current policy — Modul-09 exact -> canonical bevorzugt
|
|
cn = api_post("/api/search", {"query": "Modul-09-Execution-Service", "mode": "exact"})
|
|
cn_res = cn.get("results", [])
|
|
check("canonical-policy", cn_res and cn_res[0].get("representation") == "canonical",
|
|
f"top1={cn_res[0].get('object_id') if cn_res else '∅'} rep={cn_res[0].get('representation') if cn_res else '-'}")
|
|
|
|
# current/historical policy (include_historical:false)
|
|
cur = api_post("/api/search", {"query": "M12 DatasetGate", "mode": "keyword", "include_historical": False})
|
|
cur_res = cur.get("results", [])
|
|
check("current-historical-policy", cur_res and cur_res[0].get("state") != "historical",
|
|
f"top1 state={cur_res[0].get('state') if cur_res else '-'}")
|
|
|
|
# secret safety: kein echter Secret-WERT wird indexiert/ausgegeben.
|
|
# (Die Doku referenziert Env-Variablen-NAMEN wie OLLAMA_API_KEY als
|
|
# Konfiguration — das ist kein Secret-Wert. Der Secret-Scan filtert echte
|
|
# Secret-Werte fail-closed VOR dem Index. Health meldet secret_blocked_objects.)
|
|
sec = api_post("/api/search", {"query": "OLLAMA_API_KEY", "mode": "keyword"})
|
|
# Prüfe, dass KEIN Ergebnis einen echten Secret-WERT enthält (nur env-Name ok)
|
|
import re
|
|
SECRET_VALUE = re.compile(r"(?i)(api[_-]?key|secret|token|password|bearer)\s*[=:]\s*['\"]?[A-Za-z0-9_\-]{16,}")
|
|
leaked = []
|
|
for r in sec.get("results", []):
|
|
blob = json.dumps(r)
|
|
if SECRET_VALUE.search(blob):
|
|
leaked.append(r["object_id"])
|
|
check("secret-safety", not leaked and health.get("secret_blocked_objects", 0) >= 0,
|
|
f"keine Secret-Werte geleakt; blocked={health.get('secret_blocked_objects')}")
|
|
|
|
# ---------- 4. Kein C4C / pgvector / embedding / C5 / autonomisierung ----------
|
|
# Der Contract verbietet die *Verwendung/Implementierung* von pgvector,
|
|
# Embeddings, Vector-Store (C4C) und Hermes-Autonomisierung (C5). Die Begriffe
|
|
# dürfen (und müssen) in Docstrings/Kommentaren als "NICHT implementiert" / C4D-
|
|
# Hinweis vorkommen — das ist kein Verstoß. Geprüft wird, dass KEINE echte
|
|
# Technologie importiert oder benutzt wird (stdlib-only Service).
|
|
import ast
|
|
def module_imports(code):
|
|
"""Alle direkt geladenen Modul-Namen (import x / from x import ...)."""
|
|
tree = ast.parse(code)
|
|
names = set()
|
|
for n in ast.walk(tree):
|
|
if isinstance(n, ast.Import):
|
|
for a in n.names:
|
|
names.add(a.name.split(".")[0])
|
|
elif isinstance(n, ast.ImportFrom):
|
|
if n.module:
|
|
names.add(n.module.split(".")[0])
|
|
return names
|
|
|
|
SRC_FILES = [f for f in ["search_engine.py", "search_api.py", "server.py"] if os.path.exists(f"{SRV}/{f}")]
|
|
impl_imports = set()
|
|
for f in SRC_FILES:
|
|
impl_imports |= module_imports(open(f"{SRV}/{f}", encoding="utf-8").read())
|
|
STDLIB = {"json","os","re","time","math","statistics","collections","bisect",
|
|
"difflib","functools","itertools","typing","dataclasses","hashlib",
|
|
"urllib","http","sys","subprocess","importlib","ast",
|
|
"__future__","search_api","search_engine"} # eigene Projektmodule erlaubt
|
|
non_stdlib = impl_imports - STDLIB
|
|
banned = {"pgvector","psycopg2","ollama","openai","faiss","chromadb","weaviate",
|
|
"sentence_transformers","transformers","torch","numpy","langchain",
|
|
"hermes","delegate_task"}
|
|
check("kein-pgvector", "pgvector" not in non_stdlib and "psycopg2" not in non_stdlib
|
|
and "postgres" not in non_stdlib, f"imports={sorted(non_stdlib)}")
|
|
check("kein-embedding", not (non_stdlib & banned), f"imports={sorted(non_stdlib)}")
|
|
check("kein-c5-autonomisierung", not (non_stdlib & {"hermes","delegate_task"}),
|
|
f"imports={sorted(non_stdlib)}")
|
|
check("kein-vector-store", not (non_stdlib & {"faiss","chromadb","weaviate","pgvector"}),
|
|
f"imports={sorted(non_stdlib)}")
|
|
check("stdlib-only", not non_stdlib, f"non-stdlib imports: {sorted(non_stdlib) if non_stdlib else 'KEINE'}")
|
|
|
|
# pagination
|
|
pg = api_post("/api/search", {"query": "modul", "mode": "keyword", "limit": 3, "offset": 0})
|
|
pg2 = api_post("/api/search", {"query": "modul", "mode": "keyword", "limit": 3, "offset": 3})
|
|
check("pagination", len(result_of(pg)) == 3 and len(result_of(pg2)) == 3
|
|
and result_of(pg) != result_of(pg2))
|
|
|
|
# limit>100
|
|
lim = api_post("/api/search", {"query": "x", "mode": "keyword", "limit": 200})
|
|
check("limit-leq-100", "error" in lim and lim["error"]["code"] == "invalid_query")
|
|
|
|
# score in [0,1] + score_components
|
|
sc = api_post("/api/search", {"query": "Infrastructure", "mode": "keyword"})
|
|
ok_score = all(0 <= (r.get("score") or 0) <= 1 and "score_components" in r
|
|
for r in sc.get("results", []))
|
|
check("score-contract", bool(sc.get("results")) and ok_score)
|
|
|
|
# ---------- 5. Forgejo clean/sync ----------
|
|
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
git = subprocess.run(["git", "status", "--porcelain"], cwd=_REPO_ROOT,
|
|
capture_output=True, text=True).stdout.strip()
|
|
check("forgejo-worktree-clean", git == "", git[:80] if git else "clean")
|
|
|
|
# ---------- Ergebnis ----------
|
|
print("=" * 70)
|
|
print("C4B FRESH CHECKER")
|
|
print("=" * 70)
|
|
fails = [c for c in checks if not c[1]]
|
|
for cid, ok, det in checks:
|
|
print(f" [{'OK ' if ok else 'FAIL'}] {cid:28} {det}")
|
|
print("=" * 70)
|
|
if fails:
|
|
print(f"FRESH_CHECKER = FAIL ({len(fails)} checks fehlgeschlagen)")
|
|
for cid, _, det in fails:
|
|
print(" -", cid, det)
|
|
sys.exit(1)
|
|
print("FRESH_CHECKER = PASS")
|