runtime/wave_reranker.py

"""Classical, quantum-inspired path-field reranker for HSCM.

This module does not claim physical quantum state.  It represents competing
HSCM paths as normalized complex amplitudes, mixes them through a bounded
similarity kernel, and measures the resulting intensities.  The construction is
phase-sensitive, norm-normalized, deterministic, and NumPy-only at runtime.
"""
from __future__ import annotations

import json
import hashlib
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Sequence

import numpy as np


WAVE_ARTIFACT_SCHEMA_VERSION = 1
WAVE_DEPLOYMENT_SCHEMA_VERSION = 1
WAVE_FEATURE_ORDER = (
    "semantic_support",
    "lexical_support",
    "coherence",
    "path_quality",
)
WAVE_AUX_ORDER = (
    "phase_residual",
    "holonomy",
    "relative_weight",
    "state_amplitude",
)


def _vector(values: Sequence[float], size: int, label: str) -> np.ndarray:
    result = np.asarray(values, dtype=np.float64)
    if result.shape != (size,) or not np.all(np.isfinite(result)):
        raise ValueError(f"{label} must contain {size} finite values")
    return result


def _bounded(value: float, lower: float, upper: float, label: str) -> float:
    result = float(value)
    if not math.isfinite(result) or not lower <= result <= upper:
        raise ValueError(f"{label} must be finite and within [{lower}, {upper}]")
    return result


@dataclass(frozen=True)
class WaveRerankerArtifact:
    artifact_id: str
    mode: str
    input_weights: tuple[float, ...]
    logit_scale: float
    kernel_log_weights: tuple[float, ...]
    residual_gain: float
    holonomy_gain: float
    mixing: float
    decoherence: float
    fusion_weight: float = 0.10
    training_stage: str = "experimental"
    metadata: Mapping | None = None

    def __post_init__(self) -> None:
        if not str(self.artifact_id).strip():
            raise ValueError("wave artifact_id must not be empty")
        if self.mode not in {"wave", "real-control"}:
            raise ValueError("wave artifact mode must be wave or real-control")
        _vector(self.input_weights, 6, "input_weights")
        _vector(self.kernel_log_weights, 4, "kernel_log_weights")
        _bounded(self.logit_scale, 0.05, 10.0, "logit_scale")
        for value, label in ((self.residual_gain, "residual_gain"),
                             (self.holonomy_gain, "holonomy_gain")):
            if not math.isfinite(float(value)):
                raise ValueError(f"{label} must be finite")
        _bounded(self.mixing, 0.0, 0.75, "mixing")
        _bounded(self.decoherence, 0.0, 1.0, "decoherence")
        _bounded(self.fusion_weight, 0.0, 0.25, "fusion_weight")
        try:
            json.dumps(dict(self.metadata or {}), allow_nan=False)
        except (TypeError, ValueError) as exc:
            raise ValueError("wave metadata must be finite JSON data") from exc

    def to_mapping(self) -> dict:
        return {
            "schema_version": WAVE_ARTIFACT_SCHEMA_VERSION,
            "artifact_id": self.artifact_id,
            "mode": self.mode,
            "feature_order": list(WAVE_FEATURE_ORDER),
            "aux_order": list(WAVE_AUX_ORDER),
            "input_weights": [float(value) for value in self.input_weights],
            "logit_scale": float(self.logit_scale),
            "kernel_log_weights": [
                float(value) for value in self.kernel_log_weights],
            "phase_gains": {
                "residual": float(self.residual_gain),
                "holonomy": float(self.holonomy_gain),
            },
            "mixing": float(self.mixing),
            "decoherence": float(self.decoherence),
            "fusion_weight": float(self.fusion_weight),
            "training_stage": self.training_stage,
            "metadata": dict(self.metadata or {}),
        }

    @classmethod
    def from_mapping(cls, payload: Mapping) -> "WaveRerankerArtifact":
        if int(payload.get("schema_version", -1)) != WAVE_ARTIFACT_SCHEMA_VERSION:
            raise ValueError("unsupported wave artifact schema")
        if tuple(payload.get("feature_order", ())) != WAVE_FEATURE_ORDER:
            raise ValueError("wave feature order mismatch")
        if tuple(payload.get("aux_order", ())) != WAVE_AUX_ORDER:
            raise ValueError("wave auxiliary order mismatch")
        gains = payload.get("phase_gains") or {}
        return cls(
            artifact_id=str(payload.get("artifact_id", "")),
            mode=str(payload.get("mode", "")),
            input_weights=tuple(float(value) for value in
                                payload.get("input_weights", ())),
            logit_scale=float(payload.get("logit_scale", 1.0)),
            kernel_log_weights=tuple(float(value) for value in
                                     payload.get("kernel_log_weights", ())),
            residual_gain=float(gains.get("residual", 0.0)),
            holonomy_gain=float(gains.get("holonomy", 0.0)),
            mixing=float(payload.get("mixing", 0.0)),
            decoherence=float(payload.get("decoherence", 0.0)),
            fusion_weight=float(payload.get("fusion_weight", 0.10)),
            training_stage=str(payload.get("training_stage", "experimental")),
            metadata=dict(payload.get("metadata") or {}),
        )


class QuantumInspiredWaveReranker:
    """Measure a normalized phase-sensitive field over one candidate set."""

    def __init__(self, artifact: WaveRerankerArtifact):
        self.artifact = artifact
        self.artifact_id = artifact.artifact_id
        self.fusion_weight = float(artifact.fusion_weight)
        metadata = dict(artifact.metadata or {})
        self.nested_wave_alpha = float(
            metadata.get("nested_wave_alpha", 0.0))
        self.nested_shortlist = int(metadata.get("nested_shortlist", 0))
        if not 0.0 <= self.nested_wave_alpha <= 1.0:
            raise ValueError("nested_wave_alpha must be within [0, 1]")
        if self.nested_shortlist < 0:
            raise ValueError("nested_shortlist must be non-negative")

    @staticmethod
    def _inputs(features: np.ndarray, auxiliary: np.ndarray
                ) -> tuple[np.ndarray, np.ndarray]:
        values = np.asarray(features, dtype=np.float64)
        aux = np.asarray(auxiliary, dtype=np.float64)
        if values.ndim != 2 or values.shape[1] != 4:
            raise ValueError("wave feature batch must have shape (n, 4)")
        if aux.shape != (values.shape[0], 4):
            raise ValueError("wave auxiliary batch must have shape (n, 4)")
        if not np.all(np.isfinite(values)) or not np.all(np.isfinite(aux)):
            raise ValueError("wave inputs contain non-finite values")
        if np.any(aux[:, 2] < 0.0) or np.any(aux[:, 3] <= 0.0):
            raise ValueError("relative weights and state amplitudes are invalid")
        return values, aux

    def score_paths(self, features: np.ndarray, auxiliary: np.ndarray) -> np.ndarray:
        values, aux = self._inputs(features, auxiliary)
        count = len(values)
        if count == 0:
            return np.zeros(0, dtype=np.float64)
        relative = aux[:, 2]
        relative = relative - float(np.mean(relative))
        depth = np.clip(np.log(np.maximum(aux[:, 3], 0.05) / 0.5), -1.0, 1.0)
        model_inputs = np.column_stack((values, relative, depth))
        logits = ((model_inputs @ np.asarray(
            self.artifact.input_weights, dtype=np.float64))
                  * float(self.artifact.logit_scale))

        residual_signal = np.sin(aux[:, 0])
        holonomy_signal = np.sin(aux[:, 1])
        phase = (float(self.artifact.residual_gain) * residual_signal
                 + float(self.artifact.holonomy_gain) * holonomy_signal)
        if self.artifact.mode == "real-control":
            # Same observations and parameter count, but phase is consumed as an
            # ordinary real logit rather than through complex interference.
            logits = logits + phase
            phase = np.zeros(count, dtype=np.float64)

        logits = logits - float(np.max(logits))
        base_probability = np.exp(np.clip(logits, -60.0, 0.0))
        base_probability /= max(float(np.sum(base_probability)), 1e-12)
        magnitude = np.sqrt(base_probability)
        state = magnitude * np.exp(1j * phase)

        kernel_scale = np.exp(np.clip(np.asarray(
            self.artifact.kernel_log_weights, dtype=np.float64), -6.0, 6.0))
        kernel_values = values * kernel_scale
        norms = np.linalg.norm(kernel_values, axis=1, keepdims=True)
        normalized = kernel_values / np.maximum(norms, 1e-12)
        kernel = np.clip(normalized @ normalized.T, 0.0, 1.0)
        np.fill_diagonal(kernel, 0.0)
        row_sums = np.sum(kernel, axis=1, keepdims=True)
        kernel = np.divide(kernel, row_sums, out=np.zeros_like(kernel),
                           where=row_sums > 1e-12)

        mixing = float(self.artifact.mixing)
        evolved = (1.0 - mixing) * state + mixing * (kernel @ state)
        coherent = np.abs(evolved) ** 2
        incoherent = ((1.0 - mixing) * base_probability
                      + mixing * (kernel @ base_probability))
        measured = ((1.0 - float(self.artifact.decoherence)) * coherent
                    + float(self.artifact.decoherence) * incoherent)
        measured = np.maximum(np.asarray(measured, dtype=np.float64), 0.0)
        measured /= max(float(np.sum(measured)), 1e-12)
        if not np.all(np.isfinite(measured)):
            raise ValueError("wave measurement produced non-finite values")
        return measured


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        for block in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def verify_active_wave_deployment(
        artifact_path: str | Path,
        deployment_path: str | Path | None = None) -> dict:
    """Verify the exact allow-list for the protected phase controller.

    Active phase use is intentionally narrower than general wave reranking: it
    may only reorder an already-ranked scalar shortlist, it may not participate
    in evidence admission, and the deployment must record the manual override
    of the statistically inconclusive end-to-end result.
    """
    artifact = Path(artifact_path)
    deployment = (Path(deployment_path) if deployment_path is not None
                  else artifact.with_suffix(".deployment.json"))
    payload = json.loads(deployment.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("wave deployment root must be an object")
    if int(payload.get("schema_version", -1)) != WAVE_DEPLOYMENT_SCHEMA_VERSION:
        raise ValueError("unsupported wave deployment schema")
    if payload.get("status") != "active-protected":
        raise ValueError("wave deployment is not active-protected")
    if payload.get("rollback_mode") != "shadow-observer":
        raise ValueError(
            "active wave deployment must declare shadow-observer rollback")
    if payload.get("artifact_sha256") != _sha256(artifact):
        raise ValueError("wave deployment artifact hash mismatch")
    gates = payload.get("activation_gates")
    if (not isinstance(gates, dict) or not gates
            or any(value is not True for value in gates.values())):
        raise ValueError("wave deployment activation gates are not all passing")
    override = payload.get("operator_override")
    if (not isinstance(override, dict)
            or override.get("authorized") is not True
            or override.get("statistically_conclusive") is not False):
        raise ValueError(
            "active wave deployment must record the inconclusive operator override")
    safety = payload.get("safety_invariants")
    required_safety = {
        "protected_scalar_shortlist": True,
        "unrestricted_wave_ranker": False,
        "can_admit_evidence": False,
        "telemetry_contains_text": False,
    }
    if (not isinstance(safety, dict)
            or any(safety.get(key) is not value
                   for key, value in required_safety.items())):
        raise ValueError("wave deployment safety invariants are invalid")
    return payload


def load_wave_reranker(
        path: str | Path, *, require_active: bool = False,
        deployment_path: str | Path | None = None
        ) -> QuantumInspiredWaveReranker:
    """Load a wave artifact, optionally requiring protected active approval."""
    artifact_path = Path(path)
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("wave artifact root must be an object")
    reranker = QuantumInspiredWaveReranker(
        WaveRerankerArtifact.from_mapping(payload))
    if require_active:
        deployment = verify_active_wave_deployment(
            artifact_path, deployment_path)
        if deployment.get("artifact_id") != reranker.artifact_id:
            raise ValueError("wave deployment artifact id mismatch")
        if float(deployment.get("nested_wave_alpha", -1.0)) != (
                reranker.nested_wave_alpha):
            raise ValueError("wave deployment nested alpha mismatch")
        if int(deployment.get("nested_shortlist", -1)) != (
                reranker.nested_shortlist):
            raise ValueError("wave deployment nested shortlist mismatch")
        if reranker.nested_wave_alpha <= 0.0 or reranker.nested_shortlist <= 0:
            raise ValueError("active wave artifact has no protected nested controller")
    return reranker