runtime_adapters.py

"""
runtime_adapters.py — the REAL injected components for harness.EMoEHarness.

`harness.py` is a deterministic control-flow skeleton: every heavy component
(encode, rewrite, cluster geometry, retrieval, mint+run, verify) is INJECTED as a
plain callable conforming to a Protocol. This module supplies the real
implementations, wired against the actual project modules:

    Encoder         -> BGEEncoder            (sentence-transformers, BAAI/bge-base-en-v1.5)
    Rewriter        -> IdentityRewriter       (default) | BaseRewriter (opt-in, base greedy)
    ClusterGeometry -> ClusterIndex           (built from runs/hyper_v1/z_cache.pt train means)
    Retriever       -> FaissRetriever          (FAISS index built by build_rag_index.py)
    ExpertRunner    -> HyperExpertRunner       (frozen v11 base + trained hypernetwork; mint+gen)
    Verifier        -> BuiltinVerifier         (degeneracy + optional structural_checks)
                       LadderVerifier (stub)    (wire to verification_ladder once confirmed)

`build_runtime(cfg)` assembles an EMoEHarness from on-disk paths.

THREE THINGS THAT ARE LOAD-BEARING AND VERIFIED AT RUNTIME (not asserted):
  1. z-encoding consistency. The controller geometry (tau_rag_on=0.27,
     tau_k_escalate=0.30, AUC 0.95-0.98) was calibrated on z_cache.pt, which
     was produced by `encode_ar.py --backend st --st_model BAAI/bge-base-en-v1.5`.
     If the runtime BGE encoding does not reproduce that cache, every distance
     and every threshold is meaningless. `ClusterIndex.consistency_check()`
     re-encodes known descriptors and compares to the cache; build_runtime ABORTS
     if cosine is below --z_consistency_floor. (See encode_ar.py if it fails.)
  2. alpha path. The trained hypernetwork's LoRALinear scaling is fixed at 1.0
     (wrapped with alpha=None in hyper_lora.wrap_model). The global backstop
     alpha therefore CANNOT come from the wrapper; HyperExpertRunner applies it
     by scaling the MINTED deltas (alpha=1.0 == training; alpha=0 == bare base).
  3. prompt distribution. The hypernetwork was trained (train_hyper_sft.make_batch)
     on raw `input + "\n" + output`, NOT ChatML-wrapped. The runtime prompt
     mirrors that exactly so the mint sees its training distribution. RAG context
     is the one untrained degree of freedom (prepended); flagged below.

Heavy imports (torch / faiss / sentence_transformers) are LAZY so this module
imports cleanly for logic tests without them.
"""

from __future__ import annotations

import json
import math
import os
import random
from dataclasses import dataclass, field, asdict
from typing import Any, Callable, Optional

import numpy as np


# ===========================================================================
# Verbose step logging — the user anticipates hiccups and wants to SEE, live,
# which step is running and (especially) which verifier rung fired. Logs go to
# stderr (tagged), so stdout stays clean for ANSWER/TRACE in --batch mode.
#   level 0 = quiet, 1 = steps (default), 2 = debug (per-encode etc.)
# ===========================================================================

import sys
from time import perf_counter

_VERBOSE = 1


def set_verbose(level: int):
    global _VERBOSE
    _VERBOSE = int(level)


def log(msg: str, *, level: int = 1, tag: str = "STEP"):
    if _VERBOSE >= level:
        print(f"  [{tag:<8}] {msg}", file=sys.stderr, flush=True)


def _short(s: str, n: int = 56) -> str:
    s = (s or "").replace("\n", "\\n")
    return s if len(s) <= n else s[: n - 1] + "…"


# ===========================================================================
# Outcome shim — shape-compatible with verification_ladder / structural_checks
# (harness reads .verdict in {ACCEPT,REJECT,ABSTAIN}, .recovery in {reretrieve,None},
#  .score float). If you wire the real ladder, return ITS Outcome instead.
# ===========================================================================

@dataclass
class Outcome:
    verdict: str                 # "ACCEPT" | "REJECT" | "ABSTAIN"
    recovery: Optional[str] = None   # "reretrieve" | None
    score: float = 0.0
    rung: str = ""               # which check fired (instrumentation)


# ===========================================================================
# Encoder — BGE, must match encode_ar.py --backend st (no instruction prefix
# on the z side; the ar.jsonl text was the raw descriptor with no template).
# ===========================================================================

class BGEEncoder:
    """L2-normalized BAAI/bge-base-en-v1.5 embeddings (d_z=768).

    __call__(text) is the Z encoder (NO instruction prefix) — this is what must
    reproduce z_cache.pt. encode_query/encode_passages are for RAG and may use
    the asymmetric query instruction (separate knob; does NOT touch z geometry).
    """

    # bge-base-en-v1.5 official asymmetric-retrieval query instruction.
    DEFAULT_QUERY_INSTRUCTION = "Represent this sentence for searching relevant passages: "

    def __init__(self, st_model: str = "BAAI/bge-base-en-v1.5",
                 device: Optional[str] = None,
                 query_instruction: Optional[str] = DEFAULT_QUERY_INSTRUCTION):
        from sentence_transformers import SentenceTransformer  # lazy
        self.model = SentenceTransformer(st_model, device=device)
        self.query_instruction = query_instruction or ""
        _dim_fn = (getattr(self.model, "get_embedding_dimension", None)
                   or self.model.get_sentence_embedding_dimension)
        self.dim = int(_dim_fn())

    def __call__(self, text: str) -> np.ndarray:
        v = self.model.encode([text], normalize_embeddings=True,
                              show_progress_bar=False)[0]
        log(f"encode z: \"{_short(text)}\" -> z[{len(v)}]", level=2, tag="ENCODE")
        return np.asarray(v, dtype=np.float32)

    def encode_many(self, texts: list[str], batch_size: int = 256) -> np.ndarray:
        v = self.model.encode(texts, normalize_embeddings=True,
                              batch_size=batch_size, show_progress_bar=False)
        return np.asarray(v, dtype=np.float32)

    def encode_query(self, text: str) -> np.ndarray:
        v = self.model.encode([self.query_instruction + text],
                              normalize_embeddings=True, show_progress_bar=False)[0]
        return np.asarray(v, dtype=np.float32)


# ===========================================================================
# Rewriter — default identity (safest on the v11 placeholder base, whose greedy
# output is incoherent). BaseRewriter is opt-in once the owned base can rewrite.
# ===========================================================================

class IdentityRewriter:
    def __call__(self, raw_request: str) -> str:
        return raw_request


class BaseRewriter:
    """One constrained greedy generation: raw request -> clean descriptor.
    Uses the SAME frozen base as the experts, with NO adapter set (deltas=None).
    On the v11 placeholder base this is expected to be poor — keep IdentityRewriter
    as the default until the owned base is in place (context §11 open question)."""

    PROMPT = ("Rewrite the user request as one short, canonical task description.\n"
              "Request: {req}\nTask description:")

    def __init__(self, runner: "HyperExpertRunner", max_new_tokens: int = 32):
        self.runner = runner
        self.max_new_tokens = max_new_tokens

    def __call__(self, raw_request: str) -> str:
        text = self.runner.generate_base(self.PROMPT.format(req=raw_request),
                                          max_new_tokens=self.max_new_tokens)
        # take the first line only; fall back to raw if empty
        line = text.strip().splitlines()[0].strip() if text.strip() else ""
        return line or raw_request


# ===========================================================================
# ClusterGeometry — nearest TRAINING-task mean-z (cosine distance) + neighbors.
# Built from z_cache.pt, with the trainer's seed-0 split replicated so the
# "known clusters" are exactly the tasks the hypernetwork trained on.
# ===========================================================================

class ClusterIndex:
    def __init__(self, means: np.ndarray, task_ids: list[str]):
        # means: (N, d), each row L2-normalized
        assert means.ndim == 2 and means.shape[0] == len(task_ids)
        self.M = means.astype(np.float32)
        self.task_ids = list(task_ids)
        # guard: rows should be unit norm (cosine == dot)
        norms = np.linalg.norm(self.M, axis=1)
        if not np.allclose(norms, 1.0, atol=1e-3):
            self.M = self.M / (norms[:, None] + 1e-12)

    def nearest(self, z: np.ndarray) -> tuple[float, Any]:
        z = _unit(z)
        sims = self.M @ z                      # (N,)
        i = int(np.argmax(sims))
        dist = float(1.0 - sims[i])            # cosine distance
        return dist, self.task_ids[i]

    def top_k_z(self, z: np.ndarray, k: int) -> list[np.ndarray]:
        z = _unit(z)
        sims = self.M @ z
        order = np.argsort(-sims)[:max(0, k)]
        return [self.M[i].copy() for i in order]

    # ---- instrumentation / the load-bearing self-check --------------------

    def distance_stats(self, zs: np.ndarray) -> dict:
        """Distances of a batch of z's (rows) to their nearest cluster."""
        zs = zs / (np.linalg.norm(zs, axis=1, keepdims=True) + 1e-12)
        sims = zs @ self.M.T                   # (B, N)
        d = 1.0 - sims.max(axis=1)
        return {"mean": float(d.mean()), "p50": float(np.percentile(d, 50)),
                "p90": float(np.percentile(d, 90)), "p95": float(np.percentile(d, 95)),
                "max": float(d.max()), "min": float(d.min())}


def _unit(v: np.ndarray) -> np.ndarray:
    v = np.asarray(v, dtype=np.float32).reshape(-1)
    n = float(np.linalg.norm(v))
    return v / (n + 1e-12)


def _replicate_split(tasks: list[dict], val_frac: float, seed: int):
    """EXACTLY mirrors train_hyper_sft.train(): random.seed(seed) then
    random.shuffle(tasks); val = tasks[:n_val]; train = tasks[n_val:]."""
    tasks = list(tasks)
    random.seed(seed)
    random.shuffle(tasks)
    n_val = int(len(tasks) * val_frac)
    return tasks[n_val:], tasks[:n_val]        # (train, val)


def build_cluster_index(tasks_path: str, z_cache_path: str, *,
                        val_frac: float = 0.1, seed: int = 0,
                        scope: str = "train") -> tuple[ClusterIndex, dict]:
    """clusters = per-task MEAN of cached variant z's (then renormalized).
    scope='train' matches the §10 calibration (val->nearest-TRAIN distance)."""
    import torch  # lazy
    with open(tasks_path) as f:
        tasks = [json.loads(ln) for ln in f if ln.strip()]
    z_cache = torch.load(z_cache_path, map_location="cpu")
    train, val = _replicate_split(tasks, val_frac, seed)
    chosen = {"train": train, "val": val, "all": tasks}[scope]

    means, ids, missing = [], [], 0
    for t in chosen:
        tid = t["task_id"]
        if tid not in z_cache:
            missing += 1
            continue
        Z = np.asarray(z_cache[tid].float().cpu().numpy(), dtype=np.float32)
        if Z.ndim == 1:
            Z = Z[None, :]
        m = Z.mean(axis=0)
        means.append(m / (np.linalg.norm(m) + 1e-12))
        ids.append(tid)
    if not means:
        raise SystemExit(f"no cluster means built from {z_cache_path} "
                         f"(scope={scope}); check that z_cache keys match task_ids.")
    M = np.stack(means, axis=0)
    info = {"scope": scope, "n_clusters": len(ids), "n_tasks_total": len(tasks),
            "n_train": len(train), "n_val": len(val), "missing_in_cache": missing,
            "d_z": int(M.shape[1])}
    return ClusterIndex(M, ids), info


# ===========================================================================
# Retriever — FAISS over BGE-embedded passages. Returns scored passages, highest
# first; the harness applies the score floor, profile-shape granularity, and the
# token budget (harness._assemble_context).
# ===========================================================================

class FaissRetriever:
    def __init__(self, index_path: str, passages_path: str, encoder: BGEEncoder,
                 over_fetch: int = 20, use_query_instruction: bool = True):
        import faiss  # lazy
        self.index = faiss.read_index(index_path)
        self.passages = _load_passages(passages_path)
        if self.index.ntotal != len(self.passages):
            raise SystemExit(f"FAISS ntotal {self.index.ntotal} != #passages "
                             f"{len(self.passages)} — index and passages out of sync. "
                             f"Rebuild with build_rag_index.py.")
        self.encoder = encoder
        self.over_fetch = over_fetch
        self.use_query_instruction = use_query_instruction

    def __call__(self, query: str, budget_tokens: int) -> list[tuple[float, str]]:
        t0 = perf_counter()
        qv = (self.encoder.encode_query(query) if self.use_query_instruction
              else self.encoder(query)).astype(np.float32)[None, :]
        scores, idx = self.index.search(qv, self.over_fetch)
        out = []
        for s, i in zip(scores[0], idx[0]):
            if i < 0:
                continue
            out.append((float(s), self.passages[int(i)]))   # IP on unit vecs == cosine
        top = ", ".join(f"{s:.2f}" for s, _ in out[:5])
        log(f"retrieve: q=\"{_short(query)}\" -> {len(out)} candidates "
            f"(top: [{top}]) in {1e3*(perf_counter()-t0):.0f}ms", tag="RETRIEVE")
        return out


def _load_passages(path: str) -> list[str]:
    passages = []
    if path.endswith(".jsonl"):
        with open(path) as f:
            for ln in f:
                ln = ln.strip()
                if not ln:
                    continue
                o = json.loads(ln)
                passages.append(o["text"] if isinstance(o, dict) else str(o))
    else:  # one passage per line
        with open(path) as f:
            passages = [ln.rstrip("\n") for ln in f if ln.strip()]
    return passages


# ===========================================================================
# ExpertRunner — load frozen v11 base + trained hypernetwork; mint per request,
# apply the global alpha backstop by SCALING the minted deltas, generate greedily.
# ===========================================================================

def _scale_deltas(deltas: dict, alpha: float) -> dict:
    """Global backstop. LoRA dW = scaling * B@A (scaling fixed at 1.0 in the
    trained wrapper) -> scale B by alpha. FiLM deltas are additive -> scale both.
    alpha=1.0 reproduces training EXACTLY; alpha=0.0 -> zero delta -> bare base."""
    if alpha == 1.0:
        return deltas
    out = {}
    for name, payload in deltas.items():
        kind = payload[0]
        if kind == "lora":
            _, A, B = payload
            out[name] = ("lora", A, B * alpha)
        else:  # "film": (kind, dlog_alpha, dscale)
            _, dlog, dscale = payload
            out[name] = ("film", dlog * alpha, dscale * alpha)
    return out


def _banned_ngram_tokens(seq: list[int], n: int) -> set:
    """HF-style no_repeat_ngram: tokens whose emission would complete an n-gram
    that already occurred in `seq`. Looks at the (n-1)-token suffix as the prefix."""
    if n < 2 or len(seq) < n:
        return set()
    prefix = tuple(seq[-(n - 1):])
    banned = set()
    for i in range(len(seq) - n + 1):
        if tuple(seq[i:i + n - 1]) == prefix:
            banned.add(seq[i + n - 1])
    return banned


@dataclass
class GenConfig:
    max_new_tokens: int = 256
    temperature: float = 0.0      # greedy -> deterministic orchestration
    top_k: Optional[int] = None
    context_sep: str = "\n\n"     # between RAG context and the request
    request_sep: str = "\n"       # mirrors train_hyper_sft.make_batch sep
    # anti-loop controls — OFF by default so the v11 baseline stays vanilla/greedy.
    # These fix the SYMPTOM (loops); they do not make a weak base coherent.
    repetition_penalty: float = 1.0      # CTRL-style; >1 penalizes already-seen tokens
    no_repeat_ngram_size: int = 0        # >0 blocks repeating any n-gram (try 3)
    # correctness: never emit untrained pad logits (tokenizer vocab 50264, model pads to 50304)
    mask_pad_vocab: bool = True
    # stop at ChatML turn end so the base can't ramble into a hallucinated next turn
    stop_on_eos: bool = True
    # wrap the GENERATION prompt in ChatML (base is ChatML-pretrained). z is still
    # encoded from the CLEAN descriptor, so this does NOT pollute the controller.
    # NOTE: the v11 hyper was trained on RAW input+\n, so adapter+ChatML is
    # off-distribution for the adapter; the clean comparison is bare-base+ChatML
    # (--chat_format --alpha 0.0). For V12, align the hyper TRAINING data to ChatML.
    chat_format: bool = False


class HyperExpertRunner:
    def __init__(self, base_ckpt: str, hyper_ckpt: str, tokenizer,
                 device: Optional[str] = None, gen: Optional[GenConfig] = None,
                 max_context: Optional[int] = None):
        import torch  # lazy
        from dataclasses import fields as _fields
        from model_hybrid import GPT, GPTConfig
        import hyper_lora as Hmod

        self.torch = torch
        self.tok = tokenizer
        self.gen = gen or GenConfig()
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")

        # --- frozen base (same load path as train_hyper_sft.train) ---
        ckpt = torch.load(base_ckpt, map_location="cpu")
        margs = ckpt.get("model_args") or ckpt.get("config") or ckpt.get("args") or {}
        if hasattr(margs, "__dict__"):
            margs = vars(margs)
        valid = {f.name for f in _fields(GPTConfig)}
        gcfg = GPTConfig(**{k: v for k, v in margs.items() if k in valid})
        sd = (ckpt.get("model") or ckpt.get("state_dict")
              or ckpt.get("model_state_dict"))
        if sd is None and all(torch.is_tensor(v) for v in ckpt.values()):
            sd = ckpt
        if sd is None:
            raise SystemExit(f"no weights in {base_ckpt}; keys={list(ckpt.keys())}")
        sd = {k.replace("_orig_mod.", ""): v for k, v in sd.items()}
        base = GPT(gcfg)
        miss, unexp = base.load_state_dict(sd, strict=False)
        if miss or unexp:
            print(f"NOTE base state_dict: {len(miss)} missing / {len(unexp)} unexpected "
                  f"(first missing {miss[:2]})")
        self.block_size = gcfg.block_size
        self.max_context = max_context or gcfg.block_size

        # --- rebuild the hypernetwork exactly as it was built for training ---
        hst = torch.load(hyper_ckpt, map_location="cpu")
        if "hyper" not in hst:
            raise SystemExit(f"{hyper_ckpt} is not a train_hyper_sft checkpoint "
                             f"(no 'hyper' key); keys={list(hst.keys())[:6]}")
        acfg = Hmod.AdaptConfig(**hst["adapt_cfg"])
        d_z = int(hst["d_z"]); d_trunk = int(hst.get("d_trunk", 512))
        adapted, hyper, sites = Hmod.build(base, d_z=d_z, cfg=acfg, d_trunk=d_trunk)
        if len(sites) != len(hst["sites"]):
            raise SystemExit(f"site mismatch: rebuilt {len(sites)} vs checkpoint "
                             f"{len(hst['sites'])} — base config differs from the one "
                             f"the hypernetwork was trained on. Use the matching --base_ckpt.")
        hyper.load_state_dict(hst["hyper"])
        self.adapted = adapted.to(self.device).eval()
        self.hyper = hyper.to(self.device).eval()
        self.d_z = d_z
        self.n_sites = len(sites)
        self.tok_vocab = getattr(tokenizer, "vocab_size", None)
        # resolve ChatML end tokens for EOS stopping (single-id specials only)
        self.stop_ids = set()
        for s in ("<|im_end|>", "<|endoftext|>"):
            try:
                e = tokenizer.encode(s)
                if len(e) == 1:
                    self.stop_ids.add(int(e[0]))
            except Exception:
                pass
        print(f"HyperExpertRunner: base {base.get_num_params()/1e6:.1f}M frozen | "
              f"hyper {sum(p.numel() for p in hyper.parameters())/1e6:.2f}M | "
              f"{self.n_sites} sites | d_z {d_z} | device {self.device}")

    # ---- generation primitives -------------------------------------------

    def _generate(self, prompt_text: str, max_new_tokens: int) -> str:
        """Greedy/sampled decode with optional anti-loop controls. Replaces the
        base model.generate so repetition_penalty / no_repeat_ngram / pad-masking
        are available; with all controls off and temperature 0 it is identical to
        the base greedy generate."""
        torch = self.torch
        g = self.gen
        ids = self.tok.encode(prompt_text)[: self.block_size]
        idx = torch.tensor([ids], dtype=torch.long, device=self.device)
        t0 = perf_counter()
        new_ids: list[int] = []
        n_blocked = 0
        for _ in range(max_new_tokens):
            cond = idx if idx.size(1) <= self.max_context else idx[:, -self.max_context:]
            logits, _ = self.adapted(cond)             # (1, 1, V): last position only
            logits = logits[:, -1, :].float()          # (1, V)
            # never emit untrained pad logits (ids >= tokenizer vocab)
            if g.mask_pad_vocab and self.tok_vocab and self.tok_vocab < logits.size(-1):
                logits[:, self.tok_vocab:] = -float("inf")
            # CTRL repetition penalty on already-emitted tokens
            if g.repetition_penalty and g.repetition_penalty != 1.0:
                seen = torch.unique(idx[0])
                vals = logits[0, seen]
                logits[0, seen] = torch.where(vals > 0, vals / g.repetition_penalty,
                                              vals * g.repetition_penalty)
            # block n-grams that would repeat one already generated
            if g.no_repeat_ngram_size and g.no_repeat_ngram_size >= 2:
                banned = _banned_ngram_tokens(idx[0].tolist(), g.no_repeat_ngram_size)
                if banned:
                    n_blocked += len(banned)
                    logits[0, list(banned)] = -float("inf")
            if g.temperature <= 0:
                nxt = int(logits.argmax(-1))
            else:
                lg = logits / g.temperature
                if g.top_k:
                    v, _ = torch.topk(lg, min(g.top_k, lg.size(-1)))
                    lg[lg < v[:, [-1]]] = -float("inf")
                nxt = int(torch.multinomial(torch.softmax(lg, -1), 1))
            new_ids.append(nxt)
            idx = torch.cat([idx, torch.tensor([[nxt]], device=self.device)], dim=1)
            if g.stop_on_eos and nxt in self.stop_ids:
                new_ids.pop()                      # don't include the control token in output
                break
        extra = (f", rep_pen={g.repetition_penalty}" if g.repetition_penalty != 1.0 else "")
        extra += (f", no_repeat_{g.no_repeat_ngram_size}gram(blocked {n_blocked})"
                  if g.no_repeat_ngram_size >= 2 else "")
        log(f"generate: prompt {len(ids)} tok -> +{len(new_ids)} tok "
            f"in {perf_counter()-t0:.2f}s "
            f"({'greedy' if g.temperature <= 0 else f'T={g.temperature}'}{extra})",
            tag="GENERATE")
        return self.tok.decode(new_ids)

    def generate_base(self, prompt_text: str, max_new_tokens: int = 32) -> str:
        """Bare base (no adapter) — used by BaseRewriter."""
        log("rewrite: bare base (no adapter), greedy", tag="REWRITE")
        with self.torch.no_grad():
            self.adapted.set_deltas(None)
            return self._generate(prompt_text, max_new_tokens)

    def _build_prompt(self, request: str, context: str) -> str:
        g = self.gen
        if getattr(g, "chat_format", False):
            user = f"{context}{g.context_sep}{request}" if context else request
            return (f"<|im_start|>user\n{user}<|im_end|>\n"
                    f"<|im_start|>assistant\n")
        if context:
            return f"{context}{g.context_sep}{request}{g.request_sep}"
        return f"{request}{g.request_sep}"

    # ---- the ExpertRunner protocol ---------------------------------------

    def __call__(self, z: Any, request: str, context: str, alpha: float) -> str:
        torch = self.torch
        z_t = torch.as_tensor(np.asarray(z, dtype=np.float32), device=self.device)
        with torch.no_grad():
            if alpha == 0.0:
                log("mint: alpha=0.0 -> BARE BASE (no adapter)", tag="MINT")
                self.adapted.set_deltas(None)           # bare base (trivial path)
            else:
                log(f"mint: alpha={alpha} -> {self.n_sites} sites; "
                    f"context={'yes' if context else 'no'}", tag="MINT")
                self.adapted.set_deltas(_scale_deltas(self.hyper(z_t), alpha))
            return self._generate(self._build_prompt(request, context),
                                  self.gen.max_new_tokens)


# ===========================================================================
# Verifier — the runtime defense against shatter (CATCH, don't predict).
#
# LadderVerifier wraps the REAL verification_ladder.verify() and maps its
# two-axis Verdict(passed, verifier_available, rung, score) onto the harness's
# three-way Outcome. This is the intended production verifier.
#
# Mapping (faithful to verification_ladder.run_loop semantics):
#   rung == "open"  (verifier_available=False) -> ABSTAIN(None)
#                       => harness ships the single alpha-damped mint (k=1).
#                          This is the ladder's "undefended remainder" — the
#                          accurate defense on open-ended work.
#   passed & available                          -> ACCEPT  (a rung fired and passed)
#   !passed & available  (failed exec/compile/  -> REJECT  (a selector exists ->
#                         rag.match, or degeneracy)           harness escalates via
#                                                             neighbor-cluster z's)
#
# There is NO "reretrieve" in the ladder: a failed rag.match is a REJECT that
# permits ESCALATION (mint more experts), not re-retrieval. The harness's
# reretrieve branch therefore stays inert under the ladder (rag_recovery is a
# separate, still-unmeasured experiment — current-state §8).
#
# exec.execute (STRONG) only fires when the Request carries its own `tests`.
# Serving requests usually don't, so exec degrades to the parse/compile FLOOR —
# the ladder's documented asymmetry, not a limitation added here. set_request_tests()
# lets you supply tests per request (e.g. {"expected": "..."}) to reach the strong rung.
#
# BuiltinVerifier is the dependency-free FALLBACK for when verification_ladder /
# exec_checks aren't importable; it keeps the system running but is weaker.
# ===========================================================================

def _verdict_to_outcome(v) -> Outcome:
    """Map a verification_ladder.Verdict (or anything with .passed /
    .verifier_available / .rung / .score) onto the harness Outcome."""
    rung = getattr(v, "rung", "?")
    score = getattr(v, "score", None)
    score = float(score) if score is not None else 0.0
    if not getattr(v, "verifier_available", False):
        return Outcome("ABSTAIN", recovery=None, score=score, rung=rung)
    if getattr(v, "passed", False):
        return Outcome("ACCEPT", recovery=None, score=score, rung=rung)
    return Outcome("REJECT", recovery=None, score=score, rung=rung)


class LadderVerifier:
    """Adapter to verification_ladder.verify(). Requires verification_ladder +
    exec_checks on PYTHONPATH. Pass verifier='ladder' (or 'auto') to build_runtime."""

    def __init__(self, encoder, tau: float = 0.60, executor_fn=None,
                 enabled_rungs=None):
        self.VL = _try_import("verification_ladder")
        if self.VL is None:
            raise SystemExit(
                "verification_ladder not importable (it also needs exec_checks). "
                "Put both .py on PYTHONPATH, or use verifier='builtin'.")
        self.encoder = encoder           # embed_fn for rag.match (output vs passages)
        self.tau = tau
        self.executor_fn = executor_fn   # None -> ladder's canonical match_executor
        # None -> all rungs on. Else a set of names from VL.ALL_RUNGS; disabled
        # rungs are skipped (output falls through). Order is fixed in the ladder.
        self.enabled_rungs = set(enabled_rungs) if enabled_rungs is not None else None
        self._tests = None               # optional per-request bundled tests

    def set_request_tests(self, tests):
        """Attach bundled tests/expected for the NEXT request so the STRONG
        exec.execute rung is reachable. Persists across that request's escalation
        re-verifies; clear (set None) before the next request."""
        self._tests = tests

    def __call__(self, request: str, output: str, context: str) -> Outcome:
        VL = self.VL
        # harness joined passages with "\n\n" in _assemble_context -> split back
        passages = [p for p in (context.split("\n\n") if context else []) if p.strip()]
        req = VL.Request(text=request, tests=self._tests)
        rag = VL.Rag(passages=passages) if passages else None
        active = ("all" if self.enabled_rungs is None
                  else "+".join(r for r in VL.ALL_RUNGS if r in self.enabled_rungs))
        log(f"verify: ladder dispatch [{active}]  (tests={'yes' if self._tests else 'no'}, "
            f"passages={len(passages)})", tag="VERIFY")
        v = VL.verify(output, req, rag, embed_fn=self.encoder,
                      executor_fn=self.executor_fn, tau=self.tau,
                      enabled_rungs=self.enabled_rungs)
        oc = _verdict_to_outcome(v)
        sc = f" score={v.score:.3f}" if getattr(v, "score", None) is not None else ""
        log(f"verify: RUNG FIRED = {v.rung}  (passed={v.passed} "
            f"available={v.verifier_available}{sc})  ->  {oc.verdict}", tag="VERIFY")
        if _VERBOSE >= 2 and getattr(v, "detail", ""):
            log(f"verify: detail = {v.detail}", level=2, tag="VERIFY")
        return oc


# ---- dependency-free fallback --------------------------------------------

def _degenerate(text: str) -> bool:
    """Cheap repetition / emptiness gate (fallback only; the ladder uses
    exec_checks.degeneracy when available)."""
    t = (text or "").strip()
    if len(t) < 1:
        return True
    words = t.split()
    if len(words) >= 8:
        uniq = len(set(words)) / len(words)
        if uniq < 0.25:                  # e.g. "the the the the ..."
            return True
    if len(words) >= 12:
        for n in (1, 2, 3):
            tail = words[-3 * n:]
            if len(tail) == 3 * n and tail[:n] == tail[n:2 * n] == tail[2 * n:]:
                return True
    return False


class BuiltinVerifier:
    """Dependency-free FALLBACK. Degeneracy gate + optional structural_checks +
    safe ABSTAIN(None). Weaker than the ladder (no exec/rag rungs); only used when
    verification_ladder/exec_checks aren't importable.

    knowledge_recovery=True flips the unverifiable fallback to ABSTAIN('reretrieve')
    to exercise the answer-conditioned re-retrieval path (off by default)."""

    def __init__(self, knowledge_recovery: bool = False):
        self.knowledge_recovery = knowledge_recovery
        self._structural = _try_import("structural_checks")

    def __call__(self, request: str, output: str, context: str) -> Outcome:
        if _degenerate(output):
            log("verify: RUNG FIRED = degeneracy  ->  REJECT", tag="VERIFY")
            return Outcome("REJECT", score=0.0, rung="degeneracy")
        if self._structural is not None:
            oc = _call_structural(self._structural, request, output)
            if oc is not None:
                log(f"verify: RUNG FIRED = structural  ->  {oc.verdict}", tag="VERIFY")
                return oc
        rec = "reretrieve" if self.knowledge_recovery else None
        log(f"verify: RUNG FIRED = open (builtin fallback)  ->  ABSTAIN"
            + (f"(recovery={rec})" if rec else ""), tag="VERIFY")
        return Outcome("ABSTAIN", recovery=rec, score=0.0, rung="open")


def _try_import(name):
    try:
        import importlib
        return importlib.import_module(name)
    except Exception:
        return None


def _call_structural(mod, request: str, output: str) -> Optional[Outcome]:
    """Best-effort adapter to structural_checks; normalizes its three-way result
    into an Outcome. Returns None if no constraint was parseable (-> fall through)."""
    fn = None
    for cand in ("check", "verify", "check_structural", "run"):
        if hasattr(mod, cand):
            fn = getattr(mod, cand)
            break
    if fn is None:
        return None
    try:
        res = fn(request, output)
    except TypeError:
        try:
            res = fn(request=request, output=output)
        except Exception:
            return None
    except Exception:
        return None
    verdict = getattr(res, "verdict", None) or (res.get("verdict") if isinstance(res, dict) else None)
    if not verdict:
        return None
    if str(verdict).upper() == "ABSTAIN":
        return None
    score = getattr(res, "score", None)
    if score is None and isinstance(res, dict):
        score = res.get("score", 0.0)
    return Outcome(str(verdict).upper(), recovery=None,
                   score=float(score or 0.0), rung="structural")


# ===========================================================================
# Executor dispatch — the config-driven executor_fn for the ladder's
# exec.execute rung. Signature matches verification_ladder: (out, tests) -> bool.
# It ONLY fires when the request carries `tests` (rung 1); serving requests
# without tests never reach it (they hit the parse/compile FLOOR or `open`).
#
# Routing key (in this order): tests["executor"] (explicit) -> tests["lang"] /
# ["kind"] / ["task"] mapped through the config `route`. No match -> the SAFE
# default (canonical match against tests["expected"]). The real runners (g++,
# Lean, a locked python subprocess) are INJECTED infra, never baked in here —
# an unconfigured runner FAILS SAFE (REJECT + a loud log), it never silently
# "passes". Inject real runners via build_runtime(runner_registry=...).
# ===========================================================================

def _match_runner(spec):
    import verification_ladder as VL
    return lambda out, tests: VL.match_executor(out, tests)


def _unwired_runner(kind):
    def run(out, tests):
        log(f"executor '{kind}' has no real runner injected -> fail-safe REJECT "
            f"(pass runner_registry to build_runtime to wire it)", tag="VERIFY")
        return False
    return run


# runner factories keyed by the executor entry's "type". Override/extend by
# passing runner_registry={"gpp_compile": my_factory, ...} to build_runtime.
RUNNER_FACTORIES = {
    "match":             _match_runner,
    "python_subprocess": lambda spec: _unwired_runner("python_subprocess"),
    "gpp_compile":       lambda spec: _unwired_runner("gpp_compile"),
    "lean_check":        lambda spec: _unwired_runner("lean_check"),
}


class DispatchExecutor:
    """executor_fn(out, tests) -> bool. Routes to a registered runner by what the
    request bundled in `tests`; falls back to canonical match."""

    def __init__(self, route: dict, runners: dict, default):
        self.route = route or {}
        self.runners = runners or {}
        self.default = default

    def _pick(self, tests):
        if isinstance(tests, dict):
            if tests.get("executor"):
                return tests["executor"]
            key = tests.get("lang") or tests.get("kind") or tests.get("task")
            if key in self.route:
                return self.route[key]
            if key in self.runners:
                return key
        return None

    def __call__(self, out, tests) -> bool:
        lbl = self._pick(tests)
        runner = self.runners.get(lbl) if lbl else None
        if runner is None:
            log(f"executor: no route for tests -> canonical match", level=2, tag="VERIFY")
            return bool(self.default(out, tests))
        log(f"executor: routed to '{lbl}'", level=2, tag="VERIFY")
        return bool(runner(out, tests))


def make_dispatch_executor(executors_cfg: dict, runner_registry: Optional[dict] = None):
    """Build a DispatchExecutor from the validated `executors` config section.
    Returns None when no executors are configured (-> ladder uses its default
    match_executor). Raises SystemExit on an unknown runner type."""
    if not executors_cfg:
        return None
    reg = {**RUNNER_FACTORIES, **(runner_registry or {})}
    runners = {}
    for label, spec in executors_cfg.items():
        if label == "route" or str(label).startswith("_"):
            continue
        t = spec.get("type")
        if t not in reg:
            raise SystemExit(
                f"executor '{label}' wants unknown runner type {t!r}; "
                f"known: {sorted(reg)} (add it via build_runtime(runner_registry=...))")
        runners[label] = reg[t](spec)
    return DispatchExecutor(executors_cfg.get("route", {}), runners, _match_runner(None))


# ===========================================================================
# Factory — assemble an EMoEHarness from on-disk paths, with the load-bearing
# z-consistency guard.
# ===========================================================================

@dataclass
class RuntimeConfig:
    base_ckpt: str = "hybrid_base/ckpt_hybrid_11r_8750.pt"
    hyper_ckpt: str = "runs/hyper_v1/hyper_ckpt.best.pt"
    z_cache: str = "runs/hyper_v1/z_cache.pt"
    tasks: str = "data/hyper_v1.jsonl"
    st_model: str = "BAAI/bge-base-en-v1.5"
    rag_index: str = "rag/wiki.faiss"
    rag_passages: str = "rag/wiki_passages.jsonl"
    # split replication — MUST match the run that trained hyper_ckpt, or the set
    # of "known clusters" won't be the tasks the hypernetwork actually trained on.
    # current-state §2 banked the 1142 run at val_frac 0.15, seed 0 (971/171).
    # run_hyper.sh's template shows 0.1 — confirm which YOUR run used.
    val_frac: float = 0.15
    seed: int = 0
    cluster_scope: str = "train"
    # behavior
    use_rag: bool = True
    use_base_rewrite: bool = False        # identity by default on v11
    verifier: str = "auto"                # "ladder" | "builtin" | "auto" (ladder if importable)
    tau_match: float = 0.60               # rag.match cosine threshold (ladder TAU_MATCH)
    knowledge_recovery: bool = False      # builtin-fallback only (ladder has no reretrieve)
    device: Optional[str] = None
    verbose: int = 1                      # 0 quiet / 1 steps / 2 debug
    # OOD controller thresholds — set EXPLICITLY so a stale on-disk harness.py
    # default can't silently govern. Grounded values (current-state §10): the
    # rejected placeholders were 0.40/0.45 (above the in-dist max 0.386 -> inert).
    tau_rag_on: float = 0.27
    tau_k_escalate: float = 0.30
    # the load-bearing guard
    z_consistency_floor: float = 0.99
    z_consistency_n: int = 16
    # generation
    max_new_tokens: int = 256
    repetition_penalty: float = 1.0      # >1 to suppress loops (e.g. 1.3); off by default
    no_repeat_ngram_size: int = 0        # >=2 to block repeating n-grams (e.g. 3); off by default
    stop_on_eos: bool = True             # stop at <|im_end|>/<|endoftext|>
    chat_format: bool = False            # wrap generation prompt in ChatML (z stays clean)
    # harness knobs (override HarnessConfig defaults if set non-None)
    alpha_backstop: float = 0.8


def build_runtime(cfg: RuntimeConfig, *, harness_cfg=None, enabled_rungs=None,
                  executors_cfg=None, runner_registry=None):
    """Returns (harness, info). Raises SystemExit with a clear message if the
    z-encoding does not reproduce z_cache.pt (geometry/tau would be invalid).

    harness_cfg     : a fully-built harness.HarnessConfig (full decision surface);
                      None -> built from the three RuntimeConfig knobs (legacy).
    enabled_rungs   : set of ladder rung names; None -> all rungs on.
    executors_cfg   : validated `executors` config section -> a DispatchExecutor
                      is passed as the ladder's executor_fn. None -> ladder default.
    runner_registry : {type: factory} to inject REAL runners (g++, lean, sandbox).
    """
    import harness as Hh
    import tok_v9

    set_verbose(cfg.verbose)

    for p in (cfg.base_ckpt, cfg.hyper_ckpt, cfg.z_cache, cfg.tasks):
        if not os.path.exists(p):
            raise SystemExit(f"missing required artifact: {p}")

    encoder = BGEEncoder(cfg.st_model, device=cfg.device)
    if encoder.dim != _zcache_dim(cfg.z_cache):
        raise SystemExit(f"encoder dim {encoder.dim} != z_cache dim "
                         f"{_zcache_dim(cfg.z_cache)} — wrong --st_model for this cache.")

    clusters, cinfo = build_cluster_index(
        cfg.tasks, cfg.z_cache, val_frac=cfg.val_frac, seed=cfg.seed,
        scope=cfg.cluster_scope)

    # ---- THE LOAD-BEARING GUARD: runtime BGE must reproduce z_cache ----
    cos = _z_consistency(encoder, cfg.tasks, cfg.z_cache, n=cfg.z_consistency_n)
    if cos < cfg.z_consistency_floor:
        raise SystemExit(
            f"z-consistency FAILED: runtime BGE reproduces cached descriptors at "
            f"mean cosine {cos:.4f} < floor {cfg.z_consistency_floor}. The cluster "
            f"geometry and controller thresholds (tau_rag_on, tau_k_escalate) were "
            f"calibrated on z_cache.pt and are INVALID under a different encoding. "
            f"Inspect encode_ar.py for the exact pooling / instruction / prompt "
            f"template it used and match it in BGEEncoder, or re-run encode_ar.py "
            f"with this same model. (Pass --skip_z_check ONLY to debug other paths.)")
    print(f"[guard] z-consistency OK: mean cosine to cache = {cos:.4f} "
          f"(>= {cfg.z_consistency_floor})")

    tok = tok_v9.build()
    runner = HyperExpertRunner(
        cfg.base_ckpt, cfg.hyper_ckpt, tok, device=cfg.device,
        gen=GenConfig(max_new_tokens=cfg.max_new_tokens,
                      repetition_penalty=cfg.repetition_penalty,
                      no_repeat_ngram_size=cfg.no_repeat_ngram_size,
                      stop_on_eos=cfg.stop_on_eos, chat_format=cfg.chat_format))
    if runner.d_z != encoder.dim:
        raise SystemExit(f"hyper d_z {runner.d_z} != encoder dim {encoder.dim}")

    rewrite = BaseRewriter(runner) if cfg.use_base_rewrite else IdentityRewriter()

    if cfg.use_rag and os.path.exists(cfg.rag_index) and os.path.exists(cfg.rag_passages):
        retrieve = FaissRetriever(cfg.rag_index, cfg.rag_passages, encoder)
        rag_state = "on"
    else:
        retrieve = _NullRetriever()
        rag_state = "off (no index)" if cfg.use_rag else "off (disabled)"

    # ---- verifier: real ladder by default; builtin only as fallback ----
    executor_fn = make_dispatch_executor(executors_cfg, runner_registry)
    vname = None
    verify = None
    if cfg.verifier in ("ladder", "auto"):
        try:
            verify = LadderVerifier(encoder, tau=cfg.tau_match,
                                    executor_fn=executor_fn,
                                    enabled_rungs=enabled_rungs)
            vname = "LadderVerifier"
            import verification_ladder as _VL
            active = ("all" if enabled_rungs is None
                      else "+".join(r for r in _VL.ALL_RUNGS if r in enabled_rungs))
            ex_lbls = ([] if not executors_cfg
                       else sorted(l for l in executors_cfg
                                   if l != "route" and not str(l).startswith("_")))
            print(f"[verifier] LadderVerifier (verification_ladder + exec_checks), "
                  f"tau_match={cfg.tau_match}, rungs={active}, executors={ex_lbls or 'default-match'}")
        except SystemExit as e:
            if cfg.verifier == "ladder":
                raise
            print(f"[verifier] ladder unavailable -> BuiltinVerifier fallback ({e})")
    if verify is None:
        verify = BuiltinVerifier(knowledge_recovery=cfg.knowledge_recovery)
        vname = "BuiltinVerifier"
        print("[verifier] BuiltinVerifier (degeneracy + structural; no exec/rag rungs)")

    # controller thresholds: set explicitly; surface any stale on-disk default
    disk = Hh.HarnessConfig()
    if (abs(disk.tau_rag_on - cfg.tau_rag_on) > 1e-9 or
            abs(disk.tau_k_escalate - cfg.tau_k_escalate) > 1e-9):
        print(f"[controller] NOTE on-disk harness.py defaults "
              f"(tau_rag_on={disk.tau_rag_on}, tau_k_escalate={disk.tau_k_escalate}) "
              f"differ from runtime ({cfg.tau_rag_on}, {cfg.tau_k_escalate}); "
              f"USING RUNTIME VALUES. (0.40/0.45 are the rejected, inert placeholders.)")
    hcfg = harness_cfg or Hh.HarnessConfig(alpha_backstop=cfg.alpha_backstop,
                                            tau_rag_on=cfg.tau_rag_on,
                                            tau_k_escalate=cfg.tau_k_escalate)
    harness = Hh.EMoEHarness(encode=encoder, rewrite=rewrite, clusters=clusters,
                             retrieve=retrieve, run_expert=runner, verify=verify,
                             cfg=hcfg)
    info = {"clusters": cinfo, "z_consistency_cos": cos, "rag": rag_state,
            "d_z": encoder.dim, "n_sites": runner.n_sites,
            "verifier": vname, "rewrite": type(rewrite).__name__,
            "tau_rag_on": hcfg.tau_rag_on, "tau_k_escalate": hcfg.tau_k_escalate,
            "alpha_backstop": hcfg.alpha_backstop,
            "k_policy": hcfg.k_policy, "k_max": hcfg.k_max, "k_fixed": hcfg.k_fixed,
            "enabled_rungs": ("all" if enabled_rungs is None else sorted(enabled_rungs)),
            "executors": ([] if not executors_cfg
                          else sorted(l for l in executors_cfg
                                      if l != "route" and not str(l).startswith("_")))}
    return harness, info


class _NullRetriever:
    def __call__(self, query: str, budget_tokens: int):
        return []


def _zcache_dim(z_cache_path: str) -> int:
    import torch
    z = torch.load(z_cache_path, map_location="cpu")
    v = next(iter(z.values()))
    return int(v.shape[-1])


def _z_consistency(encoder: BGEEncoder, tasks_path: str, z_cache_path: str,
                   n: int = 16) -> float:
    """Re-encode n canonical descriptors and compare to z_cache[tid][0] (the
    canonical-variant z that encode_ar wrote first). Returns mean cosine."""
    import torch
    with open(tasks_path) as f:
        tasks = [json.loads(ln) for ln in f if ln.strip()]
    z_cache = torch.load(z_cache_path, map_location="cpu")
    cos, used = [], 0
    for t in tasks:
        tid = t.get("task_id"); desc = t.get("descriptor")
        if not tid or not desc or tid not in z_cache:
            continue
        cached = np.asarray(z_cache[tid][0].float().cpu().numpy(), dtype=np.float32)
        got = encoder(desc)
        cos.append(float(np.dot(_unit(cached), _unit(got))))
        used += 1
        if used >= n:
            break
    return float(np.mean(cos)) if cos else 0.0


# ===========================================================================
# Self-test — pure numpy/python, no torch/faiss/ST. Exercises the logic this
# module owns: cluster math, split replication, delta scaling, degeneracy,
# prompt building, structural adapter normalization.
# ===========================================================================

def _self_test():
    set_verbose(0)
    print("== runtime_adapters self-test (no torch/faiss/ST) ==")
    res = []

    # 1. ClusterIndex nearest + distance is cosine distance, neighbors ordered
    M = np.array([[1, 0, 0], [0, 1, 0], [0.6, 0.8, 0]], dtype=np.float32)
    ci = ClusterIndex(M.copy(), ["a", "b", "c"])
    d, cid = ci.nearest(np.array([0.99, 0.14, 0.0], dtype=np.float32))
    assert cid == "a" and d < 0.02, (cid, d)
    nbrs = ci.top_k_z(np.array([1.0, 0, 0], dtype=np.float32), 2)
    assert len(nbrs) == 2 and np.allclose(nbrs[0], [1, 0, 0])
    res.append(("cluster nearest+neighbors", f"{cid} d={d:.3f}"))

    # 2. split replication matches the trainer's shuffle EXACTLY
    tasks = [{"task_id": f"t{i}"} for i in range(10)]
    train, val = _replicate_split(tasks, val_frac=0.3, seed=0)
    # reproduce the trainer's own sequence independently
    ref = [{"task_id": f"t{i}"} for i in range(10)]
    random.seed(0); random.shuffle(ref)
    assert val == ref[:3] and train == ref[3:], "split does not match trainer"
    res.append(("split replication", f"{len(train)} train / {len(val)} val"))

    # 3. delta scaling: alpha=1 identity, alpha=0 kills LoRA-B and FiLM, 0.5 scales
    deltas = {"s.lora": ("lora", np.ones((2, 3)), np.ones((4, 2)) * 2.0),
              "s.film": ("film", np.ones(5) * 3.0, np.ones(5) * 4.0)}
    assert _scale_deltas(deltas, 1.0) is deltas
    z0 = _scale_deltas(deltas, 0.0)
    assert np.allclose(z0["s.lora"][2], 0) and np.allclose(z0["s.film"][1], 0) \
        and np.allclose(z0["s.film"][2], 0)
    assert np.allclose(z0["s.lora"][1], np.ones((2, 3)))   # A untouched (dW=B@A=0)
    z5 = _scale_deltas(deltas, 0.5)
    assert np.allclose(z5["s.lora"][2], 1.0) and np.allclose(z5["s.film"][1], 1.5)
    res.append(("alpha delta-scaling", "a=1 id / a=0 base / a=.5 scaled"))

    # 4. degeneracy gate
    assert _degenerate("") and _degenerate("the the the the the the the the the")
    assert not _degenerate("def reverse(x): return x[::-1]  # clean answer here ok")
    res.append(("degeneracy gate", "empty+repeat caught, real text passes"))

    # 5. BuiltinVerifier fallback is safe ABSTAIN(None); recovery flag flips it
    v = BuiltinVerifier()
    oc = v("q", "a perfectly ordinary unverifiable answer string", "")
    assert oc.verdict == "ABSTAIN" and oc.recovery is None
    vr = BuiltinVerifier(knowledge_recovery=True)
    assert vr("q", "ordinary answer text goes here", "").recovery == "reretrieve"
    assert v("q", "", "").verdict == "REJECT"            # degeneracy still hard-rejects
    res.append(("builtin verifier", "abstain-safe + recovery flag + reject"))

    # 5b. ladder Verdict -> harness Outcome mapping (the real-ladder integration)
    from dataclasses import dataclass as _dc

    @_dc
    class _V:
        passed: bool; verifier_available: bool; rung: str; score: float = None
    # open (nothing applicable) -> ABSTAIN -> harness ships single mint
    assert _verdict_to_outcome(_V(True, False, "open")).verdict == "ABSTAIN"
    # a rung fired and passed -> ACCEPT
    assert _verdict_to_outcome(_V(True, True, "exec.compile")).verdict == "ACCEPT"
    # rung fired and FAILED (bad syntax / rag below tau / degeneracy) -> REJECT->escalate
    assert _verdict_to_outcome(_V(False, True, "exec.compile")).verdict == "REJECT"
    assert _verdict_to_outcome(_V(False, True, "degeneracy")).verdict == "REJECT"
    o = _verdict_to_outcome(_V(False, True, "rag.match", 0.42))
    assert o.verdict == "REJECT" and abs(o.score - 0.42) < 1e-9 and o.rung == "rag.match"
    # never emits reretrieve (the ladder has no such verdict)
    assert _verdict_to_outcome(_V(True, False, "open")).recovery is None
    res.append(("ladder verdict->outcome", "open=ABSTAIN pass=ACCEPT fail=REJECT"))

    # 5c. no_repeat_ngram blocking helper
    assert _banned_ngram_tokens([5, 1, 2, 3, 1, 2], 3) == {3}   # (1,2)->3 would repeat
    assert _banned_ngram_tokens([1, 2, 3], 4) == set()          # too short
    assert _banned_ngram_tokens([7, 7, 7], 2) == {7}            # bigram 7->7 repeats
    res.append(("no_repeat_ngram helper", "blocks completing token"))

    # 6. prompt building mirrors make_batch (request + '\n'); context prepended;
    #    chat_format wraps in ChatML (and z stays clean — encoded elsewhere)
    class _G: context_sep = "\n\n"; request_sep = "\n"; chat_format = False
    class _R: gen = _G()
    bp = HyperExpertRunner._build_prompt.__get__(_R())
    assert bp("reverse [1,2,3]", "") == "reverse [1,2,3]\n"
    assert bp("Q", "CTX") == "CTX\n\nQ\n"
    class _Gc(_G): chat_format = True
    class _Rc: gen = _Gc()
    bpc = HyperExpertRunner._build_prompt.__get__(_Rc())
    assert bpc("Q", "") == "<|im_start|>user\nQ<|im_end|>\n<|im_start|>assistant\n"
    assert "CTX\n\nQ" in bpc("Q", "CTX") and bpc("Q", "CTX").endswith("assistant\n")
    res.append(("prompt build", "raw + chatml variants"))

    print()
    for k, vv in res:
        print(f"  [ok ] {k:<28} -> {vv}")
    print(f"ALL {len(res)}/{len(res)} RUNTIME-ADAPTER LOGIC TESTS PASSED.")
    print("(torch/faiss/ST paths are exercised on the pod by serve_emoe.py --smoke)")


if __name__ == "__main__":
    _self_test()