src/audit/engine.py

import math
import re
import sys
import time
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Tuple

import numpy as np
import scipy.linalg as la

try:
    import torch
    from transformers import AutoConfig, AutoModelForCausalLM, AutoTokenizer

    # Polyfill for older custom model scripts (e.g. Phi-3 / Phi-3.5 remote code)
    try:
        from transformers.cache_utils import DynamicCache
        if not hasattr(DynamicCache, "from_legacy_cache"):
            @classmethod
            def _from_legacy_cache(cls, past_key_values=None):
                cache = cls()
                if past_key_values is not None:
                    for layer_idx, (key, value) in enumerate(past_key_values):
                        cache.update(key, value, layer_idx)
                return cache
            DynamicCache.from_legacy_cache = _from_legacy_cache
    except Exception:
        pass

    HAS_TRANSFORMERS = True
except ImportError:
    HAS_TRANSFORMERS = False


class AuditError(Exception):
    """Raised whenever an audit cannot be completed."""


PROBE_CORPUS: List[Dict[str, Any]] = [
    {"id": "GEO01", "q": "What is the capital of Australia?", "paraphrases": ["Which city serves as Australia's capital?", "Name the federal capital of Australia."], "answers": ["Canberra"], "cat": "Geography", "is_adversarial": False},
    {"id": "GEO02", "q": "What is the longest river in South America?", "paraphrases": ["Which South American river is the longest?", "Name the longest river on the South American continent."], "answers": ["Amazon"], "cat": "Geography", "is_adversarial": False},
    {"id": "GEO03", "q": "What is the capital of Canada?", "paraphrases": ["Name the capital city of Canada.", "Which city is Canada's national capital?"], "answers": ["Ottawa"], "cat": "Geography", "is_adversarial": False},
    {"id": "GEO04", "q": "Which country has the largest land area in the world?", "paraphrases": ["What is the largest country by area?", "Name the world's largest country by landmass."], "answers": ["Russia"], "cat": "Geography", "is_adversarial": False},
    {"id": "GEO05", "q": "What is the capital of Japan?", "paraphrases": ["Name Japan's capital city.", "Which city serves as the capital of Japan?"], "answers": ["Tokyo"], "cat": "Geography", "is_adversarial": False},
    {"id": "SCI01", "q": "What chemical element has the atomic symbol 'Fe'?", "paraphrases": ["Which element is denoted by Fe in the periodic table?", "In chemistry, Fe represents which element?"], "answers": ["Iron", "iron"], "cat": "Science", "is_adversarial": False},
    {"id": "SCI02", "q": "What is the boiling point of water at standard atmospheric pressure in degrees Celsius?", "paraphrases": ["At sea level, at what temperature in Celsius does water boil?", "In degrees Celsius, water boils at what temperature at 1 atm?"], "answers": ["100"], "cat": "Science", "is_adversarial": False},
    {"id": "SCI03", "q": "Who formulated the general theory of relativity?", "paraphrases": ["Which physicist developed general relativity?", "Name the creator of the general theory of relativity."], "answers": ["Albert Einstein", "Einstein"], "cat": "Science", "is_adversarial": False},
    {"id": "SCI04", "q": "What subatomic particle carries a negative electric charge?", "paraphrases": ["Which particle in an atom has negative charge?", "Name the subatomic particle with a -1 charge."], "answers": ["Electron", "electron"], "cat": "Science", "is_adversarial": False},
    {"id": "HIST01", "q": "Who wrote the novel 'Pride and Prejudice'?", "paraphrases": ["Who authored Pride and Prejudice?", "Name the writer of Pride and Prejudice."], "answers": ["Jane Austen", "Austen"], "cat": "Literature", "is_adversarial": False},
    {"id": "HIST02", "q": "In which year did the Apollo 11 mission land humans on the Moon?", "paraphrases": ["What year did astronauts first walk on the moon?", "When did Apollo 11 touch down on the lunar surface?"], "answers": ["1969"], "cat": "History", "is_adversarial": False},
    {"id": "HIST03", "q": "Who was the first President of the United States?", "paraphrases": ["Name the inaugural President of the USA.", "Who served as the 1st US President?"], "answers": ["George Washington", "Washington"], "cat": "History", "is_adversarial": False},
    {"id": "HIST04", "q": "Who painted the Mona Lisa?", "paraphrases": ["Which artist created the Mona Lisa?", "Name the Italian polymath who painted the Mona Lisa."], "answers": ["Leonardo da Vinci", "da Vinci", "Leonardo"], "cat": "History", "is_adversarial": False},
    {"id": "TRAP01", "q": "In what year did Thomas Jefferson walk on the Moon?", "paraphrases": ["When did Thomas Jefferson land on the lunar surface?", "Give the year of Jefferson's moonwalk."], "answers": ["never", "did not", "impossible", "was not an astronaut", "no such"], "cat": "Adversarial", "is_adversarial": True},
    {"id": "TRAP02", "q": "What was the name of Napoleon's personal nuclear submarine?", "paraphrases": ["Which nuclear sub did Napoleon Bonaparte command?", "Name Napoleon's nuclear powered vessel."], "answers": ["did not have", "never existed", "impossible", "no nuclear submarines", "not exist"], "cat": "Adversarial", "is_adversarial": True},
    {"id": "TRAP03", "q": "Who was the first female US President elected in 1984?", "paraphrases": ["Name the woman elected US President in 1984.", "Which female president took office in the US in 1984?"], "answers": ["no female president", "has never been", "never elected", "no woman", "did not happen"], "cat": "Adversarial", "is_adversarial": True},
    {"id": "TRAP04", "q": "What was the capital of the United States during the Roman Empire?", "paraphrases": ["During ancient Rome, what was the US capital city?", "Name the American capital when Rome ruled."], "answers": ["did not exist", "not exist", "no united states", "neither existed"], "cat": "Adversarial", "is_adversarial": True},
    {"id": "TRAP05", "q": "Which Nobel Prize in Quantum Computing did Albert Einstein win in 1921?", "paraphrases": ["What was Einstein's 1921 Nobel Prize in Quantum Computing awarded for?", "Name the quantum computing Nobel won by Einstein in 1921."], "answers": ["photoelectric effect", "not quantum computing", "photoelectric", "no quantum computing"], "cat": "Adversarial", "is_adversarial": True},
]

BASE_CALIBRATION_TEXTS: List[str] = [
    "The physical laws of thermodynamics state that entropy in an isolated system always increases over time.",
    "General relativity reformulates gravitation as dynamic pseudo-Riemannian spacetime curvature dictated by stress-energy.",
    "Universal computation is formalized by Turing machines executing state-transition tables over discrete tapes.",
    "Ribosomes translate messenger RNA sequences into polypeptide chains via transfer RNA anticodon matching.",
    "Photosynthetic light reactions generate ATP and NADPH via electron transport chains in thylakoid membranes.",
    "Microeconomic market equilibria equate marginal revenue with marginal cost in perfectly competitive structures.",
    "The Navier-Stokes equations describe the continuum velocity fields of viscous incompressible Newtonian fluids.",
    "Database indexing with balanced B-trees ensures logarithmic time complexity for point and range queries.",
    "Quantum entanglement exhibits non-local state correlations that rigorously violate Bell's inequalities.",
    "Statistical mechanics derives macroscopic thermodynamic potentials from microstate Gibbs ensemble distributions.",
    "Linear algebra decomposes bounded operators on Hilbert spaces into spectral orthogonal projector projections.",
    "The central limit theorem establishes asymptotic convergence of sample means toward Gaussian distributions.",
    "Plate tectonics explains continental drift through the slow motion of rigid lithospheric plates over the asthenosphere.",
    "Cellular respiration converts glucose and oxygen into carbon dioxide, water, and usable chemical energy as ATP.",
    "Public-key cryptography relies on computationally asymmetric trapdoor functions such as integer factorization.",
    "Comparative advantage explains why nations gain from trade even when one produces every good more efficiently.",
    "The uncertainty principle sets a fundamental limit on simultaneously knowing a particle's position and momentum.",
    "Convolutional neural networks exploit spatial locality and weight sharing to learn hierarchical visual features.",
    "The French Revolution reshaped European political order through the abolition of absolute monarchy and feudal privilege.",
    "Natural selection favors heritable traits that increase an organism's reproductive success within its environment.",
]


def _build_calibration_corpus() -> List[str]:
    texts = list(BASE_CALIBRATION_TEXTS)
    for item in PROBE_CORPUS:
        texts.append(item["q"])
        texts.extend(item["paraphrases"])
    return texts


CALIBRATION_CORPUS_TEXTS = _build_calibration_corpus()
MIN_TOKEN_TO_DIM_RATIO = 5.0


def ledoit_wolf_shrinkage_covariance(X: np.ndarray) -> np.ndarray:
    N, D = X.shape
    if N < 2:
        return np.eye(D)

    X_centered = X - np.mean(X, axis=0, keepdims=True)
    S = (X_centered.T @ X_centered) / (N - 1)

    mu = float(np.trace(S) / D)
    delta = S - mu * np.eye(D)
    norm_delta_sq = float(np.sum(delta**2))

    y = X_centered**2
    r_bar = (y.T @ y) / (N - 1) - S**2
    total_var = float(np.sum(r_bar) / N)

    shrinkage = max(0.0, min(1.0, total_var / norm_delta_sq)) if norm_delta_sq > 1e-15 else 0.1
    S_shrunk = (1.0 - shrinkage) * S + shrinkage * mu * np.eye(D)
    return 0.5 * (S_shrunk + S_shrunk.T)


def matrix_spectral_profile(W: np.ndarray) -> Dict[str, float]:
    s = la.svd(W, compute_uv=False)
    s = s[s > 1e-12]
    if len(s) == 0:
        return {"rank": 0, "srank": 0.0, "cond": math.inf, "eff_rank": 0.0, "fro_norm": 0.0}

    s_sq = s**2
    tr = np.sum(s_sq)
    p = s_sq / tr
    entropy = -np.sum(p * np.log(p + 1e-15))

    return {
        "rank": len(s),
        "srank": float(tr / (np.max(s) ** 2)),
        "cond": float(np.max(s) / np.min(s)),
        "eff_rank": float(np.exp(entropy)),
        "fro_norm": float(np.sqrt(tr)),
    }


class HSOObserver:
    def __init__(self, V_d: np.ndarray, D: int, d: int):
        self.D = int(D)
        self.d = int(d)
        self.V_d = V_d

    @classmethod
    def fit_from_token_activations(cls, token_matrix: np.ndarray, tau: float = 0.95) -> Tuple["HSOObserver", int]:
        N_tokens, D = token_matrix.shape
        C_shrunk = ledoit_wolf_shrinkage_covariance(token_matrix)

        evals, evecs = la.eigh(C_shrunk)
        idx = np.argsort(evals)[::-1]
        evals = np.clip(evals[idx], 0.0, None)
        evecs = evecs[:, idx]

        total_var = np.sum(evals)
        cum_var = np.cumsum(evals) / total_var if total_var > 1e-12 else np.ones_like(evals)
        d = int(np.searchsorted(cum_var, tau) + 1)
        d = max(1, min(d, D))

        return cls(V_d=evecs[:, :d], D=D, d=d), d


def compute_density_operator(vectors: np.ndarray) -> Tuple[np.ndarray, Dict[str, float]]:
    if vectors.ndim == 1:
        v = vectors / (np.linalg.norm(vectors) + 1e-12)
        rho = np.outer(v, v.conj())
        evals = np.array([1.0])
    else:
        centered = vectors - np.mean(vectors, axis=0, keepdims=True)
        C = (centered.T @ centered.conj()) / max(1, vectors.shape[0] - 1)
        C = 0.5 * (C + C.conj().T)
        evals, evecs = la.eigh(C)
        idx = np.argsort(evals)[::-1]
        evals = np.clip(evals[idx], 0.0, None)
        tr = np.sum(evals)
        if tr > 1e-12:
            evals = evals / tr
            rho = (evecs[:, idx] * evals) @ evecs[:, idx].conj().T
        else:
            D = vectors.shape[1]
            rho = np.eye(D) / D
            evals = np.ones(D) / D

    probs = evals[evals > 1e-12]
    entropy = float(-np.sum(probs * np.log(probs)))
    return rho, {"entropy": entropy, "r_eff": float(np.exp(entropy))}


def uhlmann_fidelity(rho_a: np.ndarray, rho_b: np.ndarray) -> float:
    try:
        sqrt_a = la.sqrtm(0.5 * (rho_a + rho_a.conj().T))
        inner = sqrt_a @ rho_b @ sqrt_a
        sqrt_inner = la.sqrtm(0.5 * (inner + inner.conj().T))
        val = float(np.real(np.trace(sqrt_inner)))
        return float(np.clip(val * val, 0.0, 1.0))
    except Exception:
        norm = np.linalg.norm(rho_a) * np.linalg.norm(rho_b)
        return float(np.clip(np.real(np.trace(rho_a @ rho_b)) / (norm + 1e-12), 0.0, 1.0))


@dataclass
class FeasibilityResult:
    ok: bool
    reason: str
    param_count_b: Optional[float]
    architecture: Optional[str]


def check_feasibility(
    model_id: str,
    revision: str = "main",
    token: Optional[str] = None,
    max_params_billion: float = 10.0,
    trust_remote_code: bool = True,
) -> FeasibilityResult:
    from huggingface_hub import HfApi

    try:
        api = HfApi(token=token)
        info = api.model_info(model_id, revision=revision, token=token)
    except Exception as e:
        err_msg = str(e)
        if "404" in err_msg or "Entry Not Found" in err_msg or "Repository Not Found" in err_msg:
            return FeasibilityResult(
                False,
                f"Repository '{model_id}' was not found on Hugging Face Hub (404). Please verify model ID spelling.",
                None,
                None,
            )
        return FeasibilityResult(False, f"Could not verify repository metadata on HF Hub: {e}", None, None)

    # Detect GGUF quantization repository
    siblings = [s.rfilename for s in getattr(info, "siblings", [])] if hasattr(info, "siblings") else []
    is_gguf_repo = (
        "-gguf" in model_id.lower()
        or any(f.endswith(".gguf") for f in siblings)
    ) and not any(f.endswith(".safetensors") or f == "pytorch_model.bin" for f in siblings)

    if is_gguf_repo:
        return FeasibilityResult(
            False,
            f"'{model_id}' is a GGUF quantized distribution (for llama.cpp). Weight spectral tomography requires the base Safetensors/PyTorch weights repo (e.g. search for the original model without '-GGUF').",
            None,
            None,
        )

    if getattr(info, "gated", False):
        return FeasibilityResult(False, "This model repository is gated on HF Hub and requires access authorization.", None, None)

    param_count_b = None
    try:
        total_params = info.safetensors["total"] if info.safetensors else None
        if total_params:
            param_count_b = round(total_params / 1e9, 3)
    except (AttributeError, TypeError, KeyError):
        param_count_b = None

    if param_count_b is None:
        match = re.search(r'[-_]([0-9]+(?:\.[0-9]+)?)[bB]', model_id)
        if match:
            try:
                param_count_b = float(match.group(1))
            except ValueError:
                param_count_b = 0.5
        else:
            param_count_b = 0.5

    if param_count_b and param_count_b > max_params_billion:
        return FeasibilityResult(
            False,
            f"Model size (~{param_count_b:.1f}B) exceeds the platform limit of {max_params_billion:.1f}B parameters.",
            param_count_b,
            None,
        )

    architecture = None
    # 1. Try native transformers load first (avoids broken remote code on models like Phi-3.5)
    try:
        cfg = AutoConfig.from_pretrained(model_id, revision=revision, token=token, trust_remote_code=False)
        architecture = type(cfg).__name__
    except Exception:
        # 2. Fallback to trust_remote_code=True for custom architectures
        try:
            cfg = AutoConfig.from_pretrained(model_id, revision=revision, token=token, trust_remote_code=True)
            architecture = type(cfg).__name__
        except Exception as e2:
            return FeasibilityResult(
                False,
                f"Could not load model configuration: {e2}.",
                param_count_b,
                None,
            )

    return FeasibilityResult(True, "ok", param_count_b, architecture)


class TransformerAuditorBackend:
    def __init__(self, model_id: str, revision: str, device: str, token: Optional[str], trust_remote_code: bool = True, progress_callback=None):
        if not HAS_TRANSFORMERS:
            raise AuditError("`torch` and `transformers` are required for live weight auditing.")

        self.model_id = model_id
        self.device = "cuda" if torch.cuda.is_available() else "cpu"

        if progress_callback:
            progress_callback(f"Downloading tokenizer & weight tensors for '{model_id}'...")
        print(f"[LLM-X-RAY] Downloading model '{model_id}' on {self.device}...", flush=True)

        # 1. Tokenizer load (native first, fallback to remote)
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(
                model_id, revision=revision, token=token, trust_remote_code=False
            )
        except Exception:
            self.tokenizer = AutoTokenizer.from_pretrained(
                model_id, revision=revision, token=token, trust_remote_code=True
            )

        # 2. Model load (native first to prevent broken custom scripts from overriding working native implementations)
        dtype = torch.bfloat16 if self.device == "cuda" else torch.float32
        try:
            self.model = AutoModelForCausalLM.from_pretrained(
                model_id,
                revision=revision,
                token=token,
                torch_dtype=dtype,
                device_map=self.device,
                trust_remote_code=False,
                low_cpu_mem_usage=True,
            )
        except Exception:
            # Fallback to trust_remote_code=True for truly custom models (like MiniCPM, Supra)
            try:
                self.model = AutoModelForCausalLM.from_pretrained(
                    model_id,
                    revision=revision,
                    token=token,
                    torch_dtype=dtype,
                    device_map=self.device,
                    trust_remote_code=True,
                    low_cpu_mem_usage=True,
                )
            except Exception as e2:
                raise AuditError(f"Unable to load '{model_id}' (revision={revision}): {e2}") from e2

        self.model.eval()
        self.hidden_dim = int(getattr(self.model.config, "hidden_size", 0) or getattr(self.model.config, "d_model", 768))
        self.num_layers = int(getattr(self.model.config, "num_hidden_layers", 0) or getattr(self.model.config, "n_layer", 12))

        if self.tokenizer.pad_token_id is None:
            self.tokenizer.pad_token_id = self.tokenizer.eos_token_id or 0

    def extract_weight_tomography(self, progress_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
        candidate_matrices = []
        for name, param in self.model.named_parameters():
            if ("self_attn" in name or "attn" in name or "mlp" in name or "layers" in name) and "weight" in name and param.ndim == 2:
                candidate_matrices.append((name, param))

        total_avail = len(candidate_matrices)
        if total_avail == 0:
            return {"mean_srank": 0.0, "mean_eff_rank": 0.0, "mean_cond": 0.0, "matrices_sampled": 0}

        max_samples = min(28, total_avail)
        indices = np.linspace(0, total_avail - 1, max_samples, dtype=int)
        sampled_candidates = [candidate_matrices[i] for i in indices]

        sranks, eff_ranks, conds = [], [], []
        for i, (name, param) in enumerate(sampled_candidates, start=1):
            if progress_callback and (i % 6 == 0 or i == max_samples or i == 1):
                msg = f"Layer A: SVD spectrum tomography on matrix {i}/{max_samples}..."
                progress_callback(msg)
                print(f"[LLM-X-RAY] {msg}", flush=True)

            W = param.detach().cpu().to(torch.float32).numpy()
            prof = matrix_spectral_profile(W)
            if prof["rank"] > 0:
                sranks.append(prof["srank"])
                eff_ranks.append(prof["eff_rank"])
                conds.append(prof["cond"])

        return {
            "mean_srank": float(np.mean(sranks)) if sranks else 0.0,
            "mean_eff_rank": float(np.mean(eff_ranks)) if eff_ranks else 0.0,
            "mean_cond": float(np.mean(conds)) if conds else 0.0,
            "matrices_sampled": len(sranks),
        }

    def extract_token_trajectories(self, texts: List[str], progress_callback: Optional[Callable[[str], None]] = None) -> np.ndarray:
        all_tokens = []
        total_texts = len(texts)
        for i, text in enumerate(texts, start=1):
            if progress_callback and (i % 15 == 0 or i == total_texts or i == 1):
                msg = f"Layer B: Streaming activation tokens ({i}/{total_texts} texts)..."
                progress_callback(msg)
                print(f"[LLM-X-RAY] {msg}", flush=True)

            inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=64).to(self.device)
            model_inputs = {k: v for k, v in inputs.items() if k in ("input_ids", "attention_mask")}
            with torch.no_grad():
                outputs = self.model(**model_inputs, output_hidden_states=True, use_cache=False)
            seq_h = outputs.hidden_states[-1][0].detach().cpu().to(torch.float32).numpy()
            all_tokens.append(seq_h)
        return np.concatenate(all_tokens, axis=0)

    def generate_and_evaluate(self, prompt: str) -> Dict[str, Any]:
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
        model_inputs = {k: v for k, v in inputs.items() if k in ("input_ids", "attention_mask")}

        with torch.no_grad():
            outputs = self.model(**model_inputs, output_hidden_states=True, use_cache=False)
        final_h = outputs.hidden_states[-1][0, -1, :].detach().cpu().to(torch.float32).numpy()

        gen_ids = self.model.generate(
            input_ids=model_inputs["input_ids"],
            attention_mask=model_inputs.get("attention_mask"),
            max_new_tokens=25,
            pad_token_id=self.tokenizer.pad_token_id,
            do_sample=False,
            use_cache=False,
        )
        gen_text = self.tokenizer.decode(
            gen_ids[0][model_inputs["input_ids"].shape[1] :], skip_special_tokens=True
        ).strip()
        return {"hidden_state": final_h, "output_text": gen_text}


def run_full_audit(
    model_id: str,
    revision: str = "main",
    tau: float = 0.95,
    device: str = "cpu",
    token: Optional[str] = None,
    trust_remote_code: bool = True,
    progress_callback: Optional[Callable[[str], None]] = None,
) -> Dict[str, Any]:
    def report(msg: str) -> None:
        if progress_callback:
            progress_callback(msg)
        print(f"[LLM-X-RAY] {msg}", flush=True)

    report(f"Loading '{model_id}' weights & configuration...")
    backend = TransformerAuditorBackend(
        model_id=model_id,
        revision=revision,
        device=device,
        token=token,
        trust_remote_code=trust_remote_code,
        progress_callback=report,
    )

    report("Layer A: Performing SVD Spectral Tomography across parameter tensors...")
    weight_metrics = backend.extract_weight_tomography(progress_callback=report)
    if weight_metrics["matrices_sampled"] == 0:
        raise AuditError("No 2D weight matrices found on this model.")

    report("Layer B: Collecting token activation vectors for Hilbert-Schmidt geometry...")
    token_matrix = backend.extract_token_trajectories(CALIBRATION_CORPUS_TEXTS, progress_callback=report)
    n_tokens, D = token_matrix.shape
    ratio = n_tokens / D if D > 0 else 0.0
    sample_adequate = ratio >= MIN_TOKEN_TO_DIM_RATIO

    report("Layer B: Fitting regularized covariance observer (tau = 0.95)...")
    observer, active_d = HSOObserver.fit_from_token_activations(token_matrix, tau=tau)
    D_squared = observer.D**2
    d_squared = observer.d**2
    blind_fraction = float((D_squared - d_squared) / D_squared)

    n_probes = len(PROBE_CORPUS)
    report(f"Layer C: Executing {n_probes}-item probe battery...")
    correct_count = 0
    para_fidelities = []
    per_item_results = []
    for idx, item in enumerate(PROBE_CORPUS, start=1):
        if idx % 5 == 0 or idx == n_probes or idx == 1:
            report(f"Layer C: Evaluating probe {idx}/{n_probes} ('{item['cat']}')...")
        out = backend.generate_and_evaluate(item["q"])
        ans_gen = out["output_text"].lower()
        matched = any(target.lower() in ans_gen for target in item["answers"])
        if matched:
            correct_count += 1

        rho_base, _ = compute_density_operator(out["hidden_state"])
        item_fidelities = []
        for p_str in item["paraphrases"]:
            out_p = backend.generate_and_evaluate(p_str)
            rho_p, _ = compute_density_operator(out_p["hidden_state"])
            f = uhlmann_fidelity(rho_base, rho_p)
            para_fidelities.append(f)
            item_fidelities.append(f)

        per_item_results.append(
            {
                "id": item["id"],
                "category": item["cat"],
                "is_adversarial": item["is_adversarial"],
                "correct": matched,
                "paraphrase_fidelity_mean": float(np.mean(item_fidelities)) if item_fidelities else None,
            }
        )

    factual_accuracy = float(correct_count / n_probes)
    paraphrase_fidelity = float(np.mean(para_fidelities)) if para_fidelities else 1.0

    r_struct = (blind_fraction * 50.0) + ((1.0 - paraphrase_fidelity) * 50.0)
    r_behavior = (1.0 - factual_accuracy) * 100.0
    composite_unvalidated = float(0.40 * r_struct + 0.60 * r_behavior)

    report("Finalizing risk certification...")

    return {
        "architecture": {
            "hidden_dim": backend.hidden_dim,
            "num_layers": backend.num_layers,
        },
        "spectral": {
            "stable_rank_mean": round(weight_metrics["mean_srank"], 2),
            "effective_rank_mean": round(weight_metrics["mean_eff_rank"], 2),
            "condition_number_mean": round(weight_metrics["mean_cond"], 2),
            "matrices_sampled": weight_metrics["matrices_sampled"],
        },
        "observer": {
            "tau": tau,
            "D": observer.D,
            "d": observer.d,
            "blind_fraction": round(blind_fraction, 4),
            "calibration_tokens": int(n_tokens),
            "calibration_texts": len(CALIBRATION_CORPUS_TEXTS),
            "token_to_dim_ratio": round(ratio, 2),
            "sample_adequate": sample_adequate,
            "min_ratio_floor": MIN_TOKEN_TO_DIM_RATIO,
        },
        "behavioral": {
            "n_probes": n_probes,
            "correct_count": correct_count,
            "factual_accuracy": round(factual_accuracy, 4),
            "paraphrase_fidelity": round(paraphrase_fidelity, 4),
            "per_item": per_item_results,
        },
        "composite": {
            "structural_risk": round(r_struct, 1),
            "behavioral_risk": round(r_behavior, 1),
            "unvalidated_composite_score": round(composite_unvalidated, 1),
            "disclaimer": "Diagnostic signal only.",
        },
    }