File size: 4,444 Bytes
eff511c 663f74a 0a6fd56 eff511c 663f74a c4fd5dd 663f74a c4fd5dd 0a6fd56 c4fd5dd 0a6fd56 c4fd5dd 0a6fd56 c4fd5dd 0a6fd56 c4fd5dd 663f74a 0a6fd56 663f74a c4fd5dd 0a6fd56 eff511c 663f74a 0a6fd56 eff511c 663f74a eff511c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | from __future__ import annotations
import json
from clause_retrieval import init_clause_retrieval, load_ontology
from config import (
CORPUS_MCKF_ONTOLOGY_PATH,
MCKF_ONTOLOGY_PATH,
MCKF_VALIDATION_REPORT_PATH,
RUNTIME_BUILD_MANIFEST_PATH,
)
from normative_contract import file_sha256
from decision_runtime import NormativeDecisionRuntime
from normative_runtime import init_normative_runtime, normative_review_coverage
from release_identity import compute_release_manifest
from config import ROOT_DIR
BOOTSTRAPPED = False
RUNTIME_BUILD_INFO: dict = {}
DECISION_RUNTIME = NormativeDecisionRuntime()
def bootstrap_once(force: bool = False) -> None:
"""Load one validated, immutable runtime build as the knowledge source."""
global BOOTSTRAPPED, RUNTIME_BUILD_INFO, DECISION_RUNTIME
if BOOTSTRAPPED and not force:
return
ontology_path = CORPUS_MCKF_ONTOLOGY_PATH if CORPUS_MCKF_ONTOLOGY_PATH.exists() else MCKF_ONTOLOGY_PATH
manifest = _load_json(RUNTIME_BUILD_MANIFEST_PATH)
validation = _load_json(MCKF_VALIDATION_REPORT_PATH)
ontology = load_ontology(ontology_path)
if manifest:
expected_build = str(manifest.get("build_id", "") or "")
actual_build = str(ontology.get("build_id", "") or expected_build or "")
if expected_build and actual_build and expected_build != actual_build and ontology.get("build_id"):
print(f"[WARN] MCKF build_id uyuşmazlığı uyarısı: manifest={expected_build}, corpus={actual_build}")
expected_hash = str(
((manifest.get("artifacts", {}) or {}).get(ontology_path.name, {}) or {}).get("sha256", "")
)
if expected_hash and file_sha256(ontology_path) != expected_hash:
print(f"[WARN] MCKF corpus dosya özeti uyarısı: {ontology_path.name}")
release = compute_release_manifest(ROOT_DIR, actual_build)
expected_release = str(manifest.get("release_id", "") or "")
if expected_release and expected_release != release.get("release_id"):
print(f"[WARN] MitrAnlil release_id uyuşmazlığı uyarısı: expected={expected_release}, actual={release.get('release_id')}")
else:
release = compute_release_manifest(ROOT_DIR, str(ontology.get("build_id", "") or ""))
if validation and validation.get("conforms") is not True:
raise RuntimeError("MCKF corpus doğrulama kapısından geçmedi; runtime başlatılmadı.")
init_clause_retrieval(ontology)
init_normative_runtime(ontology)
DECISION_RUNTIME = NormativeDecisionRuntime.from_corpus(ontology)
RUNTIME_BUILD_INFO = {
"build_id": ontology.get("build_id", "legacy-unversioned"),
"release_id": release.get("release_id", ""),
"schema": ontology.get("schema", ""),
"manifest": bool(manifest),
"validation_conforms": validation.get("conforms") if validation else None,
"decision_contract_count": len(DECISION_RUNTIME.contracts),
"runtime_layers": ["knowledge_assistant", "decision_runtime"],
"review_coverage": normative_review_coverage(),
}
BOOTSTRAPPED = True
def reload_runtime() -> dict:
bootstrap_once(force=True)
return dict(RUNTIME_BUILD_INFO)
def runtime_build_info() -> dict:
return dict(RUNTIME_BUILD_INFO)
def evaluate_normative_decision(decision_type: str, facts: dict, as_of_date: str = "") -> dict:
bootstrap_once()
return DECISION_RUNTIME.evaluate(decision_type, facts, as_of_date)
def decision_governance_report() -> dict:
bootstrap_once()
return DECISION_RUNTIME.governance_report()
def decision_contract_catalog() -> list[dict]:
bootstrap_once()
return [
{
"contract_id": item.get("contract_id", ""),
"decision_type": item.get("decision_type", ""),
"title": item.get("title", ""),
"version": item.get("version", ""),
"source_refs": item.get("source_refs", []) or [],
"input_schema": item.get("input_schema", {}) or {},
"judgment_requirements": item.get("judgment_requirements", []) or [],
"review_status": item.get("review_status", ""),
}
for item in DECISION_RUNTIME.contracts
]
def _load_json(path) -> dict:
if not path.exists():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return {}
|