"""Process-wide singletons.""" from __future__ import annotations import os from pathlib import Path from ..case_loader import CaseData, Witness, GroundTruth, load_case from ..case_state import CaseState _CASES: dict[str, CaseData] | None = None _SESSIONS: dict[str, CaseState] = {} _WITNESS_ENGINE = None def cases() -> dict[str, CaseData]: global _CASES if _CASES is None: _CASES = {} for p in sorted(Path("cases").glob("case_*.yaml")): c = load_case(p) _CASES[c.case_id] = c return _CASES def sessions() -> dict[str, CaseState]: return _SESSIONS def set_witness_engine(engine) -> None: global _WITNESS_ENGINE _WITNESS_ENGINE = engine def witness_engine(): return _WITNESS_ENGINE def init_real_witness_engine(): """Pick backend by WITNESS_BACKEND env var. - "providers" (default): HF Inference Providers, no local model download. - "llama_cpp": local llama-cpp-python, needs the GGUF on disk. """ backend = os.environ.get("WITNESS_BACKEND", "providers").lower() if backend == "llama_cpp": from ..witness_engine import WitnessEngine, load_llama from .. import MODEL_PATH if not MODEL_PATH.exists(): raise RuntimeError( f"model not found at {MODEL_PATH}; run scripts/download_model.py" ) set_witness_engine(WitnessEngine(load_llama(str(MODEL_PATH)))) else: from ..witness_engine import InferenceProvidersWitnessEngine set_witness_engine(InferenceProvidersWitnessEngine()) def add_case_from_dict(d: dict) -> CaseData: """Register a procedural case at runtime (Endless / Daily).""" dock = list(d["dock"]) def _parse(raw: dict | None): if not raw: return {} out = {} for k, v in raw.items(): out[frozenset(s.strip() for s in str(k).split(","))] = float(v) return out ws = [Witness( id=w["id"], name=w["name"], portrait_ref=w.get("portrait_ref", ""), persona=w["persona"], secret_knowledge=w["secret_knowledge"], reliability=float(w.get("reliability", 1.0)), independence_group=w.get("independence_group"), cost_days=int(w.get("cost_days", 1)), is_physical_evidence=bool(w.get("is_physical_evidence", False)), target_mass=_parse(w.get("target_mass")), ) for w in d["witnesses"]] case = CaseData( case_id=d["case_id"], title=d["title"], setting_blurb=d["setting_blurb"], victim=d["victim"], dock=dock, court_days=int(d["court_days"]), threshold=float(d["threshold"]), methods_available=list(d["methods_available"]), witnesses=ws, ground_truth=GroundTruth( culprit=d["ground_truth"]["culprit"], accomplice=d["ground_truth"].get("accomplice"), ), intended_path=d.get("intended_path", ""), reveal_expectations=d.get("reveal_expectations", {}), ) cases()[case.case_id] = case return case