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