| """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 |
|
|
