gam_terms/

construction.rs

use crate::EstimationError;
use crate::basis::analyze_penalty_block;
use crate::smooth::PenaltyStructureHint;
use faer::linalg::matmul::matmul;
use faer::{Accum, Mat, MatRef, Par, Side};
use gam_linalg::faer_ndarray::{FaerLinalgError, FaerQr, FaerSvd};
use gam_linalg::matrix::symmetrize_in_place;
use gam_linalg::utils::KahanSum;
use ndarray::{Array1, Array2, ArrayView1, ArrayView2, ArrayViewMut2, s};
use rayon::iter::{
    IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
};
use std::collections::{BTreeMap, HashSet};
use std::ops::Range;
use std::sync::Arc;

/// Relative "numerically-PSD" floor of the symmetric eigensolver: an eigenvalue
/// smaller than this fraction of the spectrum scale is roundoff, not genuine
/// curvature (~`sqrt(machine ε)`; #1619). This single floor governs BOTH which
/// eigenvalues are snapped to zero (declared null) in `classify_eigenvalues_strict`
/// AND how much per-mode relative energy the transformed penalty root may retain
/// in the resulting null block (`subspace_split_is_consistent`). Keeping the two
/// tied to one constant makes the null-space definition and the leakage-consistency
/// guard provably mutually consistent, rather than the guard demanding a precision
/// the classifier never promised.
const REL_PSD_FLOOR: f64 = 1.0e-8;

#[derive(Clone)]
pub enum PenaltyRepresentation {
    Dense(Array2<f64>),
    Banded {
        bands: Vec<Array1<f64>>,
        offsets: Vec<i32>,
    },
    Kronecker {
        /// Full penalty-block Kronecker product `left ⊗ right`: each entry of
        /// `left` scales an entire copy of `right` in the dense expansion.
        ///
        /// This is distinct from chunked kernel design assembly, where center
        /// rows are kernel-evaluation arguments rather than matrix factors.
        left: Array2<f64>,
        right: Array2<f64>,
    },
}

impl PenaltyRepresentation {

}

#[derive(Clone)]
pub struct PenaltyMatrix {
    pub col_range: Range<usize>,
    pub representation: PenaltyRepresentation,
}

impl PenaltyMatrix {
    fn accumulate_into(&self, mut dest: ArrayViewMut2<'_, f64>, weight: f64) {
        if weight == 0.0 {
            return;
        }
        match &self.representation {
            PenaltyRepresentation::Dense(block) => {
                dest.scaled_add(weight, block);
            }
            PenaltyRepresentation::Banded { bands, offsets } => {
                let positive_offsets: HashSet<usize> = offsets
                    .iter()
                    .filter_map(|&off| (off >= 0).then_some(off as usize))
                    .collect();
                for (band, &offset) in bands.iter().zip(offsets.iter()) {
                    let off = offset.unsigned_abs() as usize;
                    if offset < 0 && positive_offsets.contains(&off) {
                        continue;
                    }
                    for (idx, &value) in band.iter().enumerate() {
                        let (i, j) = if offset >= 0 {
                            (idx, idx + off)
                        } else {
                            (idx + off, idx)
                        };
                        let Some(entry_ij) = dest.get_mut((i, j)) else {
                            continue;
                        };
                        *entry_ij += weight * value;
                        if i != j
                            && let Some(entry_ji) = dest.get_mut((j, i))
                        {
                            *entry_ji += weight * value;
                        }
                    }
                }
            }
            PenaltyRepresentation::Kronecker { left, right } => {
                let (lrows, l_cols) = left.dim();
                let (rrows, r_cols) = right.dim();
                for i in 0..lrows {
                    for j in 0..l_cols {
                        let scale = left[(i, j)] * weight;
                        if scale == 0.0 {
                            continue;
                        }
                        let mut block = dest.slice_mut(s![
                            i * rrows..(i + 1) * rrows,
                            j * r_cols..(j + 1) * r_cols
                        ]);
                        block.scaled_add(scale, right);
                    }
                }
            }
        }
    }

    pub fn to_dense(&self, total_dim: usize) -> Array2<f64> {
        let mut dense = Array2::<f64>::zeros((total_dim, total_dim));
        self.accumulate_into(
            dense.slice_mut(s![self.col_range.clone(), self.col_range.clone()]),
            1.0,
        );
        dense
    }
}

pub(crate) fn array_to_faer(array: &Array2<f64>) -> Mat<f64> {
    let (rows, cols) = array.dim();
    Mat::from_fn(rows, cols, |i, j| array[[i, j]])
}

pub(crate) fn mat_to_array(mat: &Mat<f64>) -> Array2<f64> {
    let mut out = Array2::<f64>::zeros((mat.nrows(), mat.ncols()));
    for i in 0..mat.nrows() {
        for j in 0..mat.ncols() {
            out[[i, j]] = mat[(i, j)];
        }
    }
    out
}

fn mat_max_abs_element(matrix: MatRef<'_, f64>) -> f64 {
    let (rows, cols) = matrix.shape();
    let mut maxval = 0.0_f64;
    for i in 0..rows {
        for j in 0..cols {
            let val = matrix[(i, j)];
            if val.is_finite() {
                maxval = maxval.max(val.abs());
            }
        }
    }
    maxval
}

fn sanitize_symmetric_faer(matrix: &Mat<f64>) -> Mat<f64> {
    let (rows, cols) = matrix.as_ref().shape();
    assert_eq!(rows, cols, "Matrix must be square for sanitization");

    let mut sanitized = matrix.clone();

    for i in 0..rows {
        let diag = sanitized[(i, i)];
        if !diag.is_finite() {
            sanitized[(i, i)] = 0.0;
        }
        for j in (i + 1)..cols {
            let mut upper = sanitized[(i, j)];
            let mut lower = sanitized[(j, i)];
            if !upper.is_finite() {
                upper = 0.0;
            }
            if !lower.is_finite() {
                lower = 0.0;
            }
            let avg = 0.5 * (upper + lower);
            sanitized[(i, j)] = avg;
            sanitized[(j, i)] = avg;
        }
    }

    let scale = mat_max_abs_element(sanitized.as_ref());
    let tiny = (scale * 1e-14).max(1e-30);
    for i in 0..rows {
        for j in 0..cols {
            let val = sanitized[(i, j)];
            if !val.is_finite() {
                sanitized[(i, j)] = 0.0;
            } else if val.abs() < tiny {
                sanitized[(i, j)] = 0.0;
            }
        }
    }

    sanitized
}

fn penalty_from_root_faer(root: &Mat<f64>) -> Mat<f64> {
    let cols = root.ncols();
    let mut full = Mat::<f64>::zeros(cols, cols);
    let root_ref = root.as_ref();
    let root_t = root_ref.transpose();
    matmul(
        full.as_mut(),
        Accum::Replace,
        root_t,
        root_ref,
        1.0,
        Par::Seq,
    );
    sanitize_symmetric_faer(&full)
}

fn symmetrize_faer_matrix_in_place(matrix: &mut Mat<f64>) {
    let n = matrix.nrows().min(matrix.ncols());
    for i in 0..n {
        for j in 0..i {
            let avg = 0.5 * (matrix[(i, j)] + matrix[(j, i)]);
            matrix[(i, j)] = avg;
            matrix[(j, i)] = avg;
        }
    }
}

fn orthogonal_similarity_transform_faer(
    matrix: &Mat<f64>,
    block_dim: usize,
    orthogonal: &Mat<f64>,
) -> Mat<f64> {
    let matrix_block = matrix.as_ref().submatrix(0, 0, block_dim, block_dim);
    let cols = orthogonal.ncols();
    let mut temp = Mat::<f64>::zeros(block_dim, cols);
    matmul(
        temp.as_mut(),
        Accum::Replace,
        matrix_block,
        orthogonal.as_ref(),
        1.0,
        Par::Seq,
    );
    let mut rotated = Mat::<f64>::zeros(cols, cols);
    matmul(
        rotated.as_mut(),
        Accum::Replace,
        orthogonal.transpose(),
        temp.as_ref(),
        1.0,
        Par::Seq,
    );
    symmetrize_faer_matrix_in_place(&mut rotated);
    rotated
}

fn trace_penalty_in_orthogonal_basis(
    matrix: &Mat<f64>,
    block_dim: usize,
    orthogonal: &Mat<f64>,
    rotated_eigenvalues: &[f64],
    delta: f64,
) -> f64 {
    let matrix_block = matrix.as_ref().submatrix(0, 0, block_dim, block_dim);
    let cols = orthogonal.ncols();
    assert!(rotated_eigenvalues.len() >= cols);
    let mut projected = Mat::<f64>::zeros(block_dim, cols);
    matmul(
        projected.as_mut(),
        Accum::Replace,
        matrix_block,
        orthogonal.as_ref(),
        1.0,
        Par::Seq,
    );
    let mut trace = KahanSum::default();
    for l in 0..cols {
        let mut diag_ll = KahanSum::default();
        for i in 0..block_dim {
            diag_ll.add(orthogonal[(i, l)] * projected[(i, l)]);
        }
        trace.add(diag_ll.sum() / (rotated_eigenvalues[l] + delta));
    }
    trace.sum()
}

pub fn trace_reduced_penalty_covariance(
    reduced_penalty: &Array2<f64>,
    covariance_basis: &Array2<f64>,
) -> f64 {
    assert_eq!(
        reduced_penalty.dim(),
        covariance_basis.dim(),
        "trace_reduced_penalty_covariance dimension mismatch"
    );
    let r = covariance_basis.nrows();
    let mut trace = KahanSum::default();
    for i in 0..r {
        for j in 0..r {
            trace.add(covariance_basis[[i, j]] * reduced_penalty[[j, i]]);
        }
    }
    trace.sum()
}

pub fn trace_penalty_covariance_in_orthogonal_basis(
    matrix: &Array2<f64>,
    orthogonal: &Array2<f64>,
    covariance_basis: &Array2<f64>,
) -> f64 {
    let reduced = gam_linalg::faer_ndarray::fast_ab(
        &gam_linalg::faer_ndarray::fast_atb(orthogonal, matrix),
        orthogonal,
    );
    trace_reduced_penalty_covariance(&reduced, covariance_basis)
}

/// Strict spectral classifier used as a final guard on penalty eigendecompositions.
///
/// Penalty matrices fed to the GAM solver are required to be PSD by construction.
/// This routine snaps roundoff-zero eigenvalues to exact zero, accepts strictly
/// positive eigenvalues, and rejects materially-indefinite or non-finite spectra
/// with a hard error rather than silently rewriting them. The previous behaviour
/// (mass-zeroing negative or non-finite eigenvalues) hid construction bugs and
/// changed the optimisation objective downstream.
///
/// The acceptance tolerance is the larger of a machine-ε floor
/// (`C_EPS_P_FACTOR * eps_machine * p * scale`, with `C_EPS_P_FACTOR = 64`
/// absorbing the rounding accumulated in a symmetric eigendecomposition of a
/// moderate-dimension matrix) and a relative "numerically PSD" floor
/// `REL_PSD_FLOOR * scale`. The latter dominates for large high-rank penalties
/// assembled / reparameterized at extreme λ, where roundoff produces
/// ~1e-11-relative negative eigenvalues that are PSD to any reasonable precision
/// yet exceeded the bare ~12×ε machine floor and spuriously failed the inner
/// P-IRLS solve (#1619). Genuine indefiniteness is O(1) relative and is still
/// rejected far above either floor.
fn classify_eigenvalues_strict(
    eigenvalues: &mut [f64],
    context: &str,
) -> Result<(), EstimationError> {
    const C_EPS_P_FACTOR: f64 = 64.0;
    // `REL_PSD_FLOOR` (module-level): the relative threshold below which a
    // (possibly slightly negative) eigenvalue is roundoff and is snapped to zero
    // rather than rejected. Shared with the subspace-leakage guard so the null
    // definition and the leakage tolerance stay mutually consistent.
    let p = eigenvalues.len();

    let mut scale = 0.0_f64;
    for (idx, &val) in eigenvalues.iter().enumerate() {
        if !val.is_finite() {
            return Err(EstimationError::PenaltySpectrumNonFinite {
                context: context.to_string(),
                index: idx,
                value: val,
            });
        }
        scale = scale.max(val.abs());
    }

    // p * eps captures the rounding floor of a symmetric eigendecomposition of a
    // p-dimensional matrix; multiplying by `scale` lifts the floor to the actual
    // magnitude of the spectrum. For large high-rank penalties assembled at
    // extreme λ this machine floor (~12×ε relative) is tighter than the roundoff
    // actually produced, so we take the larger of it and a relative numerically-PSD
    // floor `REL_PSD_FLOOR * scale` (#1619).
    let machine_floor = C_EPS_P_FACTOR * f64::EPSILON * (p.max(1) as f64) * scale;
    let tolerance = machine_floor
        .max(REL_PSD_FLOOR * scale)
        .max(f64::MIN_POSITIVE);

    for (idx, val) in eigenvalues.iter_mut().enumerate() {
        if val.abs() <= tolerance {
            *val = 0.0;
        } else if *val < 0.0 {
            return Err(EstimationError::PenaltySpectrumIndefinite {
                context: context.to_string(),
                index: idx,
                value: *val,
                tolerance,
                scale,
            });
        }
    }
    Ok(())
}

fn robust_eighwith_policy<M, V, E, Validate, Sanitize, EigCall, MapErr>(
    matrix: &M,
    context: &str,
    validate_input: Validate,
    sanitize: Sanitize,
    mut eig_call: EigCall,
    map_error: MapErr,
) -> Result<(Vec<f64>, V), EstimationError>
where
    Validate: Fn(&M, &str) -> Result<(), EstimationError>,
    Sanitize: Fn(&M) -> M,
    EigCall: FnMut(&M) -> Result<(Vec<f64>, V), E>,
    MapErr: Fn(E, &str) -> EstimationError,
{
    validate_input(matrix, context)?;

    // The sanitize step only enforces exact symmetry by averaging M and M^T and
    // zeros sub-eps noise; it never adds a diagonal ridge. Adding ridge changes
    // the matrix being decomposed, which silently changes the optimisation
    // objective downstream. If eigh genuinely fails on a finite symmetric input,
    // surface the error instead of mutating the spectrum.
    let candidate = sanitize(matrix);
    match eig_call(&candidate) {
        Ok((mut eigenvalues, eigenvectors)) => {
            classify_eigenvalues_strict(&mut eigenvalues, context)?;
            Ok((eigenvalues, eigenvectors))
        }
        Err(err) => Err(map_error(err, context)),
    }
}

pub(crate) fn robust_eigh_faer(
    matrix: &Mat<f64>,
    side: Side,
    context: &str,
) -> Result<(Vec<f64>, Mat<f64>), EstimationError> {
    robust_eighwith_policy(
        matrix,
        context,
        |mat, ctx| {
            let (rows, cols) = mat.as_ref().shape();
            for i in 0..rows {
                for j in 0..cols {
                    let val = mat[(i, j)];
                    if !val.is_finite() {
                        let max_abs = mat_max_abs_element(mat.as_ref());
                        crate::bail_invalid_estim!(
                            "{} contains non-finite entries (max finite magnitude {:.3e})",
                            ctx,
                            max_abs
                        );
                    }
                }
            }
            Ok(())
        },
        sanitize_symmetric_faer,
        |candidate| {
            let eig = candidate.as_ref().self_adjoint_eigen(side)?;
            let diag = eig.S();
            let mut eigenvalues = Vec::with_capacity(diag.dim());
            for idx in 0..diag.dim() {
                eigenvalues.push(diag[idx]);
            }

            let vectors_ref = eig.U();
            let mut eigenvectors = Mat::<f64>::zeros(vectors_ref.nrows(), vectors_ref.ncols());
            for i in 0..vectors_ref.nrows() {
                for j in 0..vectors_ref.ncols() {
                    eigenvectors[(i, j)] = vectors_ref[(i, j)];
                }
            }
            Ok((eigenvalues, eigenvectors))
        },
        |err, _| {
            EstimationError::EigendecompositionFailed(FaerLinalgError::SelfAdjointEigen(err))
        },
    )
}

fn robust_eigh(
    matrix: &Array2<f64>,
    side: Side,
    context: &str,
) -> Result<(Array1<f64>, Array2<f64>), EstimationError> {
    let matrix_faer = array_to_faer(matrix);
    let (eigenvalues, eigenvectors) = robust_eigh_faer(&matrix_faer, side, context)?;
    Ok((Array1::from_vec(eigenvalues), mat_to_array(&eigenvectors)))
}

pub(crate) fn kronecker_marginal_eigensystems(
    marginal_penalties: &[Array2<f64>],
    context: &str,
) -> Result<Vec<(Array1<f64>, Array2<f64>)>, EstimationError> {
    let mut eigensystems = Vec::with_capacity(marginal_penalties.len());
    for (k, penalty) in marginal_penalties.iter().enumerate() {
        eigensystems.push(robust_eigh(
            penalty,
            Side::Lower,
            &format!("{context} marginal {k}"),
        )?);
    }
    Ok(eigensystems)
}

#[derive(Debug, Clone, Copy)]
struct SubspaceLeakageMetrics {
    max_abs_sq: f64,
    max_rel_sq: f64,
    worst_penalty: usize,
    max_cross_gram_abs: f64,
}

fn assess_subspace_leakage(
    qs: &Mat<f64>,
    rs_transformed: &[Mat<f64>],
    structural_rank: usize,
    p: usize,
) -> SubspaceLeakageMetrics {
    let mut max_abs_sq = 0.0_f64;
    let mut max_rel_sq = 0.0_f64;
    let mut worst_penalty = 0usize;

    for (k, rs) in rs_transformed.iter().enumerate() {
        let rows = rs.nrows();
        let cols = rs.ncols().min(p);
        let null_start = structural_rank.min(cols);
        let mut abs_sq = 0.0_f64;
        let mut total_sq = 0.0_f64;
        for i in 0..rows {
            for j in 0..cols {
                let v = rs[(i, j)];
                let vv = v * v;
                total_sq += vv;
                if j >= null_start {
                    abs_sq += vv;
                }
            }
        }
        let rel_sq = if total_sq > 0.0 {
            abs_sq / total_sq
        } else {
            0.0
        };
        if rel_sq > max_rel_sq {
            max_rel_sq = rel_sq;
            worst_penalty = k;
        }
        max_abs_sq = max_abs_sq.max(abs_sq);
    }

    let mut max_cross_gram_abs = 0.0_f64;
    let null_count = p.saturating_sub(structural_rank);
    if structural_rank > 0 && null_count > 0 {
        for i in 0..structural_rank {
            for j in 0..null_count {
                let qn_col = structural_rank + j;
                let mut dot = 0.0_f64;
                for r in 0..p {
                    dot += qs[(r, i)] * qs[(r, qn_col)];
                }
                max_cross_gram_abs = max_cross_gram_abs.max(dot.abs());
            }
        }
    }

    SubspaceLeakageMetrics {
        max_abs_sq,
        max_rel_sq,
        worst_penalty,
        max_cross_gram_abs,
    }
}

/// True when the penalized/null subspace split is numerically self-consistent.
///
/// The split has two independent invariants:
///
/// 1. **Orthogonality** — `Qs = [Q_p | Q_n]` must be orthonormal, so the range
///    and null blocks share no direction (`max |Qp'Qn| ≤ orth_tol`). This is the
///    structural correctness guarantee and is checked at machine precision.
///
/// 2. **Bounded root leakage** — the transformed penalty root must have
///    negligible energy on the null columns. The admissible relative-energy
///    leakage is DERIVED from `REL_PSD_FLOOR`, the same numerically-PSD floor
///    that `classify_eigenvalues_strict` uses to decide which modes are null:
///    a mode the classifier is entitled to call null can, by the symmetric
///    eigensolver's own eigenvector accuracy on a small-gap spectrum, retain up
///    to `REL_PSD_FLOOR` of the penalty root's relative energy in that direction.
///    Summed over the at-most-`p` near-threshold null modes the relative leakage
///    cannot exceed `p · REL_PSD_FLOOR` without signalling a genuine
///    (non-numerical) inconsistency, so that is the derived tolerance — matching
///    the `p`-scaling `classify_eigenvalues_strict` already applies to its
///    machine floor. Demanding a leakage tighter than the very floor that
///    defined the null space is self-contradictory: it rejects well-posed smooth
///    manifold / Duchon / sphere penalties whose Laplace-Beltrami spectrum decays
///    through the classification threshold with no clean rank gap (#1802), even
///    though the downstream penalty (`E'E`) is rebuilt with an EXACTLY clean null
///    block regardless. The absolute floor keeps a vanishing-scale penalty from
///    tripping the relative test on pure roundoff.
fn subspace_split_is_consistent(leakage: &SubspaceLeakageMetrics, p: usize) -> bool {
    let leakage_rel_tol = (p.max(1) as f64) * REL_PSD_FLOOR;
    let leakage_abs_tol = 1e-12;
    let orth_tol = 1e-10;
    let root_leaks = leakage.max_rel_sq > leakage_rel_tol && leakage.max_abs_sq > leakage_abs_tol;
    let split_nonorthogonal = leakage.max_cross_gram_abs > orth_tol;
    !(root_leaks || split_nonorthogonal)
}

fn compose_qs_from_split(q_pen: &Mat<f64>, q_null: &Mat<f64>, p: usize) -> Mat<f64> {
    let rank = q_pen.ncols();
    let null_count = q_null.ncols();
    let mut qs = Mat::<f64>::zeros(p, p);
    for i in 0..p {
        for j in 0..rank {
            qs[(i, j)] = q_pen[(i, j)];
        }
        for j in 0..null_count {
            qs[(i, rank + j)] = q_null[(i, j)];
        }
    }
    qs
}

pub use crate::kronecker::kronecker_product;

/// Result of the stable reparameterization algorithm from Wood (2011) Appendix B
#[derive(Clone)]
pub struct ReparamResult {
    /// Penalty matrix in TRANSFORMED coefficient coordinates.
    ///
    /// This must be compatible with `beta_transformed` and `X_transformed = X * Qs`.
    pub s_transformed: Array2<f64>,
    /// Log-determinant of the penalty matrix (stable computation)
    pub log_det: f64,
    /// First derivatives of log-determinant w.r.t. log-smoothing parameters
    pub det1: Array1<f64>,
    /// Orthogonal transformation matrix Qs
    pub qs: Array2<f64>,
    /// Canonical penalties in the TRANSFORMED coordinate frame.
    /// The single source of truth for penalty roots in the transformed frame.
    /// Downstream consumers use these for block-local `PenaltyCoordinate`
    /// construction, TK correction, and ext-coord paths.
    pub canonical_transformed: Vec<CanonicalPenalty>,
    /// Lambda-dependent penalty square root in TRANSFORMED coordinates (rank x p matrix).
    /// This is used for applying the actual penalty in the least squares solve.
    pub e_transformed: Array2<f64>,
    /// Truncated eigenvectors (p × m where m = p - structural_rank).
    ///
    /// Coordinate frame note:
    /// - This matrix is stored in the TRANSFORMED coefficient frame (post-`Qs`),
    ///   i.e. it is compatible with `canonical_transformed`, `beta_transformed`,
    ///   and transformed Hessians without additional coordinate mapping.
    ///
    /// These vectors span the structural null space used by positive-part
    /// log-determinant conventions.
    pub u_truncated: Array2<f64>,
}

/// The coefficient frame a set of penalty roots is expressed in.
///
/// The reparameterization's declared-null basis is stored in the TRANSFORMED
/// (post-`Qs`) frame, so projecting an ORIGINAL-frame penalty against it
/// requires rotating the basis by `Qs` first. Naming the frame at the call site
/// is what keeps that rotation from being a coin flip.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PenaltyFrame {
    Original,
    Transformed,
}

/// The λ-invariant declared-null subspace of a penalty reparameterization,
/// materialized in both coefficient frames (#2454).
///
/// # The invariant this type exists to hold
///
/// `gam.reparam` keeps only balanced-penalty eigendirections above a relative
/// rank tolerance and rebuilds the penalty the model applies as
/// `S̃(λ) = E(λ)ᵀE(λ)` on that subspace alone. **Everything the criterion is
/// made of is a function of `S̃`**: the inner solve minimizes `−ℓ + ½βᵀS̃β`,
/// `H = −∇²ℓ + S̃`, and the reported penalty energy is `‖E β̂‖²`.
///
/// A per-block `S_k` whose own root rank exceeds the split's penalized rank
/// therefore describes a DIFFERENT penalty. Two things go wrong if one is used
/// anywhere the criterion is differentiated or normalized:
///
///  1. `½ λ_k β̂ᵀS_kβ̂` charges β̂'s energy in directions `S̃` never penalized —
///     and `β̂` is free there, so that energy is `O(1)` and `∂/∂ρ_k` multiplies
///     it by `λ_k`. That is an additive `c·λ` in the outer gradient, invisible
///     at `‖ρ‖ ≤ 1` and sign-flipping it a dozen e-folds up.
///  2. `−½log|Σ_k λ_k S_k|₊` charges MORE directions than `½log|H|` can ever
///     inflate, because `H`'s penalty part is `S̃`. The two halves of the LAML
///     ratio then saturate at different rates and the criterion acquires an
///     asymptotic slope of `½(rank(S̃) − rank(S))` per unit ρ — unbounded
///     below, with no interior optimum and no certifiable λ=∞ rail.
///
/// Projecting restores the identity every outer derivative is built on:
/// `Σ_k λ_k (Π S_k Π) = Π (Σ_k λ_k S_k) Π = S̃(λ)`, and because `Π` is
/// λ-invariant by construction, `∂S̃/∂ρ_k = λ_k · Π S_k Π` exactly. One penalty
/// object then serves the value, the quadratic, the per-block scores, the
/// `tr(H⁻¹Ḣ_k)` drift, the `log|S|₊` and its rank, and the outer Hessian.
///
/// A `None` basis means "nothing declared null" and every `project` is the
/// identity, which is the case for any penalty whose numerical rank agrees
/// with the split's.
#[derive(Clone, Debug)]
pub struct PenaltyNullSplit {
    transformed: Option<Array2<f64>>,
    original: Option<Array2<f64>>,
}

impl PenaltyNullSplit {
    /// The identity split: nothing is declared null, every projection is a
    /// no-op. This is what a penalty system with no dense reparameterization
    /// (Kronecker marginal grids, sparse-native identity frames) carries.
    pub fn identity() -> Self {
        Self {
            transformed: None,
            original: None,
        }
    }

    /// The number of declared-null directions, or 0 for the identity split.
    pub fn declared_null_dim(&self) -> usize {
        self.transformed.as_ref().map_or(0, |n| n.ncols())
    }

    /// The declared-null basis in `frame`, if this split declares anything.
    pub fn basis(&self, frame: PenaltyFrame) -> Option<&Array2<f64>> {
        match frame {
            PenaltyFrame::Original => self.original.as_ref(),
            PenaltyFrame::Transformed => self.transformed.as_ref(),
        }
    }

    /// `Π S_k Π` for a canonical penalty expressed in `frame`.
    ///
    /// The penalty is returned unchanged when this split declares nothing null,
    /// or when the basis does not match the penalty's dimension — a shape
    /// disagreement the projection must not paper over by guessing.
    pub fn project_canonical(
        &self,
        penalty: &CanonicalPenalty,
        frame: PenaltyFrame,
    ) -> Result<CanonicalPenalty, EstimationError> {
        match self.basis(frame) {
            Some(n) if n.nrows() == penalty.total_dim => {
                penalty.project_out_null_directions(n.view())
            }
            _ => Ok(penalty.clone()),
        }
    }

    /// `Π S_k Π` for a solver-side penalty coordinate expressed in `frame`.
    pub fn project_coordinate(
        &self,
        coord: &gam_problem::PenaltyCoordinate,
        frame: PenaltyFrame,
    ) -> gam_problem::PenaltyCoordinate {
        match self.basis(frame) {
            Some(n) if n.nrows() == coord.dim() => coord.project_out_null_directions(n.view()),
            _ => coord.clone(),
        }
    }
}

impl ReparamResult {
    /// The λ-invariant subspace split this reparameterization declared.
    ///
    /// `u_truncated` is `p × m` in the transformed frame; the original-frame
    /// basis is `Qs · u_truncated` (`Qs` orthogonal, and `u_truncated` is
    /// itself defined as `Qsᵀ Q_null`). A degenerate or absent basis yields the
    /// identity split rather than a guessed rotation.
    pub fn null_split(&self) -> PenaltyNullSplit {
        let u = &self.u_truncated;
        if u.ncols() == 0 || u.nrows() == 0 {
            return PenaltyNullSplit::identity();
        }
        let qs = &self.qs;
        let original = if qs.nrows() == u.nrows() && qs.ncols() == u.nrows() {
            qs.dot(u)
        } else {
            // No usable `Qs`: the frames coincide (sparse-native / identity
            // reparameterization) or the shapes are inconsistent, in which case
            // `project` declines on the dimension check anyway.
            u.clone()
        };
        PenaltyNullSplit {
            transformed: Some(u.clone()),
            original: Some(original),
        }
    }
}

// ---------------------------------------------------------------------------
// Kronecker factor decomposition primitives
// ---------------------------------------------------------------------------

/// Per-factor decomposition result for Kronecker penalties.
struct KroneckerFactorDecomp {
    root: Array2<f64>,              // rank_j × q_j
    positive_eigenvalues: Vec<f64>, // length = rank_j
    rank: usize,
    dim: usize,
}

/// Eigendecompose each Kronecker factor separately at O(Σ q_j³).
/// Returns per-factor decompositions, or `None` if any factor is zero.
fn decompose_kronecker_factors(
    factors: &[Array2<f64>],
    context: &str,
) -> Result<Option<Vec<KroneckerFactorDecomp>>, EstimationError> {
    let mut decomps = Vec::with_capacity(factors.len());
    for (j, factor) in factors.iter().enumerate() {
        let q_j = factor.nrows();
        if q_j != factor.ncols() {
            crate::bail_invalid_estim!(
                "{context}: Kronecker factor {j} must be square, got {}x{}",
                factor.nrows(),
                factor.ncols()
            );
        }
        let is_identity = {
            let mut is_id = true;
            'outer: for r in 0..q_j {
                for c in 0..q_j {
                    let expected = if r == c { 1.0 } else { 0.0 };
                    if (factor[[r, c]] - expected).abs() > 1e-12 {
                        is_id = false;
                        break 'outer;
                    }
                }
            }
            is_id
        };
        if is_identity {
            decomps.push(KroneckerFactorDecomp {
                root: Array2::eye(q_j),
                positive_eigenvalues: vec![1.0; q_j],
                rank: q_j,
                dim: q_j,
            });
            continue;
        }
        let analysis = analyze_penalty_block(factor).map_err(|err| {
            EstimationError::InvalidInput(format!(
                "{context}: Kronecker factor {j} eigendecomp failed: {err}"
            ))
        })?;
        if analysis.rank == 0 {
            return Ok(None);
        }
        // Build the factor root from ONLY the range (positive-curvature)
        // directions via the canonical classifier — never the null or
        // negative-curvature directions (#1425).
        let factor_classes =
            crate::basis::SpectralClassification::new(
                &analysis.eigenvalues,
                analysis.rank_tol,
                analysis.noise_tol,
            );
        let mut root_j = Array2::zeros((analysis.rank, q_j));
        let mut pos_eigs = Vec::with_capacity(analysis.rank);
        for (row_idx, &i) in factor_classes.range_idx.iter().enumerate() {
            let eigenval = analysis.eigenvalues[i];
            let sqrt_ev = eigenval.sqrt();
            let evec = analysis.eigenvectors.column(i);
            for (col, &v) in evec.iter().enumerate() {
                root_j[[row_idx, col]] = sqrt_ev * v;
            }
            pos_eigs.push(eigenval);
        }
        decomps.push(KroneckerFactorDecomp {
            root: root_j,
            positive_eigenvalues: pos_eigs,
            rank: analysis.rank,
            dim: q_j,
        });
    }
    Ok(Some(decomps))
}

/// Build the block-local Kronecker root from pre-computed factor decompositions.
fn assemble_kronecker_root_local(decomps: &[KroneckerFactorDecomp]) -> Array2<f64> {
    let mut kron_root = decomps[0].root.clone();
    for fr in &decomps[1..] {
        let (r1, c1) = kron_root.dim();
        let (r2, c2) = (fr.rank, fr.dim);
        let mut new_root = Array2::zeros((r1 * r2, c1 * c2));
        for i1 in 0..r1 {
            for i2 in 0..r2 {
                for j1 in 0..c1 {
                    for j2 in 0..c2 {
                        new_root[[i1 * r2 + i2, j1 * c2 + j2]] =
                            kron_root[[i1, j1]] * fr.root[[i2, j2]];
                    }
                }
            }
        }
        kron_root = new_root;
    }
    kron_root
}

/// Compute eigenvalues of the Kronecker product from per-factor eigenvalues.
fn kronecker_eigenvalues(decomps: &[KroneckerFactorDecomp], block_dim: usize) -> (Vec<f64>, usize) {
    let mut kron_eigs = decomps[0].positive_eigenvalues.clone();
    for fd in &decomps[1..] {
        let mut new_eigs = Vec::with_capacity(kron_eigs.len() * fd.positive_eigenvalues.len());
        for &a in &kron_eigs {
            for &b in &fd.positive_eigenvalues {
                new_eigs.push(a * b);
            }
        }
        kron_eigs = new_eigs;
    }
    let max_ev = kron_eigs.iter().copied().fold(0.0_f64, f64::max);
    let tol = max_ev * 1e-10 * (block_dim as f64);
    let positive: Vec<f64> = kron_eigs.into_iter().filter(|&ev| ev > tol).collect();
    let nullity = block_dim - positive.len();
    (positive, nullity)
}

// ---------------------------------------------------------------------------
// CanonicalPenalty — block-local processed penalty for the solver
// ---------------------------------------------------------------------------

/// A canonicalized penalty with block-local root, ready for the solver.
///
/// Instead of storing a full `p x p` penalty matrix, this stores only the
/// `rank x block_dim` root and the column range, enabling O(p_k^2) operations
/// instead of O(p^2).
#[derive(Clone)]
pub struct CanonicalPenalty {
    /// Square root matrix: S_k = root^T * root.
    /// Shape: `rank x block_dim` for block-local, `rank x p` for dense.
    pub root: Array2<f64>,
    /// Column range in the global coefficient vector [start..end).
    /// For dense penalties this is `0..p`.
    pub col_range: std::ops::Range<usize>,
    /// Full parameter dimension p.
    pub total_dim: usize,
    /// Structural nullity of the local penalty.
    pub nullity: usize,
    /// The symmetrized block-local penalty matrix (block_dim × block_dim).
    /// Cached at construction time to avoid recomputing root^T * root
    /// in hot paths (penalty assembly, trace products).
    pub local: Array2<f64>,
    /// Block-local prior mean used to center this penalty.
    pub prior_mean: Array1<f64>,
    /// Positive eigenvalues of the local penalty matrix (length = rank).
    /// Cached at construction time for REML logdet block-factored paths.
    pub positive_eigenvalues: Vec<f64>,
    /// Optional operator-form handle bit-equivalent to `local`. Propagated
    /// from `PenaltySpec::Block.op`. Downstream PIRLS and REML exact operator
    /// algebra route through this for dense-Gram-free matvec when present.
    pub op: Option<std::sync::Arc<dyn crate::analytic_penalties::PenaltyOp>>,
}

impl std::fmt::Debug for CanonicalPenalty {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CanonicalPenalty")
            .field(
                "root",
                &format_args!("{}×{}", self.root.nrows(), self.root.ncols()),
            )
            .field("col_range", &self.col_range)
            .field("total_dim", &self.total_dim)
            .field("nullity", &self.nullity)
            .field(
                "local",
                &format_args!("{}×{}", self.local.nrows(), self.local.ncols()),
            )
            .field("prior_mean_len", &self.prior_mean.len())
            .field("positive_eigenvalues", &self.positive_eigenvalues)
            .field("op", &self.op.as_ref().map(|o| o.dim()))
            .finish()
    }
}

impl CanonicalPenalty {
    /// Construct a dense (full-width) canonical penalty from a `rank x p` root.
    /// Used to wrap reparam-transformed roots for consumers that expect
    /// `&[CanonicalPenalty]`.
    pub fn from_dense_root(root: Array2<f64>, p: usize) -> Self {
        Self::from_dense_root_with_mean(root, p, Array1::zeros(p))
    }

    pub fn from_dense_root_with_mean(root: Array2<f64>, p: usize, prior_mean: Array1<f64>) -> Self {
        assert_eq!(prior_mean.len(), p);
        let local = root.t().dot(&root);
        let positive_eigenvalues = Vec::new(); // not needed for TK paths
        Self {
            root,
            col_range: 0..p,
            total_dim: p,
            nullity: 0,
            local,
            prior_mean,
            positive_eigenvalues,
            op: None,
        }
    }

    /// Embed the block-local root into a full-width `rank × total_dim` matrix.
    /// For dense penalties (col_range = 0..p), returns the root unchanged.
    pub fn full_width_root(&self) -> Array2<f64> {
        if self.col_range.start == 0 && self.col_range.end == self.total_dim {
            return self.root.clone();
        }
        let rank = self.root.nrows();
        let mut full = Array2::<f64>::zeros((rank, self.total_dim));
        full.slice_mut(ndarray::s![.., self.col_range.clone()])
            .assign(&self.root);
        full
    }

    /// Numerical rank of this penalty.
    pub fn rank(&self) -> usize {
        self.root.nrows()
    }

    /// Block dimension (number of columns this penalty covers).
    pub fn block_dim(&self) -> usize {
        self.col_range.len()
    }

    /// Whether this penalty is block-local (col_range != 0..total_dim).
    pub const fn is_block_local(&self) -> bool {
        self.col_range.start != 0 || self.col_range.end != self.total_dim
    }

    /// Return a reference to the cached local penalty matrix.
    /// Shape: `block_dim x block_dim`.
    pub fn local_ref(&self) -> &Array2<f64> {
        &self.local
    }

    /// Return an owned copy of the local penalty matrix.
    /// Prefer `local_ref()` when a reference suffices.
    pub fn local_penalty(&self) -> Array2<f64> {
        self.local.clone()
    }

    /// Accumulate lambda * S_k into a pre-allocated `p x p` target matrix.
    /// Only touches the block [col_range × col_range].
    pub fn accumulate_weighted(&self, target: &mut Array2<f64>, lambda: f64) {
        if lambda == 0.0 || self.rank() == 0 {
            return;
        }
        let r = &self.col_range;
        target
            .slice_mut(s![r.start..r.end, r.start..r.end])
            .scaled_add(lambda, &self.local);
    }

    /// Compute `scale * v^T S_k v` (quadratic form).
    /// Only reads `v[start..end]` — O(rank × block_dim) not O(rank × p).
    pub fn quadratic(&self, v: &Array1<f64>, scale: f64) -> f64 {
        if self.rank() == 0 || scale == 0.0 {
            return 0.0;
        }
        let v_block = v.slice(s![self.col_range.start..self.col_range.end]);
        let rv = self.root.dot(&v_block);
        scale * rv.dot(&rv)
    }

    /// Compute `scale * S_k * prior_mean` embedded into the global basis.
    pub fn prior_linear_shift(&self, scale: f64) -> Array1<f64> {
        let mut out = Array1::<f64>::zeros(self.total_dim);
        if self.rank() == 0 || scale == 0.0 || self.prior_mean.iter().all(|&v| v == 0.0) {
            return out;
        }
        let block = self.local.dot(&self.prior_mean) * scale;
        out.slice_mut(s![self.col_range.start..self.col_range.end])
            .assign(&block);
        out
    }

    /// Compute `scale * prior_mean' S_k prior_mean`.
    pub fn prior_constant_shift(&self, scale: f64) -> f64 {
        if self.rank() == 0 || scale == 0.0 || self.prior_mean.iter().all(|&v| v == 0.0) {
            return 0.0;
        }
        scale * self.prior_mean.dot(&self.local.dot(&self.prior_mean))
    }

    /// Embed this block's prior mean into the global coefficient basis.
    pub fn full_width_prior_mean(&self) -> Array1<f64> {
        if self.col_range.start == 0 && self.col_range.end == self.total_dim {
            return self.prior_mean.clone();
        }
        let mut out = Array1::<f64>::zeros(self.total_dim);
        out.slice_mut(s![self.col_range.start..self.col_range.end])
            .assign(&self.prior_mean);
        out
    }

    /// `Π S_k Π` for the declared-null basis `N` (orthonormal, `total_dim × m`)
    /// in THIS penalty's coefficient frame, with `Π = I − N Nᵀ` (#2454).
    ///
    /// See [`PenaltyNullSplit`] for why every penalty the criterion touches has
    /// to be projected onto the reparameterization's penalized subspace. This
    /// is the `CanonicalPenalty` face of [`gam_problem::PenaltyCoordinate::
    /// project_out_null_directions`]; both go through the same root primitive.
    ///
    /// Returns `self` unchanged — `op` handle, cached spectrum and all — when
    /// the projection is below the root's own representation noise
    /// (`‖R N‖_max ≤ ‖R‖_max · ε · p`). That is not a shortcut but a statement
    /// of exactness: a null direction the root does not resolve carries no
    /// energy the criterion's `log|S|₊` can see either, so the projected and
    /// unprojected penalties agree to every digit either object represents.
    /// It is also the ordinary case — for any penalty whose structural null
    /// space is exact, `N` spans `∩_k null(S_k)` and `Π S_k Π = S_k` in exact
    /// arithmetic.
    ///
    /// When the projection does bite, block locality is preserved if the null
    /// basis is itself block-local, `local` is rebuilt as `RᵀR` from the
    /// projected root, and `positive_eigenvalues` becomes the projected root's
    /// squared singular values — so `Σ positive_eigenvalues = tr(Π S_k Π)`
    /// stays exact for the consumers that read it as a trace. The `op` handle
    /// is dropped: it is declared bit-equivalent to `local`, and the projected
    /// `local` is a different operator.
    pub fn project_out_null_directions(
        &self,
        null_basis: ndarray::ArrayView2<'_, f64>,
    ) -> Result<Self, EstimationError> {
        if null_basis.ncols() == 0 || self.rank() == 0 {
            return Ok(self.clone());
        }
        if null_basis.nrows() != self.total_dim {
            return Err(EstimationError::LayoutError(format!(
                "CanonicalPenalty::project_out_null_directions: null-basis row count {} does \
                 not match the penalty's total dimension {}",
                null_basis.nrows(),
                self.total_dim
            )));
        }
        let root_scale = self.root.iter().fold(0.0_f64, |a, v| a.max(v.abs()));
        let (projected_full, stays_block_local) =
            gam_problem::project_block_root_out_of_null_directions(
                self.root.view(),
                self.col_range.start,
                self.col_range.end,
                self.total_dim,
                null_basis,
            );
        // How much the projection actually moved, against the root's own
        // representation noise.
        let mut moved = 0.0_f64;
        {
            let original_full = self.full_width_root();
            for (a, b) in projected_full.iter().zip(original_full.iter()) {
                moved = moved.max((a - b).abs());
            }
        }
        if moved <= root_scale * f64::EPSILON * (self.total_dim as f64) {
            return Ok(self.clone());
        }

        let (root, col_range) = if stays_block_local {
            (
                projected_full
                    .slice(s![.., self.col_range.start..self.col_range.end])
                    .to_owned(),
                self.col_range.clone(),
            )
        } else {
            (projected_full, 0..self.total_dim)
        };
        let prior_mean = if col_range == self.col_range {
            self.prior_mean.clone()
        } else {
            let mut full = Array1::<f64>::zeros(self.total_dim);
            full.slice_mut(s![self.col_range.start..self.col_range.end])
                .assign(&self.prior_mean);
            full
        };
        // `RᵀR` eigenvalues are the squared singular values of `R`; taking them
        // from the root rather than from an eigendecomposition of the assembled
        // `local` keeps the small ones (the whole point of a root-scale
        // representation) and costs O(rank² · block_dim).
        let positive_eigenvalues = match root.svd(false, false) {
            Ok((_, singular_values, _)) => {
                let mut eigs = vec![0.0_f64; root.nrows()];
                for (slot, &sigma) in eigs.iter_mut().zip(singular_values.iter()) {
                    *slot = sigma * sigma;
                }
                eigs
            }
            Err(error) => {
                return Err(EstimationError::LayoutError(format!(
                    "CanonicalPenalty::project_out_null_directions: SVD of the projected root \
                     failed: {error:?}"
                )));
            }
        };
        let local = root.t().dot(&root);
        Ok(Self {
            root,
            col_range,
            total_dim: self.total_dim,
            // The projection can only remove curvature, so the block's nullity
            // can only grow; the honest floor is what it already declared.
            nullity: self.nullity,
            local,
            prior_mean,
            positive_eigenvalues,
            op: None,
        })
    }

    /// Convert to a PenaltyCoordinate for the unified REML evaluator.
    pub fn to_penalty_coordinate(&self) -> gam_problem::PenaltyCoordinate {
        use gam_problem::PenaltyCoordinate;
        if self.is_block_local() {
            PenaltyCoordinate::from_block_root_with_mean(
                self.root.clone(),
                self.col_range.start,
                self.col_range.end,
                self.total_dim,
                self.prior_mean.clone(),
            )
        } else {
            PenaltyCoordinate::from_dense_root_with_mean(self.root.clone(), self.prior_mean.clone())
        }
    }
}

/// Measure and report how close each pair of penalties in the canonical bundle
/// comes to being proportional.
///
/// Two penalties `S_i`, `S_j` with the same `col_range` are compared by the
/// **relative defect of the proportionality claim itself**,
///
/// ```text
///     delta(S_i, S_j) = min_c ||S_j - c S_i||_F / ||S_i||_F,
///     attained at      c = <S_i, S_j>_F / ||S_i||_F^2,
/// ```
///
/// which is what `sum_i w_i S_i = 0` actually asks about. The cosine is the
/// same information (`delta^2 = 1 - cos^2` for unit-normalized operators), but
/// it is the WRONG COORDINATE to report it in and this issue is the cost of
/// having done so: `delta` enters the cosine SQUARED, so a pair that is
/// `1.9e-5` apart — five orders above any arithmetic — prints as
/// `cos = 1.000000` and reads as an exact identity. #2676 ran on that reading
/// for its whole life. Measured on `geo_disease_matern` (centers=10, n=1500,
/// n_pcs=16, the #2676 penalty-map probe):
///
/// ```text
///     length_scale   delta       certified nullity
///       2.05e-2      2.079e-15          1     <- exact, to the residual's own round-off
///       1.64e-1      1.874e-5           0     <- prints cos = 1.000000
///       1.27e0       3.396e-1           0     <- what the fit realizes
/// ```
///
/// so the "structural identity" that issue is named for is the small-length-
/// scale LIMIT of two genuinely different operators, and it is not present at
/// the geometry any of those fits settle on.
/// Because `local` is symmetric, `<A, B>_F = sum_{r,c} A[r,c] * B[r,c] =
/// tr(A·B)`. Pairs with different `col_range` cannot be proportional by
/// construction and are skipped.
///
/// # ⚠ This detector is a SCREEN, not the authority (#2676)
///
/// It sees proportional PAIRS. The criterion's actual invariance is the null
/// space of the penalty map's Gram, `{w : sum_i w_i S_i = 0}`, which contains
/// every linear redundancy — including three-term ones no pairwise measure can
/// find. `gam_solve::penalty_invariance::PenaltyMapInvariance` computes that
/// subspace and is what the curvature certificate and the smoothing correction
/// consult. Use this function for the human-facing report; do not use it to
/// decide anything numeric.
///
/// # ⚠ And an EXACT redundancy is not a saddle
///
/// This warning used to say the redundancy makes the LAML cost carry "a
/// Z₂-symmetric saddle". It does not. Where the redundancy is EXACT the
/// criterion is exactly constant along it in `lambda` — a flat direction, not
/// a descending one — and every apparent negative curvature there is the
/// chain-rule term of
/// `H_rho = diag(lambda) H_lambda diag(lambda) + diag(g_rho)`, whose sign is
/// the sign of a rounding residual. An optimizer converging there has not
/// converged to a saddle; it has converged to a point on a manifold of points
/// that all give the SAME fit, because they all assemble the same penalty.
/// That is a non-identifiability of `lambda`, worth reporting, and not a defect
/// in the answer.
///
/// Where the redundancy is only NEAR-exact none of that holds: the criterion is
/// not constant along the direction, it carries genuine curvature of order
/// `delta^2`, and calling it "structurally identical" asserts a flatness the
/// evidence does not support. Which of the two a pair is is decided below, at
/// the residual's own arithmetic floor, and it is said in the message.
///
/// # The exactness bar, and why it is derived rather than chosen
///
/// `delta` is computed as a residual norm over `m = block_dim^2` products of
/// `O(1)` entries, so its own round-off is `sqrt(m) * EPSILON` — the error of
/// the norm, not of the operators. A pair at or under that is proportional to
/// everything this arithmetic can see; a pair above it is measurably not.
/// Measured, the two populations sit ten orders apart (`2.079e-15` against
/// `1.874e-5` on the sweep above, at `sqrt(m) * EPSILON = 2.2e-15` for
/// `block_dim = 10`), so nothing rides on where in that gap the bar falls.
///
/// Logging policy, `delta`-denominated throughout:
/// - `delta <= sqrt(m) * EPSILON` → `log::warn!` with `[PENALTY-REDUNDANCY]`.
///   Only their combination `lambda_i + c*lambda_j` is identified.
/// - `sqrt(m) * EPSILON < delta <= 1e-1` → `log::info!` with
///   `[PENALTY-SIMILARITY]`, carrying `delta`. At large scale (`k > 64`) only
///   the three smallest-`delta` such pairs are logged to bound log volume.
///
/// Returns `Vec<(i, j, delta)>` for the **exactly proportional** pairs,
/// primarily to make this function unit-testable without a log capture.
///
/// Performance: this is O(k² · block_dim²); intended to be called exactly
/// once per fit (e.g. from `RemlState::newwith_offset_shared`).
pub fn report_penalty_pair_redundancy(canonical: &[CanonicalPenalty]) -> Vec<(usize, usize, f64)> {
    // A pair `delta` above this is measurably NOT proportional, and calling it
    // "similar" past a tenth of the operator's own size says nothing. Purely a
    // log-volume bound: no verdict is taken on it.
    const SIMILARITY_REPORTING_DEFECT: f64 = 1e-1;
    const LARGE_SCALE_K_THRESHOLD: usize = 64;
    const TOP_SIMILARITY_PAIRS: usize = 3;

    let k = canonical.len();
    let mut redundant: Vec<(usize, usize, f64)> = Vec::new();
    let mut similar: Vec<(usize, usize, f64, f64)> = Vec::new();

    // Pre-compute tr(S_i^2) = sum of squares of S_i entries (Frobenius norm
    // squared). `local` is symmetric, so this equals tr(S_i^T S_i) = tr(S_i^2).
    let trace_sq: Vec<f64> = canonical
        .iter()
        .map(|p| p.local.iter().map(|&v| v * v).sum::<f64>())
        .collect();

    for i in 0..k {
        if trace_sq[i] == 0.0 {
            continue;
        }
        for j in (i + 1)..k {
            if trace_sq[j] == 0.0 {
                continue;
            }
            // Different col_range → cannot be proportional by construction (the
            // block-local matrices live in disjoint or mismatched parameter
            // subspaces).
            if canonical[i].col_range != canonical[j].col_range {
                continue;
            }
            // Shapes must match — they do when col_range matches because
            // `local` is `block_dim × block_dim` and `block_dim = col_range.len()`.
            assert_eq!(canonical[i].local.dim(), canonical[j].local.dim());

            let inner: f64 = canonical[i]
                .local
                .iter()
                .zip(canonical[j].local.iter())
                .map(|(&a, &b)| a * b)
                .sum();
            // The least-squares proportionality constant and the residual it
            // leaves, formed DIRECTLY rather than through `1 - cos`: the cosine
            // route squares `delta` and then subtracts from one, which loses
            // every digit of a defect under `sqrt(EPSILON)` — the loss this
            // whole issue is made of.
            let scale = inner / trace_sq[i];
            let residual_sq: f64 = canonical[i]
                .local
                .iter()
                .zip(canonical[j].local.iter())
                .map(|(&a, &b)| {
                    let residual = b - scale * a;
                    residual * residual
                })
                .sum();
            let defect = (residual_sq / trace_sq[i]).sqrt();
            let entries = canonical[i].local.len() as f64;
            let arithmetic_floor = entries.sqrt() * f64::EPSILON;

            if defect <= arithmetic_floor {
                redundant.push((i, j, defect));
            } else if defect <= SIMILARITY_REPORTING_DEFECT {
                similar.push((i, j, defect, scale));
            }
        }
    }

    // Always emit every exact redundancy — these are structural model errors.
    for &(i, j, defect) in &redundant {
        log::warn!(
            "[PENALTY-REDUNDANCY] penalties i={i} j={j} are proportional to the arithmetic \
             that formed them (relative defect min_c ||S_{j} - c S_{i}||_F / ||S_{i}||_F = \
             {defect:.6e}) — only their COMBINATION is identified, so the criterion is exactly \
             constant along their antisymmetric direction and lambda_{i} / lambda_{j} are not \
             separately estimable. The fit itself is unaffected (every point of that manifold \
             assembles the same penalty), and the curvature certificate deflates the direction \
             rather than judging a chain-rule term there (#2676). Consider re-specifying (e.g. \
             anisotropic→isotropic for spatial smoothers with weak axis signal) if you need the \
             individual smoothing parameters to mean something."
        );
    }

    // Cap similarity log volume at large scale.
    if k > LARGE_SCALE_K_THRESHOLD && similar.len() > TOP_SIMILARITY_PAIRS {
        similar.sort_by(|a, b| a.2.partial_cmp(&b.2).unwrap_or(std::cmp::Ordering::Equal));
        similar.truncate(TOP_SIMILARITY_PAIRS);
    }
    for (i, j, defect, scale) in similar {
        log::info!(
            "[PENALTY-SIMILARITY] penalties i={i} j={j} are close but MEASURABLY distinct \
             (relative defect {defect:.6e} at the best scale c={scale:.6e}) — the outer Hessian \
             may be ill-conditioned along their antisymmetric direction, and the criterion \
             carries genuine curvature of order defect^2 there. This is NOT an invariance: \
             nothing is exactly constant along it and no direction is deflated for it (#2676)."
        );
    }

    redundant
}

/// Canonicalize a single `PenaltySpec` into a `CanonicalPenalty` by computing
/// the block-local eigendecomposition and extracting the root.
///
/// This is O(block_dim^3) instead of O(p^3) for block-local penalties.
/// Returns `None` if the penalty has rank zero (should be dropped).
pub fn canonicalize_penalty_spec(
    spec: &crate::PenaltySpec,
    p: usize,
    idx: usize,
    context: &str,
) -> Result<Option<CanonicalPenalty>, EstimationError> {
    use crate::PenaltySpec;

    crate::validate_penalty_spec_shape(idx, spec, p, context)?;

    let (local_matrix, col_range, prior_mean_spec, hint, op) = match spec {
        PenaltySpec::Block {
            local,
            col_range,
            prior_mean,
            structure_hint,
            op,
        } => (
            local.view(),
            col_range.clone(),
            prior_mean,
            structure_hint.as_ref(),
            op.clone(),
        ),
        PenaltySpec::Dense(m) => (
            m.view(),
            0..p,
            &gam_problem::CoefficientPriorMean::Zero,
            None,
            None,
        ),
        PenaltySpec::DenseWithMean { matrix, prior_mean } => {
            (matrix.view(), 0..p, prior_mean, None, None)
        }
    };

    let block_dim = col_range.len();
    let prior_mean = prior_mean_spec
        .evaluate(block_dim, &format!("{context}: penalty {idx}"))
        .map_err(|e| EstimationError::InvalidInput(e.0))?;

    // ── Ridge fast path: closed-form, no eigendecomposition ──
    if let Some(PenaltyStructureHint::Ridge(scale)) = hint {
        if *scale <= 0.0 {
            return Ok(None);
        }
        let sqrt_scale = scale.sqrt();
        let mut root = Array2::zeros((block_dim, block_dim));
        for i in 0..block_dim {
            root[[i, i]] = sqrt_scale;
        }
        // Ridge penalties are diagonal by construction, but still route through
        // the crate-wide ndarray symmetrizer so every construction variant uses
        // the same "average the transpose" cleanup instead of a local copy.
        let mut local_sym = local_matrix.to_owned();
        symmetrize_in_place(&mut local_sym);
        return Ok(Some(CanonicalPenalty {
            root,
            col_range,
            total_dim: p,
            nullity: 0,
            local: local_sym,
            prior_mean,
            positive_eigenvalues: vec![*scale; block_dim],
            op,
        }));
    }

    // ── Kronecker fast path: single per-factor eigendecomposition ──
    if let Some(PenaltyStructureHint::Kronecker(factors)) = hint {
        let decomps =
            match decompose_kronecker_factors(factors, &format!("{context} penalty {idx}"))? {
                None => return Ok(None),
                Some(d) => d,
            };
        let (positive_eigenvalues, nullity) = kronecker_eigenvalues(&decomps, block_dim);
        if positive_eigenvalues.is_empty() {
            return Ok(None);
        }
        let root = assemble_kronecker_root_local(&decomps);
        let mut local_sym = local_matrix.to_owned();
        symmetrize_in_place(&mut local_sym);
        return Ok(Some(CanonicalPenalty {
            root,
            col_range,
            total_dim: p,
            nullity,
            local: local_sym,
            prior_mean,
            positive_eigenvalues,
            op,
        }));
    }

    // ── Generic block-local path: eigendecompose at O(block_dim³) ──
    let local_owned = local_matrix.to_owned();
    let analysis = analyze_penalty_block(&local_owned).map_err(|err| {
        EstimationError::InvalidInput(format!(
            "{context}: penalty canonicalization failed at index {idx}: {err}"
        ))
    })?;

    if analysis.rank == 0 {
        log::debug!(
            "Dropped inactive penalty block idx={idx} reason={}",
            if analysis.iszero {
                "ZeroMatrix"
            } else {
                "NumericalRankZero"
            }
        );
        return Ok(None);
    }

    // Reuse the eigendecomposition from analyze_penalty_block and route the
    // range / null / negative-curvature split through the one canonical
    // classifier, so this root construction cannot disagree with the block's
    // own `rank` / `nullity` / `negative_dim` about which directions are
    // penalized, unpenalized, or non-PSD (#1425).
    let tolerance = analysis.rank_tol;
    let classes = crate::basis::SpectralClassification::new(
        &analysis.eigenvalues,
        tolerance,
        analysis.noise_tol,
    );
    let rank_k = classes.rank();
    assert_eq!(
        rank_k, analysis.rank,
        "penalty-root rank disagreement: SpectralClassification rank={rank_k} vs analyze_penalty_block rank={} (#1425 canonical-classifier invariant)",
        analysis.rank
    );

    // Build the penalty root R from ONLY the range directions (positive
    // curvature): R has one row per range eigenpair, scaled by sqrt(ev), so
    // RᵀR reconstructs S on range(S). Null directions contribute nothing
    // (their eigenvalue is zero); negative-curvature directions are NEVER
    // square-rooted into R (their sqrt is imaginary) and are NOT null — they
    // are simply dropped from R, exactly as the closed-form Duchon kernels at
    // high d require to preserve the q_pen / q_null invariant downstream.
    let mut root = Array2::zeros((rank_k, block_dim));
    let mut positive_eigenvalues = Vec::with_capacity(rank_k);
    for (row_idx, &i) in classes.range_idx.iter().enumerate() {
        let eigenval = analysis.eigenvalues[i];
        let eigenvec = analysis.eigenvectors.column(i);
        root.row_mut(row_idx).assign(&(&eigenvec * eigenval.sqrt()));
        positive_eigenvalues.push(eigenval);
    }

    // Surface any genuine negative curvature honestly: it is neither range
    // (dropped from R) nor null (excluded from `nullity`), so it would
    // otherwise vanish without a trace. A non-PSD penalty reaching this path
    // is a real geometric fact (e.g. high-d Duchon kernels) the operator
    // should be able to see.
    if classes.is_indefinite() {
        log::debug!(
            "{context}: penalty block idx={idx} carries {} negative-curvature \
             eigendirection(s) below -tol={tolerance:e}; dropped from the canonical \
             root and NOT counted as null space (rank={rank_k}, nullity={})",
            classes.negative_dim(),
            classes.nullity()
        );
    }

    // Store the PSD reconstruction RᵀR rather than the raw symmetrised input so
    // the cached `local` matches the rank truncation embedded in `root`
    // (negative-curvature directions are excluded from both, as above).
    let local = root.t().dot(&root);
    Ok(Some(CanonicalPenalty {
        root,
        col_range,
        total_dim: p,
        nullity: classes.nullity(),
        local,
        prior_mean,
        positive_eigenvalues,
        op,
    }))
}

/// Canonicalize a batch of penalty specs, dropping zero-rank penalties.
/// Returns (active_penalties, active_nullspace_dims).
pub fn canonicalize_penalty_specs(
    specs: &[crate::PenaltySpec],
    nullspace_dims: &[usize],
    p: usize,
    context: &str,
) -> Result<(Vec<CanonicalPenalty>, Vec<usize>), EstimationError> {
    if specs.len() != nullspace_dims.len() {
        crate::bail_invalid_estim!(
            "{context}: nullspace_dims length mismatch: penalties={}, nullspace_dims={}",
            specs.len(),
            nullspace_dims.len()
        );
    }

    let mut active = Vec::with_capacity(specs.len());
    let mut active_nullspace = Vec::with_capacity(specs.len());
    for (idx, spec) in specs.iter().enumerate() {
        if let Some(canonical) = canonicalize_penalty_spec(spec, p, idx, context)? {
            active_nullspace.push(nullspace_dims[idx]);
            active.push(canonical);
        }
    }
    Ok((active, active_nullspace))
}

/// Hard cap on the dimension `p` allowed to fall back to a dense p × p
/// eigendecomposition of the overlapping balanced penalty.
///
/// Beyond this cap the overlapping-penalty path errors out instead of
/// allocating an O(p²) workspace whose eigendecomposition would dominate the
/// solve. ResourcePolicy threading is the long-term home for this cap (the
/// resource_serialize agent is widening ResourcePolicy coverage); until that
/// lands, both overlapping branches share this single constant so they can't
/// drift apart.
pub(crate) const OVERLAPPING_PENALTY_DENSE_FALLBACK_MAX_P: usize = 4096;

/// Creates a balanced penalty root from canonical penalties.
///
/// When all penalties have non-overlapping col_ranges, the balanced sum is
/// block-diagonal and eigendecomposition is done per-block at O(Σ p_k³)
/// instead of the global O(p³). Falls back to the global path when penalties
/// overlap.
pub fn create_balanced_penalty_root_from_canonical(
    penalties: &[CanonicalPenalty],
    p: usize,
) -> Result<Array2<f64>, EstimationError> {
    if penalties.is_empty() {
        return Ok(Array2::zeros((0, p)));
    }

    // Group penalties by col_range.
    let mut block_groups: BTreeMap<(usize, usize), Vec<&CanonicalPenalty>> = BTreeMap::new();
    for cp in penalties {
        if cp.rank() == 0 {
            continue;
        }
        let key = (cp.col_range.start, cp.col_range.end);
        block_groups.entry(key).or_default().push(cp);
    }

    if block_groups.is_empty() {
        return Ok(Array2::zeros((0, p)));
    }

    // Check for overlapping ranges.
    let ranges: Vec<(usize, usize)> = block_groups.keys().copied().collect();
    let mut overlapping = false;
    for i in 1..ranges.len() {
        if ranges[i].0 < ranges[i - 1].1 {
            overlapping = true;
            break;
        }
    }

    if overlapping {
        if p > OVERLAPPING_PENALTY_DENSE_FALLBACK_MAX_P {
            return Err(EstimationError::LayoutError(format!(
                "overlapping penalty root would require dense {}x{} eigendecomposition; \
                 large-model dense fallback is disabled. Keep penalties structured or \
                 extend the overlapping-penalty solver path",
                p, p
            )));
        }
        // Fallback: accumulate into p × p and eigendecompose globally, under
        // the ONE balanced rule the reparameterization split and the
        // criterion's rank hint share (gam#2454).
        let s_balanced = balanced_penalty_sum(
            penalties
                .iter()
                .filter(|cp| cp.rank() != 0)
                .map(|cp| (cp.local_ref().view(), cp.col_range.clone())),
            p,
        );
        let (eigenvalues, eigenvectors) =
            robust_eigh(&s_balanced, Side::Lower, "balanced penalty matrix")?;
        let max_eig = eigenvalues.iter().fold(0.0f64, |max, &val| max.max(val));
        let tolerance = balanced_penalty_rank_tolerance(max_eig);
        let penalty_rank = eigenvalues.iter().filter(|&&ev| ev > tolerance).count();
        if penalty_rank == 0 {
            return Ok(Array2::zeros((0, p)));
        }
        let mut eb = Array2::zeros((p, penalty_rank));
        let mut col_idx = 0;
        for (i, &eigenval) in eigenvalues.iter().enumerate() {
            if eigenval > tolerance {
                let sqrt_ev = eigenval.sqrt();
                let evec = eigenvectors.column(i);
                eb.column_mut(col_idx).assign(&(&evec * sqrt_ev));
                col_idx += 1;
            }
        }
        return Ok(eb.t().to_owned());
    }

    // Non-overlapping: eigendecompose per block at O(Σ p_k³).
    struct BlockRoot {
        col_range: Range<usize>,
        root: Array2<f64>, // rank_b × block_dim
    }
    // Materialize the BTreeMap order first. Rayon preserves Vec collection
    // order for indexed parallel iterators, so assembly below remains stable by
    // ascending column range while independent block eigendecompositions run in
    // parallel.
    let ordered_blocks: Vec<((usize, usize), Vec<&CanonicalPenalty>)> =
        block_groups.into_iter().collect();
    let block_roots: Vec<BlockRoot> = ordered_blocks
        .into_par_iter()
        .map(
            |((start, end), cps)| -> Result<Option<BlockRoot>, EstimationError> {
                let block_dim = end - start;
                // The block's balanced sum under the same shared rule as the
                // dense fallback above and the reparameterization split.
                let s_balanced_local = balanced_penalty_sum(
                    cps.iter()
                        .map(|cp| (cp.local_ref().view(), 0..block_dim)),
                    block_dim,
                );

                let (eigenvalues, eigenvectors) =
                    robust_eigh(&s_balanced_local, Side::Lower, "balanced penalty block")?;
                let max_eig = eigenvalues.iter().fold(0.0f64, |max, &val| max.max(val));
                let tolerance = balanced_penalty_rank_tolerance(max_eig);
                let block_rank = eigenvalues.iter().filter(|&&ev| ev > tolerance).count();

                if block_rank == 0 {
                    return Ok(None);
                }

                let mut root = Array2::zeros((block_rank, block_dim));
                let mut row_idx = 0;
                for (i, &eigenval) in eigenvalues.iter().enumerate() {
                    if eigenval > tolerance {
                        let sqrt_ev = eigenval.sqrt();
                        let evec = eigenvectors.column(i);
                        root.row_mut(row_idx).assign(&(&evec * sqrt_ev));
                        row_idx += 1;
                    }
                }

                Ok(Some(BlockRoot {
                    col_range: start..end,
                    root,
                }))
            },
        )
        .collect::<Result<Vec<_>, _>>()?
        .into_iter()
        .flatten()
        .collect();
    let total_rank: usize = block_roots.iter().map(|br| br.root.nrows()).sum();

    if total_rank == 0 {
        return Ok(Array2::zeros((0, p)));
    }

    // Assemble global balanced root: total_rank × p
    let mut eb = Array2::zeros((total_rank, p));
    let mut row_offset = 0;
    for br in &block_roots {
        let rank_b = br.root.nrows();
        eb.slice_mut(s![
            row_offset..(row_offset + rank_b),
            br.col_range.start..br.col_range.end
        ])
        .assign(&br.root);
        row_offset += rank_b;
    }

    Ok(eb)
}

/// Lambda-independent reparameterization invariants derived from penalty structure.
/// Relative cut of the balanced penalty spectrum below which a direction is
/// structurally unpenalized: `eigenvalue ≤ BALANCED_PENALTY_RANK_RELATIVE_TOL ·
/// max|eigenvalue|` of [`balanced_penalty_sum`].
pub const BALANCED_PENALTY_RANK_RELATIVE_TOL: f64 = 1.0e-12;

/// The λ-invariant penalty operator the structural rank is read from:
/// `Σ_k S_k / ‖S_k‖_F`, each component embedded on its own column range.
///
/// # One rule, two readers
///
/// The reparameterization splits the coefficient space into the penalized
/// subspace `H + S̃` carries and the λ-invariant null space from THIS operator's
/// spectrum. The outer criterion's `−½ log|S(λ)|₊` must range over exactly that
/// penalized subspace — its asymptotic slope in `ρ_k` is `½(rank(S̃) −
/// rank|S|₊)` per unit `ρ_k`, so any disagreement in rank is a criterion with no
/// interior optimum along that coordinate. The criterion used to take its
/// structural rank from the UNWEIGHTED sum `Σ_k S_k` at a `100·p·ε·max` cut,
/// which is a different function of the same penalties: a component whose
/// Frobenius norm is small against its neighbours (a double-penalty null-space
/// term beside a Matérn range penalty) can sit above one cut and below the
/// other. Measured on the iso-κ Matérn double-penalty ladder (gam#2454):
/// `logdet_rank = 10` against `penalized_rank = 9`, and the gate
/// `penalty_logdet_ranks_the_same_subspace_the_hessian_carries_2454` red on
/// main. Normalizing every component first makes the rank a property of the
/// penalties' supports, not of their scales, which is what "structural" means.
pub fn balanced_penalty_sum<'a, I>(components: I, p_total: usize) -> Array2<f64>
where
    I: IntoIterator<Item = (ArrayView2<'a, f64>, std::ops::Range<usize>)>,
{
    let mut balanced = Array2::<f64>::zeros((p_total, p_total));
    for (local, range) in components {
        let frob_norm = local.iter().map(|&x| x * x).sum::<f64>().sqrt();
        if !(frob_norm > 1e-12) {
            continue;
        }
        let scale = 1.0 / frob_norm;
        for i in 0..local.nrows() {
            for j in 0..local.ncols() {
                balanced[[range.start + i, range.start + j]] += scale * local[[i, j]];
            }
        }
    }
    balanced
}

/// The rank cut for a balanced penalty spectrum whose largest magnitude is
/// `max_balanced_eigenvalue`.
pub fn balanced_penalty_rank_tolerance(max_balanced_eigenvalue: f64) -> f64 {
    if max_balanced_eigenvalue > 0.0 {
        max_balanced_eigenvalue * BALANCED_PENALTY_RANK_RELATIVE_TOL
    } else {
        BALANCED_PENALTY_RANK_RELATIVE_TOL
    }
}

/// Structural rank of a set of penalty components: the number of eigenvalues of
/// [`balanced_penalty_sum`] above [`balanced_penalty_rank_tolerance`]. This is
/// the rank the reparameterization's penalized subspace has, and the rank the
/// criterion's `log|S(λ)|₊` must range over.
pub fn balanced_penalty_structural_rank<'a, I>(
    components: I,
    p_total: usize,
) -> Result<usize, EstimationError>
where
    I: IntoIterator<Item = (ArrayView2<'a, f64>, std::ops::Range<usize>)>,
{
    if p_total == 0 {
        return Ok(0);
    }
    let balanced = array_to_faer(&balanced_penalty_sum(components, p_total));
    let (eigenvalues, _) = robust_eigh_faer(&balanced, Side::Lower, "balanced penalty matrix")?;
    let max_bal = eigenvalues.iter().fold(0.0_f64, |acc, &value| acc.max(value.abs()));
    let tol = balanced_penalty_rank_tolerance(max_bal);
    Ok(eigenvalues.iter().filter(|&&value| value > tol).count())
}

#[derive(Clone)]
struct SubspaceSplit {
    q_pen: Array2<f64>,
    q_null: Array2<f64>,
}

impl SubspaceSplit {
    fn identity(p: usize) -> Self {
        Self {
            q_pen: Array2::zeros((p, 0)),
            q_null: Array2::eye(p),
        }
    }

    fn from_ordered_qs(
        qs: &Mat<f64>,
        penalized_rank: usize,
        p: usize,
    ) -> Result<Self, EstimationError> {
        if qs.nrows() != p || qs.ncols() != p {
            return Err(EstimationError::LayoutError(format!(
                "Invalid Q basis dimensions: expected {p}x{p}, got {}x{}",
                qs.nrows(),
                qs.ncols()
            )));
        }
        if penalized_rank > p {
            return Err(EstimationError::LayoutError(format!(
                "Invalid penalized rank {penalized_rank} for p={p}"
            )));
        }

        let null_count = p - penalized_rank;
        let mut q_pen = Array2::<f64>::zeros((p, penalized_rank));
        let mut q_null = Array2::<f64>::zeros((p, null_count));
        for i in 0..p {
            for j in 0..penalized_rank {
                q_pen[(i, j)] = qs[(i, j)];
            }
            for j in 0..null_count {
                q_null[(i, j)] = qs[(i, penalized_rank + j)];
            }
        }

        Ok(Self { q_pen, q_null })
    }

    fn rank(&self) -> usize {
        self.q_pen.ncols()
    }

    fn p(&self) -> usize {
        self.q_pen.nrows()
    }

    fn compose_qs(&self) -> Array2<f64> {
        let p = self.p();
        let rank = self.rank();
        let null_count = self.q_null.ncols();
        let mut qs = Array2::<f64>::zeros((p, p));
        for i in 0..p {
            for j in 0..rank {
                qs[(i, j)] = self.q_pen[(i, j)];
            }
            for j in 0..null_count {
                qs[(i, rank + j)] = self.q_null[(i, j)];
            }
        }
        qs
    }
}

/// Lambda-independent reparameterization invariants derived from penalty structure.
#[derive(Clone)]
pub struct ReparamInvariant {
    split: SubspaceSplit,
    /// The balanced eigenvector matrix Q (p x p). Block-local roots are
    /// transformed on-the-fly as `R_block @ Q[start..end, :]` instead of
    /// storing pre-multiplied full-width roots.
    qs_base: Array2<f64>,
    has_nonzero: bool,
    /// Largest eigenvalue of the balanced (unit-Frobenius) penalty matrix.
    /// Used as the scale reference for the shrinkage floor.
    max_balanced_eigenvalue: f64,
}

/// Precompute the lambda-invariant reparameterization structure from canonical penalties.
///
/// Uses block-local roots directly instead of requiring rank x p global roots.
/// Each `CanonicalPenalty` carries its own block-local root and column range,
/// so the balanced sum can be assembled without ever materializing full-size
/// penalty matrices.
pub fn precompute_reparam_invariant_from_canonical(
    penalties: &[CanonicalPenalty],
    p_total: usize,
) -> Result<ReparamInvariant, EstimationError> {
    use std::cmp::Ordering;

    let m = penalties.len();

    if m == 0 {
        return Ok(ReparamInvariant {
            split: SubspaceSplit::identity(p_total),
            qs_base: Array2::eye(p_total),
            has_nonzero: false,
            max_balanced_eigenvalue: 0.0,
        });
    }

    // Group penalties by col_range to detect block-diagonal structure.
    struct PenRef {
        penalty_index: usize,
    }
    let mut block_groups: BTreeMap<(usize, usize), Vec<PenRef>> = BTreeMap::new();
    let mut has_nonzero = false;
    for (i, cp) in penalties.iter().enumerate() {
        if cp.rank() == 0 {
            continue;
        }
        let local = cp.local_ref();
        let frob_norm = local.iter().map(|&x| x * x).sum::<f64>().sqrt();
        if frob_norm > 1e-12 {
            has_nonzero = true;
        }
        let key = (cp.col_range.start, cp.col_range.end);
        block_groups
            .entry(key)
            .or_default()
            .push(PenRef { penalty_index: i });
    }

    if !has_nonzero {
        return Ok(ReparamInvariant {
            split: SubspaceSplit::identity(p_total),
            qs_base: Array2::eye(p_total),
            has_nonzero: false,
            max_balanced_eigenvalue: 0.0,
        });
    }

    // Check for overlapping ranges.
    let ranges: Vec<(usize, usize)> = block_groups.keys().copied().collect();
    let mut overlapping = false;
    for i in 1..ranges.len() {
        if ranges[i].0 < ranges[i - 1].1 {
            overlapping = true;
            break;
        }
    }

    if overlapping {
        // Mirror the dense-fallback guard from
        // `create_balanced_penalty_root_from_canonical`. Without this, large-scale-
        // scale models with overlapping penalties allocated a full
        // p_total × p_total workspace and ran an O(p³) eigendecomposition
        // before any solver code saw the problem size.
        if p_total > OVERLAPPING_PENALTY_DENSE_FALLBACK_MAX_P {
            return Err(EstimationError::LayoutError(format!(
                "overlapping penalty reparameterization would require dense {}x{} eigendecomposition; \
                 large-model dense fallback is disabled. Keep penalties structured or \
                 extend the overlapping-penalty solver path",
                p_total, p_total
            )));
        }
        // Fallback: global p×p eigendecomposition.
        let balanced = balanced_penalty_sum(
            penalties
                .iter()
                .filter(|cp| cp.rank() > 0)
                .map(|cp| (cp.local_ref().view(), cp.col_range.clone())),
            p_total,
        );
        let s_balanced = array_to_faer(&balanced);
        let (bal_eigenvalues, bal_eigenvectors) =
            robust_eigh_faer(&s_balanced, Side::Lower, "balanced penalty matrix")?;

        let mut order: Vec<usize> = (0..p_total).collect();
        order.sort_by(|&i, &j| {
            bal_eigenvalues[j]
                .partial_cmp(&bal_eigenvalues[i])
                .unwrap_or(Ordering::Equal)
                .then(i.cmp(&j))
        });

        let mut qs = Mat::<f64>::zeros(p_total, p_total);
        for (col_idx, &idx) in order.iter().enumerate() {
            for row in 0..p_total {
                qs[(row, col_idx)] = bal_eigenvectors[(row, idx)];
            }
        }

        let max_bal = order
            .iter()
            .map(|&idx| bal_eigenvalues[idx].abs())
            .fold(0.0_f64, f64::max);
        let rank_tol = balanced_penalty_rank_tolerance(max_bal);
        let penalized_rank = order
            .iter()
            .take_while(|&&idx| bal_eigenvalues[idx] > rank_tol)
            .count();
        let split = SubspaceSplit::from_ordered_qs(&qs, penalized_rank, p_total)?;

        return Ok(ReparamInvariant {
            split,
            qs_base: mat_to_array(&qs),
            has_nonzero,
            max_balanced_eigenvalue: max_bal,
        });
    }

    // -----------------------------------------------------------------------
    // Non-overlapping: block-diagonal eigendecomposition at O(Σ p_k³).
    // -----------------------------------------------------------------------
    // The balanced sum is block-diagonal ⟹ its eigenvectors are block-local.
    // Q_pen and Q_null are assembled by embedding block-local eigenvectors.

    // Track which columns are covered by any penalty.
    let mut covered = vec![false; p_total];
    for cp in penalties {
        for j in cp.col_range.clone() {
            covered[j] = true;
        }
    }
    let uncovered_cols: Vec<usize> = (0..p_total).filter(|j| !covered[*j]).collect();

    struct BlockResult {
        col_range: Range<usize>,
        q_pen_local: Array2<f64>,  // block_dim × pen_rank
        q_null_local: Array2<f64>, // block_dim × null_rank
        /// Largest balanced eigenvalue contributed by this block.
        max_balanced_eigenvalue: f64,
        /// Column offset of this block's penalized directions within global Q_pen.
        pen_col_offset: usize,
        /// Column offset of this block's null directions within global Q_null.
        null_col_offset: usize,
    }

    // BTreeMap iteration defines the deterministic block order; collecting the
    // indexed parallel iterator preserves that order while eigendecomposing each
    // independent canonical penalty block concurrently.
    let block_specs: Vec<_> = block_groups.iter().collect();
    let mut block_results: Vec<BlockResult> = block_specs
        .into_par_iter()
        .map(
            |(&(start, end), refs)| -> Result<BlockResult, EstimationError> {
                let block_dim = end - start;

                // Build the local balanced sum under the ONE shared rule
                // (gam#2454); a component the rule keeps has unit Frobenius
                // norm, so the block is non-zero iff any entry is.
                let s_balanced_local = balanced_penalty_sum(
                    refs.iter().map(|pref| {
                        (
                            penalties[pref.penalty_index].local_ref().view(),
                            0..block_dim,
                        )
                    }),
                    block_dim,
                );
                let block_has_nonzero = s_balanced_local.iter().any(|&value| value != 0.0);

                if !block_has_nonzero {
                    return Ok(BlockResult {
                        col_range: start..end,
                        q_pen_local: Array2::zeros((block_dim, 0)),
                        q_null_local: Array2::eye(block_dim),
                        max_balanced_eigenvalue: 0.0,
                        pen_col_offset: 0,  // set later
                        null_col_offset: 0, // set later
                    });
                }

                // Eigendecompose the local balanced penalty.
                let (bal_eigenvalues, bal_eigenvectors) =
                    robust_eigh(&s_balanced_local, Side::Lower, "balanced penalty block")?;

                let mut order: Vec<usize> = (0..block_dim).collect();
                order.sort_by(|&i, &j| {
                    bal_eigenvalues[j]
                        .partial_cmp(&bal_eigenvalues[i])
                        .unwrap_or(Ordering::Equal)
                        .then(i.cmp(&j))
                });

                let max_bal = order
                    .iter()
                    .map(|&idx| bal_eigenvalues[idx].abs())
                    .fold(0.0_f64, f64::max);
                let rank_tol = balanced_penalty_rank_tolerance(max_bal);
                let penalized_rank = order
                    .iter()
                    .take_while(|&&idx| bal_eigenvalues[idx] > rank_tol)
                    .count();
                let null_count = block_dim - penalized_rank;

                let mut q_pen_local = Array2::zeros((block_dim, penalized_rank));
                let mut q_null_local = Array2::zeros((block_dim, null_count));
                for (col_idx, &idx) in order.iter().enumerate() {
                    if col_idx < penalized_rank {
                        for row in 0..block_dim {
                            q_pen_local[[row, col_idx]] = bal_eigenvectors[[row, idx]];
                        }
                    } else {
                        let null_col = col_idx - penalized_rank;
                        for row in 0..block_dim {
                            q_null_local[[row, null_col]] = bal_eigenvectors[[row, idx]];
                        }
                    }
                }

                Ok(BlockResult {
                    col_range: start..end,
                    q_pen_local,
                    q_null_local,
                    max_balanced_eigenvalue: max_bal,
                    pen_col_offset: 0,  // set later
                    null_col_offset: 0, // set later
                })
            },
        )
        .collect::<Result<_, _>>()?;
    let global_max_bal = block_results
        .iter()
        .map(|br| br.max_balanced_eigenvalue)
        .fold(0.0_f64, f64::max);

    // Compute column offsets for each block in the global Q_pen / Q_null layout.
    let total_pen_rank: usize = block_results.iter().map(|br| br.q_pen_local.ncols()).sum();
    let total_null: usize = block_results
        .iter()
        .map(|br| br.q_null_local.ncols())
        .sum::<usize>()
        + uncovered_cols.len();
    {
        let mut pen_off = 0usize;
        let mut null_off = 0usize;
        for br in &mut block_results {
            br.pen_col_offset = pen_off;
            br.null_col_offset = null_off;
            pen_off += br.q_pen_local.ncols();
            null_off += br.q_null_local.ncols();
        }
    }

    let mut q_pen = Array2::zeros((p_total, total_pen_rank));
    let mut q_null = Array2::zeros((p_total, total_null));

    for br in &block_results {
        let start = br.col_range.start;
        let bd = br.q_pen_local.nrows();
        let pen_r = br.q_pen_local.ncols();
        let null_r = br.q_null_local.ncols();
        if pen_r > 0 {
            q_pen
                .slice_mut(s![
                    start..(start + bd),
                    br.pen_col_offset..(br.pen_col_offset + pen_r)
                ])
                .assign(&br.q_pen_local);
        }
        if null_r > 0 {
            q_null
                .slice_mut(s![
                    start..(start + bd),
                    br.null_col_offset..(br.null_col_offset + null_r)
                ])
                .assign(&br.q_null_local);
        }
    }
    let mut null_col = block_results
        .iter()
        .map(|br| br.q_null_local.ncols())
        .sum::<usize>();
    for &j in &uncovered_cols {
        q_null[[j, null_col]] = 1.0;
        null_col += 1;
    }

    let split = SubspaceSplit { q_pen, q_null };

    // Store the global Q_s = [Q_pen | Q_null] from the split.
    // Block-local roots are transformed on-the-fly as R_block @ Q[start..end, :]
    // inside the reparam engine, avoiding O(k * rank * p) storage.
    let qs_global = split.compose_qs();

    Ok(ReparamInvariant {
        split,
        qs_base: qs_global,
        has_nonzero,
        max_balanced_eigenvalue: global_max_bal,
    })
}

/// Apply stable reparameterization using precomputed lambda-invariant structures.
///
/// Write the penalized-block spectrum and rotation from a right-singular
/// factorization of the stacked scaled roots `E = [√λₖ Rₖ]ₖ`.
///
/// Both admissible routes to that factorization — the SVD of `E` and the SVD of
/// its Householder QR factor `R` — produce the identical pair, because
/// `EᵀE = RᵀR`. Absorbing them through one function is what makes "the same two
/// outputs by a different computation" a fact of the code rather than a claim
/// about it.
fn absorb_right_singular_factorization(
    singular_values: &Array1<f64>,
    vt: &Array2<f64>,
    penalized_rank: usize,
    range_eigenvalues_sorted: &mut Vec<f64>,
    range_rotation: &mut Mat<f64>,
) {
    // `singular_values` descending → eigenvalues `d_i = σ_i²` descending;
    // `vt` is `penalized_rank × penalized_rank` with row i = vᵢᵀ.
    let n_sv = singular_values.len().min(penalized_rank).min(vt.nrows());
    *range_eigenvalues_sorted = (0..penalized_rank)
        .map(|i| {
            if i < n_sv {
                let s = singular_values[i];
                s * s
            } else {
                0.0
            }
        })
        .collect();
    for col_idx in 0..n_sv {
        for row in 0..penalized_rank {
            range_rotation[(row, col_idx)] = vt[[col_idx, row]];
        }
    }
}

/// The facts a refusing stacked-root SVD was decided against (#2465): the shape
/// it was handed, whether that input was even finite, its magnitude, and the λ
/// dynamic range that set it. A refusal naming only "no convergence" cannot
/// distinguish a genuinely unusable pencil from an iteration that gave up on a
/// well-formed one, and those two want opposite responses.
fn describe_stacked_roots(e_stacked: &Array2<f64>, lambdas: &[f64]) -> String {
    let mut max_abs = 0.0_f64;
    let mut nonfinite = 0usize;
    for &value in e_stacked.iter() {
        if value.is_finite() {
            max_abs = max_abs.max(value.abs());
        } else {
            nonfinite += 1;
        }
    }
    let finite_lambdas: Vec<f64> = lambdas.iter().copied().filter(|l| l.is_finite()).collect();
    let lambda_min = finite_lambdas.iter().copied().fold(f64::INFINITY, f64::min);
    let lambda_max = finite_lambdas
        .iter()
        .copied()
        .fold(f64::NEG_INFINITY, f64::max);
    format!(
        "rows={} cols={} nonfinite_entries={} max_abs={:.6e} n_lambda={} lambda_min={:.6e} \
         lambda_max={:.6e}",
        e_stacked.nrows(),
        e_stacked.ncols(),
        nonfinite,
        max_abs,
        lambdas.len(),
        lambda_min,
        lambda_max,
    )
}

pub fn stable_reparameterizationwith_invariant(
    penalties: &[CanonicalPenalty],
    lambdas: &[f64],
    p: usize,
    invariant: &ReparamInvariant,
) -> Result<ReparamResult, EstimationError> {
    let m = penalties.len();

    if lambdas.len() != m {
        return Err(EstimationError::ParameterConstraintViolation(format!(
            "Lambda count mismatch: expected {} lambdas for {} penalties, got {}",
            m,
            m,
            lambdas.len()
        )));
    }

    // No separate length check needed — penalties are matched against lambdas above,
    // and the invariant's qs_base is p x p (dimension-checked by the split).

    // #1074: the gam#1379 finite-ceiling on λ_k = exp(ρ_k) (clamp to 1e300 to
    // avoid `∞·0 = NaN` when the outer optimizer drives a redundant penalty
    // direction's log-λ past ~709) was DELETED. It masked the real defect: the
    // optimizer drives a redundant/unidentified penalty direction off to ∞
    // instead of that direction being detected and dropped from the model.
    // The root fix (detect+drop the redundant penalty direction at construction)
    // is tracked separately; λ now passes through raw.

    if m == 0 {
        return Ok(ReparamResult {
            s_transformed: Array2::zeros((p, p)),
            log_det: 0.0,
            det1: Array1::zeros(0),
            qs: Array2::eye(p),
            canonical_transformed: vec![],
            e_transformed: Array2::zeros((0, p)),
            // All modes truncated when no penalties; already in transformed frame.
            u_truncated: Array2::eye(p),
        });
    }

    if !invariant.has_nonzero {
        let qs = invariant.split.compose_qs();
        let u_truncated = qs.t().dot(&invariant.split.q_null);
        // All penalties are zero — canonical_transformed = originals (no rotation needed).
        let canonical_transformed: Vec<CanonicalPenalty> = penalties.to_vec();
        return Ok(ReparamResult {
            s_transformed: Array2::zeros((p, p)),
            log_det: 0.0,
            det1: Array1::zeros(m),
            qs,
            canonical_transformed,
            e_transformed: Array2::zeros((0, p)),
            u_truncated,
        });
    }

    let q_pen = array_to_faer(&invariant.split.q_pen);
    let q_null = array_to_faer(&invariant.split.q_null);
    let qs_base = array_to_faer(&invariant.qs_base);
    // Each penalty root transform is independent: R_k_block @ Q[start..end, :].
    // Run those per-penalty products (and their S_k = R_k'R_k caches) in
    // parallel, then collect in slice order so all downstream accumulation stays
    // deterministic and bit-for-bit stable with respect to penalty ordering.
    let penalty_transforms: Vec<(Mat<f64>, Mat<f64>)> = penalties
        .par_iter()
        .map(|cp| {
            let r = &cp.col_range;
            let root_faer = array_to_faer(&cp.root);
            let q_block = qs_base.submatrix(r.start, 0, cp.block_dim(), p);
            let mut product = Mat::<f64>::zeros(cp.rank(), p);
            matmul(
                product.as_mut(),
                Accum::Replace,
                root_faer.as_ref(),
                q_block,
                1.0,
                Par::Seq,
            );
            let s_k = penalty_from_root_faer(&product);
            (product, s_k)
        })
        .collect();
    let (rs_transformed, s_k_penalized_cache): (Vec<Mat<f64>>, Vec<Mat<f64>>) =
        penalty_transforms.into_iter().unzip();

    let penalized_rank = invariant.split.rank();

    let mut range_eigenvalues_sorted: Vec<f64> = Vec::new();
    let mut range_rotation = Mat::<f64>::zeros(penalized_rank, penalized_rank);
    if penalized_rank > 0 {
        // Penalized-block spectrum `Σ_k λ_k S_k = U diag(d) Uᵀ` (restricted to the
        // λ-invariant penalized subspace).  Compute it from the SVD of the STACKED
        // SCALED ROOTS `E = [√λ_k R_k]_k` (root rows stacked), NOT from an
        // eigendecomposition of the assembled Gram `range_block = EᵀE`.
        //
        // Why (#2123): `S_k = R_kᵀ R_k`, so `Σ_k λ_k S_k = EᵀE` with
        // `E = vstack_k(√λ_k R_k)`.  Assembling the Gram and eigendecomposing it
        // SQUARES the condition number (`κ(EᵀE) = κ(E)²`).  When the outer optimizer
        // drives one margin toward its null space the λ dynamic range is enormous
        // (here `te(x,z)` with the near-linear z axis reaches `λ_ratio ≳ 1e8`), so a
        // recessive-penalty eigenvalue `d_min ≈ λ_min·σ²` is swamped by the
        // eigensolver's `O(ε·d_max) = O(ε·λ_max)` absolute floor — its eigenVECTOR
        // rotates into numerical noise, the genuinely-penalized direction is lost
        // from `e_transformed`, and the inner P-IRLS solve then fits that direction
        // to the data (a WIGGLY β̂ with `βᵀSβ̂ ≈ 0` despite `λ → ∞`).  That silent
        // loss-of-penalty is a discontinuous function of ρ (it flips as the noise
        // floor crosses `d_min`), so it injects spurious cliffs into the REML/LAML
        // objective and a false low-cost basin in the high-λ corner — which the
        // outer optimizer then lands in for some training-row orders but not others
        // (the row-order-dependent EDF/SE of #2123).
        //
        // The SVD operates on `E` directly, so `σ_min(E) = √d_min` is resolved
        // whenever `√λ_min·σ ≳ ε·√λ_max·σ` — a λ dynamic range up to ~1e32 instead
        // of ~1e16 — keeping the reparameterized penalty faithful across the whole ρ
        // box and the outer objective smooth and permutation-invariant.  `EᵀE`
        // equals the previous `range_block` bit-for-bit, so well-conditioned fits
        // are unchanged; only the ill-conditioned corner is corrected.
        //
        // As before, the right singular vectors `V` (= the penalized-block rotation)
        // are used ONLY to build `E`/`S⁺`/traces below; they are NOT applied to
        // `q_pen` or `rs_transformed`, so `Q_s` stays λ-independent and the
        // quasi-Newton coordinate system does not drift at eigenvalue crossings.
        let total_root_rows: usize = rs_transformed.iter().map(Mat::nrows).sum();
        // Thin SVD yields a COMPLETE orthonormal `V` (all `penalized_rank`
        // directions, including exactly-zero σ) only when `E` is tall or square
        // (`total_root_rows ≥ penalized_rank`).  That always holds structurally —
        // the union of the penalty root ranges spans the penalized subspace — but a
        // pathological degenerate layout is handled by falling back to the Gram
        // eigendecomposition so `range_rotation` is never left rank-deficient.
        //
        // ROUTE LADDER (#2581).  The shape test gates only the ATTEMPT, not the
        // choice.  Both routes below compute the SAME two outputs
        // (`range_eigenvalues_sorted`, `range_rotation`), so a stacked-root SVD
        // that fails to CONVERGE is exactly the pathological case the Gram route
        // was written for: non-convergence is a property of the bidiagonal
        // iteration, not evidence that the pencil is unusable.  Selecting on the
        // shape alone turned that into a fatal `LayoutError` raised one branch
        // away from a trusted routine for the same quantity, aborting a whole
        // converging fit (measured: nottem `cc(month, k=12)`, one 75% partition,
        // λ ≈ 5, an 11×11 `E` — the outer BFGS had already certified a nearby ρ
        // before the refinement pass hit it).
        //
        // The ladder has three rungs, in accuracy order: the direct SVD of `E`;
        // the R-SVD of its Householder QR factor, which is EXACTLY as accurate
        // and merely a different computation; and only then the Gram route,
        // whose real cost is the squared condition number.  Reaching either
        // lower rung is REPORTED, so the route that produced the answer is never
        // silently substituted.
        let mut have_rotation = false;
        let mut rescued_by_r_svd = false;
        let mut svd_refusal: Option<String> = None;
        if total_root_rows >= penalized_rank {
            let mut e_stacked = Array2::<f64>::zeros((total_root_rows, penalized_rank));
            let mut row_off = 0usize;
            for (lambda, root) in lambdas.iter().zip(rs_transformed.iter()) {
                let sqrt_lambda = lambda.max(0.0).sqrt();
                let rk = root.nrows();
                for r in 0..rk {
                    for c in 0..penalized_rank {
                        e_stacked[[row_off + r, c]] = sqrt_lambda * root[(r, c)];
                    }
                }
                row_off += rk;
            }
            match e_stacked.svd(false, true) {
                Ok((_, singular_values, Some(vt))) => {
                    absorb_right_singular_factorization(
                        &singular_values,
                        &vt,
                        penalized_rank,
                        &mut range_eigenvalues_sorted,
                        &mut range_rotation,
                    );
                    have_rotation = true;
                }
                direct => {
                    let facts = describe_stacked_roots(&e_stacked, lambdas);
                    svd_refusal = Some(match direct {
                        Err(err) => format!("failed: {err:?}; {facts}"),
                        _ => format!("returned no right singular vectors; {facts}"),
                    });
                    // RUNG 2, and the reason the Gram route is a LAST resort
                    // rather than the only alternative.  `E = QR` by Householder
                    // reflections is direct — it has no convergence criterion to
                    // miss — and `EᵀE = RᵀR`, so `R`'s right singular vectors
                    // and singular values ARE `E`'s.  It therefore delivers the
                    // same two outputs at the same `O(ε²·d_max)` resolution the
                    // direct SVD promises, on a `penalized_rank`-square problem
                    // instead of a `total_root_rows`-tall one.
                    //
                    // Measured on the refusing input (nottem `cc(month, k=12)`,
                    // split 1, an 11×11 `E`, one λ = 6.068, no non-finite
                    // entries): a power-of-two rescale of `E` refuses
                    // identically — so the failure is not a scaling artefact —
                    // while this route and `svd(Eᵀ)` both converge and agree on
                    // `σ₀ = 1.653863e0`.
                    if let Ok((_, r_factor)) = e_stacked.qr()
                        && let Ok((_, singular_values, Some(vt))) = r_factor.svd(false, true)
                    {
                        absorb_right_singular_factorization(
                            &singular_values,
                            &vt,
                            penalized_rank,
                            &mut range_eigenvalues_sorted,
                            &mut range_rotation,
                        );
                        have_rotation = true;
                        rescued_by_r_svd = true;
                    }
                }
            }
        }
        if let Some(reason) = svd_refusal.as_deref() {
            if rescued_by_r_svd {
                log::warn!(
                    "penalized-block rotation: stacked-root SVD {reason}. Recovered the SAME \
                     right-singular basis from the Householder QR of `E` followed by the SVD \
                     of its triangular factor `R`: `EᵀE = RᵀR`, so no accuracy is given up."
                );
            } else {
                // The accuracy downgrade is observable rather than silent: the
                // Gram route resolves a recessive penalized eigenvalue only down
                // to `O(ε·d_max)`, where the SVD of `E` reaches `O(ε²·d_max)`.
                log::warn!(
                    "penalized-block rotation: stacked-root SVD {reason}, and so did the R-SVD \
                     of its Householder QR factor. Recomputing it from the Gram `Σₖ λₖ Sₖ`, \
                     which squares the condition number: recessive eigenvalues are resolved to \
                     O(ε·d_max) rather than O(ε²·d_max)."
                );
            }
        }
        if !have_rotation {
            // Assemble the Gram and eigendecompose it. This route serves a layout
            // whose penalty roots cannot span the penalized subspace
            // (`total_root_rows < penalized_rank`) AND a stacked-root SVD that
            // did not converge.
            let mut range_block = Mat::<f64>::zeros(penalized_rank, penalized_rank);
            for (lambda, s_k) in lambdas.iter().zip(s_k_penalized_cache.iter()) {
                for i in 0..penalized_rank {
                    for j in 0..penalized_rank {
                        range_block[(i, j)] += *lambda * s_k[(i, j)];
                    }
                }
            }
            let (range_eigenvalues, range_eigenvectors) =
                robust_eigh_faer(&range_block, Side::Lower, "range penalty block")?;
            let mut range_order: Vec<usize> = (0..penalized_rank).collect();
            range_order.sort_by(|&i, &j| {
                range_eigenvalues[j]
                    .partial_cmp(&range_eigenvalues[i])
                    .unwrap_or(std::cmp::Ordering::Equal)
                    .then(i.cmp(&j))
            });
            range_eigenvalues_sorted = range_order
                .iter()
                .map(|&idx| range_eigenvalues[idx])
                .collect();
            for (col_idx, &idx) in range_order.iter().enumerate() {
                for row in 0..penalized_rank {
                    range_rotation[(row, col_idx)] = range_eigenvectors[(row, idx)];
                }
            }
        }
    }

    // Subspace-invariant penalty spectral calculus:
    // - Penalized and null spaces are fixed by the lambda-invariant basis `qs_base`.
    // - Runtime lambda dependence only appears in the penalized block eigenvalues.
    // This avoids basis mixing inside the degenerate zero-eigenspace.
    let structural_rank = penalized_rank;
    let range_eigs_sorted: Vec<f64> = range_eigenvalues_sorted;

    let eigenvalue_floor = invariant.max_balanced_eigenvalue.max(1.0) * 1e-12;
    let qs = compose_qs_from_split(&q_pen, &q_null, p);

    // Guard against any accidental penalized/null mixing. The transformed penalty
    // roots must have negligible support on null columns by construction.
    let leakage = assess_subspace_leakage(&qs, &rs_transformed, structural_rank, p);
    if !subspace_split_is_consistent(&leakage, p) {
        return Err(EstimationError::LayoutError(format!(
            "Reparameterization subspace split is inconsistent: max null leakage {:.3e} (rel {:.3e}, worst penalty {}), max |Qp'Qn| {:.3e}",
            leakage.max_abs_sq.sqrt(),
            leakage.max_rel_sq.sqrt(),
            leakage.worst_penalty,
            leakage.max_cross_gram_abs,
        )));
    }

    // Truncated basis in transformed coordinates:
    //   U_⊥^(t) = Qs^T U_⊥^(orig) = Qs^T Q_n.
    let mut u_truncated_mat = Mat::<f64>::zeros(p, q_null.ncols());
    matmul(
        u_truncated_mat.as_mut(),
        Accum::Replace,
        qs.transpose(),
        q_null.as_ref(),
        1.0,
        Par::Seq,
    );

    // E is represented in TRANSFORMED coordinates (beta_t).  Because the
    // penalized subspace is NOT rotated by the lambda-dependent eigenvectors
    // (to keep Q_s stable across BFGS iterations), E is no longer diagonal.
    // Instead E = diag(√d) · U' embedded in structural_rank × p, so that
    // E'E = U diag(d) U' = Σ λ_k S_k in the invariant penalized basis.
    let mut e_transformed_mat = Mat::<f64>::zeros(structural_rank, p);
    for row_idx in 0..structural_rank {
        let safe_eigenval = range_eigs_sorted[row_idx].max(eigenvalue_floor);
        let sqrt_eigenval = safe_eigenval.sqrt();
        // E[row, j] = sqrt(d_row) * U'[row, j] = sqrt(d_row) * U[j, row]
        for j in 0..penalized_rank {
            e_transformed_mat[(row_idx, j)] = sqrt_eigenval * range_rotation[(j, row_idx)];
        }
    }

    // Pseudo-logdet on the structural penalized block.  The null block is split
    // out above, so there is no nullspace normalization here.  Spectrally-noisy
    // directions (eigenvalues snapped to 0 by `classify_eigenvalues_strict`
    // because they fall below the `c * eps_machine * p * scale` tolerance in
    // the lambda-weighted sum) are floored to `eigenvalue_floor` to keep the
    // log-det finite and consistent with the floored values used to construct
    // `e_transformed_mat` above.  This avoids spurious P-IRLS failures when the
    // lambda dynamic range is wide (e.g. during BFGS line search probing extreme
    // rho candidates).  Materially negative or non-finite spectra are already
    // rejected by the strict classifier upstream; this loop re-checks the
    // *post-shrinkage* range eigenvalues against the same floor.
    //
    // The same floored spectrum is used in the trace formula tr(S⁺ S_k) below,
    // matching the rank structure embedded in `e_transformed_mat` and avoiding
    // a 1/0 in the trace contraction when an eigenvalue was floored to 0.
    let mut floored_eigs: Vec<f64> = Vec::with_capacity(range_eigs_sorted.len());
    let mut log_det_sum = KahanSum::default();
    for (idx, &ev) in range_eigs_sorted.iter().enumerate() {
        if !ev.is_finite() || ev < -eigenvalue_floor {
            return Err(EstimationError::LayoutError(format!(
                "Penalty pseudo-logdet has a non-finite or large-negative structural eigenvalue at index {idx}: {ev:.3e}"
            )));
        }
        let safe_ev = ev.max(eigenvalue_floor);
        floored_eigs.push(safe_ev);
        if idx < penalized_rank {
            log_det_sum.add(safe_ev.ln());
        }
    }
    let log_det = log_det_sum.sum();
    let delta = 0.0;

    // The det1 contractions are independent once the eigensystem is fixed.  Use
    // indexed parallel collection so the output vector preserves lambda order.
    let det1vec: Vec<f64> = (0..lambdas.len())
        .into_par_iter()
        .map(|k| {
            let s_k = &s_k_penalized_cache[k];
            // Compute tr((S+δI)⁻¹ S_k) in the range eigenbasis without ever
            // materializing (S+δI)⁻¹. Using faer's matmul keeps this contraction
            // aligned with the orthogonal-similarity debug reference path.
            let trace = trace_penalty_in_orthogonal_basis(
                s_k,
                penalized_rank,
                &range_rotation,
                &floored_eigs,
                delta,
            );
            lambdas[k] * trace
        })
        .collect();

    {
        // Guardrail: cross-check the primary Rayleigh-quotient contraction
        // against a full orthogonal similarity transform, while staying in
        // the same numerically stable eigenbasis coordinates.
        let mut maxdet1_mismatch = 0.0_f64;
        let mut det1_scale = 0.0_f64;
        for (k, lambda) in lambdas.iter().enumerate() {
            let s_k_penalized = &s_k_penalized_cache[k];
            let s_k_eigenbasis = orthogonal_similarity_transform_faer(
                s_k_penalized,
                penalized_rank,
                &range_rotation,
            );
            let mut trace = KahanSum::default();
            for l in 0..penalized_rank {
                trace.add(s_k_eigenbasis[(l, l)] / (floored_eigs[l] + delta));
            }
            let reference = *lambda * trace.sum();
            maxdet1_mismatch = maxdet1_mismatch.max((reference - det1vec[k]).abs());
            det1_scale = det1_scale.max(reference.abs()).max(det1vec[k].abs());
        }
        let det1_tolerance = 1e-7 * det1_scale.max(1.0);
        assert!(
            maxdet1_mismatch <= det1_tolerance,
            "det1 mismatch between optimized and reference formulas: max_abs={maxdet1_mismatch:.3e}, tol={det1_tolerance:.3e}"
        );
    }

    // Rebuild s_transformed from e_transformed to ensure rank consistency.
    //
    // The sum of λ*S_k may contain numerical noise modes (eigenvalues ~1e-15) that
    // become significant when λ is large (e.g., 10^12). These modes would appear in H
    // but are truncated from log|S|_+, creating a "phantom penalty" in the objective.
    //
    // By reconstructing s_transformed = E^T * E, we force the penalty matrix used
    // in H to have the EXACT same rank structure as the one used for log|S|_+.
    // Any mode truncated from the prior is now strictly zero in the Hessian
    // calculation, ensuring mathematical consistency of the gradients.
    let mut s_truncated = Mat::<f64>::zeros(p, p);
    matmul(
        s_truncated.as_mut(),
        Accum::Replace,
        e_transformed_mat.transpose(),
        e_transformed_mat.as_ref(),
        1.0,
        Par::Seq,
    );

    {
        // Structural check: transformed S must not leak into declared null coordinates.
        let mut max_null_diag = 0.0_f64;
        let mut max_null_offdiag = 0.0_f64;
        for i in structural_rank..p {
            max_null_diag = max_null_diag.max(s_truncated[(i, i)].abs());
            for j in 0..p {
                if i != j {
                    max_null_offdiag = max_null_offdiag.max(s_truncated[(i, j)].abs());
                }
            }
        }
        assert!(
            max_null_diag <= 1e-10 && max_null_offdiag <= 1e-10,
            "null-space leakage in transformed penalty: max_null_diag={max_null_diag:.3e}, max_null_offdiag={max_null_offdiag:.3e}"
        );
    }

    let qs_array = mat_to_array(&qs);
    let canonical_transformed: Vec<CanonicalPenalty> = rs_transformed
        .par_iter()
        .zip(penalties.par_iter())
        .map(|(r, cp)| {
            let mean_transformed = qs_array.t().dot(&cp.full_width_prior_mean());
            CanonicalPenalty::from_dense_root_with_mean(mat_to_array(r), p, mean_transformed)
        })
        .collect();
    Ok(ReparamResult {
        s_transformed: mat_to_array(&s_truncated),
        log_det,
        det1: Array1::from(det1vec),
        qs: qs_array,
        canonical_transformed,
        e_transformed: mat_to_array(&e_transformed_mat),
        u_truncated: mat_to_array(&u_truncated_mat),
    })
}

/// Minimal engine layout descriptor that avoids domain-specific layout coupling.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EngineDims {
    pub p: usize,
    pub k: usize,
}

impl EngineDims {
    pub fn new(p: usize, k: usize) -> Self {
        Self { p, k }
    }
}

/// Engine-facing stable reparameterization API using only `(p, k)`.
///
/// When `cached_invariant` is `Some`, reuses the precomputed eigendecomposition
/// (the hot path inside the REML loop). When `None`, computes the invariant on
/// the fly (the post-REML refit path).
/// Stable reparameterization from block-local canonical penalties.
pub fn stable_reparameterization_engine_canonical(
    penalties: &[CanonicalPenalty],
    lambdas: &[f64],
    dims: EngineDims,
    cached_invariant: Option<&ReparamInvariant>,
) -> Result<ReparamResult, EstimationError> {
    let owned;
    let invariant = match cached_invariant {
        Some(inv) => inv,
        None => {
            owned = precompute_reparam_invariant_from_canonical(penalties, dims.p)?;
            &owned
        }
    };
    stable_reparameterizationwith_invariant(penalties, lambdas, dims.p, invariant)
}

// ---------------------------------------------------------------------------
// Kronecker-factored reparameterization for tensor-product smooths
// ---------------------------------------------------------------------------

/// Result of Kronecker-factored reparameterization.
///
/// Exploits the fact that for Kronecker-structured penalties, the joint
/// eigenvector matrix is `U_1 ⊗ ... ⊗ U_d` and the reparameterized design
/// is a rowwise Kronecker of `(B_k U_k)` — all remaining factored.
#[derive(Clone)]
pub struct KroneckerReparamResult {
    /// Reparameterized marginal designs: `B_k · U_k` for each marginal k.
    ///
    /// `Arc`-shared with the λ-invariant cache so the per-outer-iterate
    /// memoized engine bumps a refcount instead of deep-copying the
    /// (n × q) reparameterized marginals every call.
    pub reparameterized_marginals: Arc<Vec<Array2<f64>>>,
    /// Marginal eigenvalues from each marginal penalty eigendecomposition.
    pub marginal_eigenvalues: Arc<Vec<Array1<f64>>>,
    /// Marginal eigenvector matrices U_k.
    pub marginal_qs: Arc<Vec<Array2<f64>>>,
    /// log|S|₊ computed from marginal eigenvalue grid.
    pub log_det: f64,
    /// First derivatives of log|S|₊ w.r.t. ρ_k = log(λ_k).
    pub det1: Array1<f64>,
    /// Second derivatives of log|S|₊ w.r.t. ρ.
    pub det2: Array2<f64>,
    /// Whether a double penalty (global ridge) is present.
    pub has_double_penalty: bool,
    /// Marginal basis dimensions.
    pub marginal_dims: Vec<usize>,
}

impl KroneckerReparamResult {
    /// Materialize the joint Qs matrix (U_1 ⊗ ... ⊗ U_d) as dense p×p.
    /// Only for fallback paths — avoid in hot loops.
    pub fn materialize_qs(&self) -> Array2<f64> {
        let mut qs = Array2::<f64>::eye(1);
        for u_k in self.marginal_qs.iter() {
            qs = kronecker_product(&qs, u_k);
        }
        qs
    }

    /// Materialize s_transformed (the penalty in the reparameterized basis).
    /// In the eigenbasis, this is diagonal with entries Σ_k λ_k μ_{k,j_k}.
    pub fn materialize_s_transformed(&self, lambdas: &[f64]) -> Array2<f64> {
        let d = self.marginal_dims.len();
        let p: usize = self.marginal_dims.iter().copied().product();
        let mut s = Array2::<f64>::zeros((p, p));

        // Delegate the per-cell tensor-penalty accumulation to the shared
        // `kronecker_cell_sigma` (#1172/#1185 single source of truth). Fold the
        // `lambdas.len() > d` guard into `has_double` to preserve exact gating.
        let eigenvalue_views: Vec<ArrayView1<'_, f64>> =
            self.marginal_eigenvalues.iter().map(|m| m.view()).collect();
        let has_double = self.has_double_penalty && lambdas.len() > d;
        let mut multi_idx = vec![0usize; d];
        let mut flat = 0usize;
        loop {
            let (sigma, _structural_sigma, _joint_null) = kronecker_cell_sigma(
                &eigenvalue_views,
                &multi_idx,
                lambdas,
                d,
                has_double,
                0.0,
            );
            s[[flat, flat]] = sigma;
            flat += 1;

            if kronecker_multi_index_advance(&mut multi_idx, &self.marginal_dims) {
                break;
            }
        }
        s
    }

    /// Explicitly materialize the dense artifact bundle expected by legacy
    /// downstream consumers. This is not part of the native Kronecker solve path.
    pub fn materialize_dense_artifact_result(
        &self,
        rs_list: &[Array2<f64>],
        lambdas: &[f64],
        p: usize,
    ) -> Result<ReparamResult, EstimationError> {
        const KRONECKER_DENSE_COMPAT_FALLBACK_MAX_P: usize = 4096;
        if p > KRONECKER_DENSE_COMPAT_FALLBACK_MAX_P {
            return Err(EstimationError::LayoutError(format!(
                "Kronecker reparameterization would materialize dense {}x{} compatibility tensors; \
                 large-model dense fallback is disabled. Wire the downstream solver to consume \
                 the factored Kronecker result directly",
                p, p
            )));
        }
        let qs = self.materialize_qs();
        let s_transformed = self.materialize_s_transformed(lambdas);

        // Transform penalty roots: R_k_transformed = R_k · Qs
        let rs_transformed: Vec<Array2<f64>> = if rs_list.len() >= 2 {
            use rayon::prelude::*;
            rs_list
                .par_iter()
                .map(|r| gam_linalg::faer_ndarray::fast_ab(r, &qs))
                .collect()
        } else {
            rs_list
                .iter()
                .map(|r| gam_linalg::faer_ndarray::fast_ab(r, &qs))
                .collect()
        };
        // rs_transposed removed — canonical_transformed is the single source of truth.

        // Build e_transformed: combined penalty square root in transformed coords.
        // For Kronecker structure, the penalty is diagonal in the eigenbasis.
        // e_transformed rows are the nonzero rows of sqrt(Σ_k λ_k S_k)^{1/2}.
        let d = self.marginal_dims.len();
        // Delegate the per-cell tensor-penalty accumulation to the shared
        // `kronecker_cell_sigma` (the #1172/#1185 single source of truth). The
        // double-penalty term is only valid when `lambdas` actually carries the
        // λ_d entry, so fold the original `lambdas.len() > d` guard into the
        // `has_double_penalty` flag passed to the helper — preserving the exact
        // gating behavior.
        let eigenvalue_views: Vec<ArrayView1<'_, f64>> =
            self.marginal_eigenvalues.iter().map(|m| m.view()).collect();
        let has_double = self.has_double_penalty && lambdas.len() > d;
        let diag_vals: Vec<f64> = {
            let mut vals = Vec::with_capacity(p);
            let mut multi_idx = vec![0usize; d];
            loop {
                let (sigma, _structural_sigma, _joint_null) = kronecker_cell_sigma(
                    &eigenvalue_views,
                    &multi_idx,
                    lambdas,
                    d,
                    has_double,
                    0.0,
                );
                vals.push(if sigma > 0.0 { sigma.sqrt() } else { 0.0 });

                if kronecker_multi_index_advance(&mut multi_idx, &self.marginal_dims) {
                    break;
                }
            }
            vals
        };
        let rank = diag_vals.iter().filter(|&&v| v > 1e-12).count();
        let mut e_transformed = Array2::<f64>::zeros((rank, p));
        let mut row = 0;
        for (j, &v) in diag_vals.iter().enumerate() {
            if v > 1e-12 {
                e_transformed[[row, j]] = v;
                row += 1;
            }
        }

        // u_truncated: null-space eigenvectors (columns with zero eigenvalue).
        let null_count = p - rank;
        let mut u_truncated = Array2::<f64>::zeros((p, null_count));
        let mut col = 0;
        for (j, &v) in diag_vals.iter().enumerate() {
            if v <= 1e-12 {
                u_truncated[[j, col]] = 1.0; // standard basis vector in eigenbasis
                col += 1;
            }
        }

        let canonical_transformed: Vec<CanonicalPenalty> = rs_transformed
            .iter()
            .map(|r| CanonicalPenalty::from_dense_root(r.clone(), p))
            .collect();
        Ok(ReparamResult {
            s_transformed,
            log_det: self.log_det,
            det1: self.det1.clone(),
            qs,
            canonical_transformed,
            e_transformed,
            u_truncated,
        })
    }
}

/// Compute `log|S|₊` and its first/second derivatives w.r.t. `ρ_k = log(λ_k)`
/// from factored marginal eigenvalues.
///
/// Shared implementation for `KroneckerPenaltySystem::logdet_and_derivatives`
/// and `kronecker_reparameterization_engine`.  Iterates over the ∏q_j
/// multi-index grid in O(d · ∏q_j) time with no O(p²) storage.
const KRONECKER_STRUCTURAL_ZERO_TOL: f64 = 1e-12;

/// Per-cell Kronecker eigenvalue accumulation — the single source of truth for
/// the #1172/#1185 tensor-penalty math.
///
/// For the multi-index cell `multi_idx`, accumulates:
///   - `sigma`            = Σ_k λ_k · μ_k  (+ joint-null double-penalty term + declared objective ridge)
///   - `structural_sigma` = Σ_k μ_k        (unweighted; classifies joint-null cells)
///   - `joint_null`       = whether the cell lies in the joint null space
///
/// `marginal_eigenvalues[k][multi_idx[k]]` is the k-th marginal eigenvalue μ_k.
/// The double-penalty (global ridge) term `λ_d` is added only on joint-null
/// cells; an explicit objective ridge is added only on structurally penalized
/// cells. This mirrors the gated logic fixed in #1172/#1185 and MUST be kept
/// identical across every caller.
#[inline]
fn kronecker_cell_sigma(
    marginal_eigenvalues: &[ArrayView1<'_, f64>],
    multi_idx: &[usize],
    lambdas: &[f64],
    d: usize,
    has_double_penalty: bool,
    objective_ridge: f64,
) -> (f64, f64, bool) {
    let mut sigma = 0.0;
    let mut structural_sigma = 0.0;
    for k in 0..d {
        let marginal_eigenvalue = marginal_eigenvalues[k][multi_idx[k]];
        structural_sigma += marginal_eigenvalue;
        sigma += lambdas[k] * marginal_eigenvalue;
    }
    let joint_null = structural_sigma <= KRONECKER_STRUCTURAL_ZERO_TOL;
    if has_double_penalty && joint_null {
        sigma += lambdas[d];
    }
    if structural_sigma > KRONECKER_STRUCTURAL_ZERO_TOL {
        sigma += objective_ridge;
    }
    (sigma, structural_sigma, joint_null)
}

/// Advance a row-major multi-index over the `dims` grid in place.
/// Returns `true` when the grid is exhausted (the index wrapped back to all-zero).
#[inline]
pub(crate) fn kronecker_multi_index_advance(multi_idx: &mut [usize], dims: &[usize]) -> bool {
    let mut carry = true;
    for dim in (0..dims.len()).rev() {
        if carry {
            multi_idx[dim] += 1;
            if multi_idx[dim] < dims[dim] {
                carry = false;
            } else {
                multi_idx[dim] = 0;
            }
        }
    }
    carry
}

pub fn kronecker_logdet_and_derivatives(
    marginal_eigenvalues: &[ArrayView1<'_, f64>],
    marginal_dims: &[usize],
    lambdas: &[f64],
    has_double_penalty: bool,
    objective_ridge: f64,
) -> (f64, Array1<f64>, Array2<f64>) {
    let d = marginal_dims.len();
    let n_pen = d + if has_double_penalty { 1 } else { 0 };

    let mut logdet = 0.0;
    let mut grad = Array1::<f64>::zeros(n_pen);
    let mut hess = Array2::<f64>::zeros((n_pen, n_pen));
    let tol = 1e-12;

    let mut multi_idx = vec![0usize; d];
    loop {
        let (sigma, _structural_sigma, joint_null) = kronecker_cell_sigma(
            marginal_eigenvalues,
            &multi_idx,
            lambdas,
            d,
            has_double_penalty,
            objective_ridge,
        );

        if sigma > tol {
            logdet += sigma.ln();
            let inv_sigma = 1.0 / sigma;
            let inv_sigma2 = inv_sigma * inv_sigma;

            for k in 0..d {
                let ck = lambdas[k] * marginal_eigenvalues[k][multi_idx[k]];
                grad[k] += ck * inv_sigma;
            }
            if has_double_penalty && joint_null {
                grad[d] += lambdas[d] * inv_sigma;
            }

            for k in 0..n_pen {
                let ck = if k < d {
                    lambdas[k] * marginal_eigenvalues[k][multi_idx[k]]
                } else if joint_null {
                    lambdas[d]
                } else {
                    0.0
                };
                // When ck == 0 (a zero λ, a zero marginal eigenvalue, or a cell
                // outside the joint null for the ridge penalty) every term this
                // index k contributes — `ck·inv_sigma − ck²·inv_sigma2` on the
                // diagonal and `−ck·cl·inv_sigma2` on every off-diagonal — is
                // exactly 0.0, so adding them to the finite running accumulators
                // is a bit-identical no-op. Skip the inner sweep entirely.
                if ck == 0.0 {
                    continue;
                }
                hess[[k, k]] += ck * inv_sigma - ck * ck * inv_sigma2;
                for l in (k + 1)..n_pen {
                    let cl = if l < d {
                        lambdas[l] * marginal_eigenvalues[l][multi_idx[l]]
                    } else if joint_null {
                        lambdas[d]
                    } else {
                        0.0
                    };
                    let off = -ck * cl * inv_sigma2;
                    hess[[k, l]] += off;
                    hess[[l, k]] += off;
                }
            }
        }

        if kronecker_multi_index_advance(&mut multi_idx, marginal_dims) {
            break;
        }
    }

    (logdet, grad, hess)
}

// #1521: `KroneckerInvariantStructure` is defined once in `crate::kronecker`
// (the leaf data+compute module). The byte-identical copy that the carve left
// here is replaced by an import so the cache and this engine share one type.
use crate::kronecker::KroneckerInvariantStructure;

/// Kronecker-factored reparameterization reusing a precomputed λ-invariant
/// structure (eigensystems, reparameterized marginals, shrinkage scale).
///
/// Bit-identical to `kronecker_reparameterization_engine` for the same marginal
/// data — the only difference is that the `eigh()` / `B_k U_k` work was hoisted
/// out of the per-iterate path into the cached `invariant`. Only the λ-dependent
/// `kronecker_logdet_and_derivatives` sweep runs here.
pub fn kronecker_reparameterization_engine_with_invariant(
    invariant: &KroneckerInvariantStructure,
    marginal_dims: &[usize],
    lambdas: &[f64],
    has_double_penalty: bool,
) -> Result<KroneckerReparamResult, EstimationError> {
    // Arc refcount bumps — the underlying eigensystems / reparameterized
    // marginals are λ-invariant and shared with the cache, not deep-copied.
    let marginal_eigenvalues = Arc::clone(&invariant.marginal_eigenvalues);
    let marginal_qs = Arc::clone(&invariant.marginal_qs);
    let reparameterized_marginals = Arc::clone(&invariant.reparameterized_marginals);

    let marginal_eigenvalue_views: Vec<_> = marginal_eigenvalues
        .iter()
        .map(|evals| evals.view())
        .collect();
    let (log_det, det1, det2) = kronecker_logdet_and_derivatives(
        &marginal_eigenvalue_views,
        marginal_dims,
        lambdas,
        has_double_penalty,
        0.0,
    );

    Ok(KroneckerReparamResult {
        reparameterized_marginals,
        marginal_eigenvalues,
        marginal_qs,
        log_det,
        det1,
        det2,
        has_double_penalty,
        marginal_dims: marginal_dims.to_vec(),
    })
}

#[cfg(test)]
mod tests {
    /// gam#2454: the reparameterization's penalized rank and the shared
    /// balanced structural rank are one number. Three unit directions penalized
    /// at Frobenius norms `1e6`, `1e-9` and `1` (overlapping full-width
    /// components, the dense fallback path): the unweighted sum would drop the
    /// `1e-9` direction at a `100·p·ε·max` cut; the balanced rule keeps all three
    /// and the split's `q_pen` spans all three.
    #[test]
    fn reparam_split_rank_is_the_balanced_structural_rank_2454() {
        use ndarray::array;
        let p = 3usize;
        let roots = [
            array![[1.0e3, 0.0, 0.0]],
            array![[0.0, 1.0e-9_f64.sqrt(), 0.0]],
            array![[0.0, 0.0, 1.0]],
        ];
        let penalties: Vec<super::CanonicalPenalty> = roots
            .iter()
            .map(|root| super::CanonicalPenalty::from_dense_root(root.clone(), p))
            .collect();
        let balanced = super::balanced_penalty_structural_rank(
            penalties
                .iter()
                .map(|cp| (cp.local_ref().view(), cp.col_range.clone())),
            p,
        )
        .expect("balanced rank");
        assert_eq!(balanced, 3);
        let invariant = super::precompute_reparam_invariant_from_canonical(&penalties, p)
            .expect("reparam invariant");
        assert_eq!(
            invariant.split.q_pen.ncols(),
            balanced,
            "the split's penalized subspace has the balanced structural rank"
        );
        assert_eq!(invariant.split.q_null.ncols(), 0);
    }

    use super::{
        CanonicalPenalty, REL_PSD_FLOOR, SubspaceLeakageMetrics, assess_subspace_leakage,
        classify_eigenvalues_strict, precompute_reparam_invariant_from_canonical,
        report_penalty_pair_redundancy, stable_reparameterizationwith_invariant,
        subspace_split_is_consistent,
    };
    use crate::EstimationError;
    use faer::Mat;
    use gam_linalg::utils::inf_norm;
    use ndarray::{Array1, Array2, array};

    /// Build CanonicalPenalty values from full-width roots for tests.
    fn canonical_from_roots(rs_list: &[Array2<f64>], p: usize) -> Vec<CanonicalPenalty> {
        rs_list
            .iter()
            .map(|r| {
                let local = r.t().dot(r);
                CanonicalPenalty {
                    root: r.clone(),
                    col_range: 0..p,
                    total_dim: p,
                    nullity: 0,
                    local,
                    prior_mean: Array1::zeros(p),
                    positive_eigenvalues: Vec::new(),
                    op: None,
                }
            })
            .collect()
    }

    fn metrics_for(
        qs: &Mat<f64>,
        rs: &[Mat<f64>],
        structural_rank: usize,
        p: usize,
    ) -> SubspaceLeakageMetrics {
        assess_subspace_leakage(qs, rs, structural_rank, p)
    }

    #[test]
    fn subspace_leakage_iszero_for_clean_split() {
        let p = 4usize;
        let structural_rank = 2usize;
        let qs = Mat::<f64>::identity(p, p);
        let mut r0 = Mat::<f64>::zeros(2, p);
        r0[(0, 0)] = 1.0;
        r0[(1, 1)] = 2.0;

        let m = metrics_for(&qs, &[r0], structural_rank, p);
        assert!(m.max_abs_sq <= 1e-16);
        assert!(m.max_rel_sq <= 1e-16);
        assert!(m.max_cross_gram_abs <= 1e-16);
    }

    #[test]
    fn subspace_leakage_detects_null_column_energy() {
        let p = 4usize;
        let structural_rank = 2usize;
        let qs = Mat::<f64>::identity(p, p);
        let mut r0 = Mat::<f64>::zeros(1, p);
        r0[(0, 2)] = 3.0;

        let m = metrics_for(&qs, &[r0], structural_rank, p);
        assert!(m.max_abs_sq > 0.0);
        assert!(m.max_rel_sq > 0.99);
    }

    #[test]
    fn subspace_leakage_detects_qp_qn_nonorthogonality() {
        let p = 3usize;
        let structural_rank = 1usize;
        let mut qs = Mat::<f64>::identity(p, p);
        qs[(0, 1)] = 0.2;
        let r0 = Mat::<f64>::zeros(1, p);

        let m = metrics_for(&qs, &[r0], structural_rank, p);
        assert!(m.max_cross_gram_abs > 1e-3);
    }

    #[test]
    fn subspace_split_admits_near_threshold_manifold_leakage_1802() {
        // #1802: on sphere / Duchon / spline-on-sphere bases the REML outer
        // startup rejected EVERY candidate seed with
        //   "Reparameterization subspace split is inconsistent:
        //    max null leakage 1.174e-4 (rel 1.031e-4, worst penalty 0),
        //    max |Qp'Qn| 4.316e-16"
        // The split is perfectly ORTHOGONAL (|Qp'Qn| ≈ 4e-16); the only tripped
        // quantity is the transformed-root null-block leakage, whose RELATIVE
        // energy (1.031e-4)² ≈ 1.06e-8 sits right at `REL_PSD_FLOOR`. That is the
        // eigensolver's own numerically-PSD floor — the same floor
        // `classify_eigenvalues_strict` uses to declare a mode null — so a
        // manifold penalty whose Laplace-Beltrami spectrum decays through the
        // rank threshold with no clean gap MUST be admitted. Reproduce that exact
        // signature and assert the guard accepts it.
        let p = 40usize;
        let structural_rank = p - 1;
        // Unit-amplitude range column + a null column at √(REL_PSD_FLOOR)
        // amplitude, so the null-block relative energy lands at ~REL_PSD_FLOOR.
        let null_amp = (1.06e-8_f64).sqrt();
        let mut rs = Mat::<f64>::zeros(1, p);
        rs[(0, 0)] = 1.0;
        rs[(0, p - 1)] = null_amp;
        let qs = Mat::<f64>::identity(p, p);
        let leakage = metrics_for(&qs, &[rs], structural_rank, p);
        // The leakage reproduces the observed band: above the OLD fixed 1e-10
        // tolerance (which rejected the sphere fit) yet a benign ~REL_PSD_FLOOR.
        assert!(
            leakage.max_rel_sq > 1e-10 && leakage.max_rel_sq < 1e-6,
            "reproduced leakage should sit in the near-REL_PSD_FLOOR band, got {:.3e}",
            leakage.max_rel_sq
        );
        assert!(leakage.max_cross_gram_abs <= 1e-12);
        assert!(
            subspace_split_is_consistent(&leakage, p),
            "near-REL_PSD_FLOOR null leakage on a manifold basis must be admitted \
             (rel_sq={:.3e}, tol={:.3e})",
            leakage.max_rel_sq,
            (p as f64) * REL_PSD_FLOOR,
        );
    }

    #[test]
    fn subspace_split_still_rejects_genuine_inconsistency_1802() {
        // The #1802 relaxation only widens the leakage tolerance to the
        // classifier's own `p · REL_PSD_FLOOR` floor; it must NOT admit a
        // genuinely broken split. A whole penalized mode dumped into the null
        // block (O(1) relative leakage) is still rejected.
        let p = 40usize;
        let structural_rank = p - 1;
        let mut rs = Mat::<f64>::zeros(1, p);
        rs[(0, p - 1)] = 1.0; // all penalty-root energy in the null column
        let qs = Mat::<f64>::identity(p, p);
        let leakage = metrics_for(&qs, &[rs], structural_rank, p);
        assert!(leakage.max_rel_sq > 0.99);
        assert!(
            !subspace_split_is_consistent(&leakage, p),
            "an O(1) null-block leakage is a real inconsistency and must be rejected"
        );

        // A non-orthogonal split is rejected regardless of root leakage.
        let mut qs_bad = Mat::<f64>::identity(3, 3);
        qs_bad[(0, 1)] = 0.2;
        let clean = Mat::<f64>::zeros(1, 3);
        let leakage2 = metrics_for(&qs_bad, &[clean], 1, 3);
        assert!(leakage2.max_cross_gram_abs > 1e-3);
        assert!(
            !subspace_split_is_consistent(&leakage2, 3),
            "a non-orthogonal Qp/Qn split must be rejected"
        );
    }

    #[test]
    fn u_truncated_is_transformed_frame_in_nonzero_case() {
        let p = 3usize;
        let rs_list = vec![array![[1.0, 0.0, 0.0]]];
        let canonical = canonical_from_roots(&rs_list, p);
        let lambdas = vec![2.0];
        let inv = precompute_reparam_invariant_from_canonical(&canonical, p)
            .expect("precompute invariant");
        let rep = stable_reparameterizationwith_invariant(&canonical, &lambdas, p, &inv)
            .expect("stable reparam");

        let expected = rep.qs.t().dot(&inv.split.q_null);
        let diff = &rep.u_truncated - &expected;
        let max_abs = inf_norm(diff.iter().copied());
        assert!(
            max_abs <= 1e-10,
            "u_truncated frame mismatch: max_abs={max_abs}"
        );
    }

    #[test]
    fn infinite_lambda_keeps_range_penalty_block_finite_1379() {
        // gam#1379 / gam#1074: a genuinely infinite λ = exp(ρ) is NOT silently
        // clamped to a finite ceiling. The original #1379 fix added a 1e300
        // ceiling so `∞ · 0` could not poison the range block Σ_k λ_k S_k, but
        // #1074 DELETED that clamp on purpose (see the comment at the top of
        // `stable_reparameterizationwith_invariant`): masking ∞ hid the real
        // defect — the outer optimizer driving a redundant/unidentified penalty
        // direction off to ∞ instead of that direction being detected and
        // dropped. With the clamp gone, a literal `f64::INFINITY` λ surfaces as
        // a clean, detectable error (the eigensolver rejects the NaN-poisoned
        // block) rather than a silent finite success. Pin that contract: ∞ must
        // ERROR, not be quietly clamped.
        //
        // Fixture: two penalties on a 3-wide block. The first penalizes only
        // coordinate 0 (so its block S_k has structural zeros everywhere except
        // [0,0]); give it λ = +∞. The second penalizes coordinate 1 at a normal
        // λ.
        let p = 3usize;
        let rs_list = vec![array![[1.0, 0.0, 0.0]], array![[0.0, 1.0, 0.0]]];
        let canonical = canonical_from_roots(&rs_list, p);
        let inv = precompute_reparam_invariant_from_canonical(&canonical, p)
            .expect("precompute invariant");

        let lambdas_inf = vec![f64::INFINITY, 3.0];
        let inf_result =
            stable_reparameterizationwith_invariant(&canonical, &lambdas_inf, p, &inv);
        assert!(
            inf_result.is_err(),
            "an infinite lambda must surface as an error, not be silently clamped (#1074)"
        );

        // A finite (even very large) λ must still produce an all-finite reparam:
        // the function is robust to large-but-finite penalties; only the
        // non-finite input is rejected.
        let lambdas_big = vec![1e300_f64, 3.0];
        let rep = stable_reparameterizationwith_invariant(&canonical, &lambdas_big, p, &inv)
            .expect("stable reparam at large-but-finite lambda");
        assert!(
            rep.s_transformed.iter().all(|v| v.is_finite()),
            "transformed penalty must be finite at large-but-finite lambda"
        );
        assert!(
            rep.qs.iter().all(|v| v.is_finite()),
            "reparam rotation must be finite at large-but-finite lambda"
        );
        assert!(
            rep.log_det.is_finite(),
            "penalty log-det must be finite at large-but-finite lambda"
        );
        assert!(
            rep.det1.iter().all(|v| v.is_finite()),
            "penalty log-det derivatives must be finite at large-but-finite lambda"
        );
    }

    #[test]
    fn u_truncated_is_identitywhen_no_penalties() {
        let p = 4usize;
        let canonical: Vec<CanonicalPenalty> = Vec::new();
        let lambdas: Vec<f64> = Vec::new();
        let inv = precompute_reparam_invariant_from_canonical(&canonical, p)
            .expect("precompute invariant");
        let rep = stable_reparameterizationwith_invariant(&canonical, &lambdas, p, &inv)
            .expect("stable reparam");
        assert_eq!(rep.u_truncated, Array2::<f64>::eye(p));
    }

    #[test]
    fn transformed_penalty_is_diagonal_in_transformed_frame() {
        let p = 3usize;
        let inv_sqrt2 = 2.0_f64.sqrt().recip();
        // Penalize a rotated direction in original space so Qs is non-trivial.
        let rs_list = vec![array![[inv_sqrt2, inv_sqrt2, 0.0]]];
        let canonical = canonical_from_roots(&rs_list, p);
        let lambdas = vec![4.0];
        let inv = precompute_reparam_invariant_from_canonical(&canonical, p)
            .expect("precompute invariant");
        let rep = stable_reparameterizationwith_invariant(&canonical, &lambdas, p, &inv)
            .expect("stable reparam");

        assert_eq!(rep.e_transformed.nrows(), 1);
        assert!(rep.e_transformed[[0, 0]].abs() > 0.0);
        assert!(rep.e_transformed[[0, 1]].abs() <= 1e-12);
        assert!(rep.e_transformed[[0, 2]].abs() <= 1e-12);
        // Exact pseudo-logdet on the structural penalized block has no
        // delta-dependent nullspace normalization.
        let expected_det1 = 1.0_f64;
        assert!((rep.det1[0] - expected_det1).abs() <= 1e-12);

        let s = rep.s_transformed;
        let mut max_offdiag = 0.0_f64;
        for i in 0..p {
            for j in 0..p {
                if i != j {
                    max_offdiag = max_offdiag.max(s[[i, j]].abs());
                }
            }
        }
        assert!(
            max_offdiag <= 1e-10,
            "transformed penalty should be diagonal, max offdiag={max_offdiag}"
        );
        assert!(s[[1, 1]].abs() <= 1e-10);
        assert!(s[[2, 2]].abs() <= 1e-10);
    }

    #[test]
    fn det1_matches_rank_for_single_full_rank_penalty() {
        let p = 2usize;
        let inv_sqrt2 = 2.0_f64.sqrt().recip();
        // Q^T for a 45-degree rotation.
        let q_t = [[inv_sqrt2, inv_sqrt2], [-inv_sqrt2, inv_sqrt2]];
        // R = diag(3, 1) * Q^T gives S = Q * diag(9, 1) * Q^T.
        let rs = array![
            [3.0 * q_t[0][0], 3.0 * q_t[0][1]],
            [1.0 * q_t[1][0], 1.0 * q_t[1][1]]
        ];
        let rs_list = vec![rs];
        let canonical = canonical_from_roots(&rs_list, p);
        let lambdas = vec![5.0];

        let inv = precompute_reparam_invariant_from_canonical(&canonical, p)
            .expect("precompute invariant");
        let rep = stable_reparameterizationwith_invariant(&canonical, &lambdas, p, &inv)
            .expect("stable reparam");

        assert_eq!(rep.e_transformed.nrows(), p);
        let det1 = rep.det1[0];
        // Exact pseudo-logdet on the structural penalized block:
        //   det1 = lambda * sum_l d_l / (lambda*d_l)
        // where d_l are eigenvalues of S_k.
        let s_k_eigs = [9.0_f64, 1.0_f64];
        let lambda = 5.0_f64;
        let expected_det1: f64 = s_k_eigs.iter().map(|&d| lambda * d / (lambda * d)).sum();
        assert!(
            (det1 - expected_det1).abs() <= 1e-12,
            "expected det1={expected_det1}, got {det1}",
        );

        let s = rep.s_transformed;
        assert!(s[[0, 1]].abs() <= 1e-10);
        assert!(s[[1, 0]].abs() <= 1e-10);
        assert!(s[[0, 0]] > 0.0);
        assert!(s[[1, 1]] > 0.0);
    }

    #[test]
    fn classify_strict_rejects_nan_eigenvalue() {
        let mut eigs = [1.0, f64::NAN, 0.5];
        match classify_eigenvalues_strict(&mut eigs, "test_nan") {
            Err(EstimationError::PenaltySpectrumNonFinite {
                context,
                index,
                value,
            }) => {
                assert_eq!(context, "test_nan");
                assert_eq!(index, 1);
                assert!(value.is_nan());
            }
            other => panic!("expected PenaltySpectrumNonFinite, got {:?}", other),
        }
    }

    #[test]
    fn classify_strict_rejects_inf_eigenvalue() {
        let mut eigs = [1.0, 0.5, f64::INFINITY];
        match classify_eigenvalues_strict(&mut eigs, "test_inf") {
            Err(EstimationError::PenaltySpectrumNonFinite { index, value, .. }) => {
                assert_eq!(index, 2);
                assert!(value.is_infinite());
            }
            other => panic!("expected PenaltySpectrumNonFinite, got {:?}", other),
        }
    }

    #[test]
    fn classify_strict_rejects_materially_indefinite() {
        // -1e-2 with scale ~1.0 is well above any reasonable roundoff tolerance.
        let mut eigs = [1.0, -1e-2, 0.5];
        match classify_eigenvalues_strict(&mut eigs, "test_indef") {
            Err(EstimationError::PenaltySpectrumIndefinite {
                context,
                index,
                value,
                ..
            }) => {
                assert_eq!(context, "test_indef");
                assert_eq!(index, 1);
                assert!((value + 1e-2).abs() <= 1e-15);
            }
            other => panic!("expected PenaltySpectrumIndefinite, got {:?}", other),
        }
    }

    #[test]
    fn classify_strict_accepts_roundoff_negative() {
        // -1e-16 * scale is well within tol = 64 * eps * p * scale.
        let scale = 1.0_f64;
        let roundoff = -1e-16 * scale;
        let mut eigs = [scale, 0.5 * scale, roundoff, 0.25 * scale];
        classify_eigenvalues_strict(&mut eigs, "test_roundoff").expect("roundoff must classify");
        // The roundoff eigenvalue is snapped to exact zero.
        assert_eq!(eigs[2], 0.0);
        // Strictly positive entries must be preserved.
        assert!(eigs[0] > 0.0 && eigs[1] > 0.0 && eigs[3] > 0.0);
    }

    #[test]
    fn classify_strict_accepts_extreme_lambda_assembly_noise_1619() {
        // #1619: high-rank thin-plate / Duchon penalties (p≈200) assembled and
        // reparameterized at extreme λ produce float-noise negative eigenvalues at
        // ~1e-11 relative to the spectrum scale (~1e13 there). The bare machine-ε
        // floor (~12×ε relative ≈ 1e-12) rejected these as "indefinite" and
        // spuriously failed the inner P-IRLS solve. They are PSD to numerical
        // precision and must be snapped to zero, not rejected.
        let scale = 8.509e12_f64;
        // Worst (eig, scale) pair observed in the issue: eig/scale ≈ -7.7e-11.
        let noise = -6.546e2_f64;
        assert!(
            (noise.abs() / scale) < 1.0e-10,
            "fixture must reproduce the ~1e-11-relative noise from #1619"
        );
        let mut eigs = vec![scale, 0.5 * scale, noise, 0.1 * scale];
        classify_eigenvalues_strict(&mut eigs, "range penalty block")
            .expect("a ~1e-11-relative roundoff-negative eigenvalue must be accepted (#1619)");
        // The roundoff-negative eigenvalue is snapped to exact zero.
        assert_eq!(eigs[2], 0.0);
        // Strictly positive entries are preserved.
        assert!(eigs[0] > 0.0 && eigs[1] > 0.0 && eigs[3] > 0.0);
    }

    #[test]
    fn classify_strict_snaps_subtol_positive_to_zero() {
        // Positive eigenvalues below the tolerance are also snapped to exact 0
        // so downstream rank counts and pseudo-logdets are deterministic.
        let scale = 10.0_f64;
        let subtol = 1e-15 * scale;
        let mut eigs = [scale, subtol];
        classify_eigenvalues_strict(&mut eigs, "test_sub_pos").expect("sub-tol positive ok");
        assert_eq!(eigs[1], 0.0);
    }

    /// Build a `CanonicalPenalty` directly from a symmetric `local` matrix.
    /// Bypasses root extraction — the redundancy diagnostic only reads `local`
    /// and `col_range`, so the rest is filler.
    fn canonical_from_local(
        local: Array2<f64>,
        col_range: std::ops::Range<usize>,
        total_dim: usize,
    ) -> CanonicalPenalty {
        let block_dim = local.nrows();
        // A trivially valid root: zero rank. The diagnostic doesn't read root.
        let root = Array2::<f64>::zeros((0, block_dim));
        CanonicalPenalty {
            root,
            col_range,
            total_dim,
            nullity: 0,
            local,
            prior_mean: Array1::zeros(block_dim),
            positive_eigenvalues: Vec::new(),
            op: None,
        }
    }

    #[test]
    fn report_penalty_pair_redundancy_detects_identical_pair() {
        // Penalty 0: a "generic" SPD matrix on cols 0..3.
        let s0 = ndarray::array![[2.0, 0.5, 0.0], [0.5, 1.0, 0.25], [0.0, 0.25, 1.5],];
        // Penalties 1 and 2: identical block-local penalty on the SAME col_range.
        // This is the Z₂-symmetric saddle scenario.
        let s_shared = ndarray::array![[1.0, -0.5, 0.0], [-0.5, 2.0, -0.5], [0.0, -0.5, 1.0],];

        let bundle = vec![
            canonical_from_local(s0, 0..3, 3),
            canonical_from_local(s_shared.clone(), 0..3, 3),
            canonical_from_local(s_shared, 0..3, 3),
        ];

        let redundant = report_penalty_pair_redundancy(&bundle);

        // Exactly one redundant pair: (1, 2). Pairs (0, 1) and (0, 2) involve
        // distinct matrices and must NOT be flagged.
        assert_eq!(
            redundant.len(),
            1,
            "expected exactly one redundant pair, got {:?}",
            redundant
        );
        let (i, j, defect) = redundant[0];
        assert_eq!((i, j), (1, 2));
        assert_eq!(
            defect, 0.0,
            "bit-identical penalties leave an EXACTLY zero residual, not a small one"
        );
    }

    /// The regression this whole screen was rewritten for (#2676): a pair whose
    /// relative defect is `1.9e-5` — the measured `geo_disease_matern` figure at
    /// its cold geometry — must NOT be reported as proportional, even though its
    /// cosine rounds to `1.000000` at six decimals.
    ///
    /// The two assertions are a pair. The first is the fix; the second is the
    /// evidence that the OLD screen would have failed it, so the test cannot go
    /// green for the boring reason that the fixture is not marginal.
    #[test]
    fn a_pair_1p9e_minus_5_apart_is_not_proportional_2676() {
        const DEFECT: f64 = 1.874_020e-5;
        let base = ndarray::array![[1.0, -0.5, 0.0], [-0.5, 2.0, -0.5], [0.0, -0.5, 1.0]];
        let base_norm = base.iter().map(|v| v * v).sum::<f64>().sqrt();
        // A perturbation ORTHOGONAL to `base` in the Frobenius inner product,
        // so the defect is exactly the perturbation's relative size and no part
        // of it is absorbed into the best scale `c`.
        let mut direction = ndarray::array![[0.0, 0.0, 1.0], [0.0, 0.0, 0.0], [1.0, 0.0, 0.0]];
        let overlap = direction
            .iter()
            .zip(base.iter())
            .map(|(a, b)| a * b)
            .sum::<f64>()
            / (base_norm * base_norm);
        direction = &direction - &(overlap * &base);
        let direction_norm = direction.iter().map(|v| v * v).sum::<f64>().sqrt();
        let perturbed = &base + &(direction * (DEFECT * base_norm / direction_norm));

        let bundle = vec![
            canonical_from_local(base.clone(), 0..3, 3),
            canonical_from_local(perturbed.clone(), 0..3, 3),
        ];
        assert!(
            report_penalty_pair_redundancy(&bundle).is_empty(),
            "a pair {DEFECT:.3e} apart is measurably distinct and must not be called proportional"
        );

        // Negative control: the cosine the old screen thresholded on cannot see
        // it. `1 - cos = defect^2/2 = 1.8e-10`, six orders under the `1e-8` bar
        // that used to declare the pair "structurally identical".
        let inner = base
            .iter()
            .zip(perturbed.iter())
            .map(|(a, b)| a * b)
            .sum::<f64>();
        let cosine =
            inner / (base_norm * perturbed.iter().map(|v| v * v).sum::<f64>().sqrt());
        assert!(
            cosine > 1.0 - 1e-8,
            "the fixture must be one the OLD cos > 1 - 1e-8 screen accepted; got cos = {cosine}"
        );
        assert_eq!(
            format!("{cosine:.6}"),
            "1.000000",
            "and one whose six-decimal print is indistinguishable from an exact identity"
        );
    }

    #[test]
    fn report_penalty_pair_redundancy_skips_different_col_ranges() {
        // Two identical local matrices but on disjoint col_ranges. The
        // function must NOT flag them — they live in different parameter
        // subspaces by construction.
        let s = ndarray::array![[1.0, 0.0], [0.0, 1.0]];
        let bundle = vec![
            canonical_from_local(s.clone(), 0..2, 4),
            canonical_from_local(s, 2..4, 4),
        ];
        let redundant = report_penalty_pair_redundancy(&bundle);
        assert!(
            redundant.is_empty(),
            "different col_ranges must not be flagged"
        );
    }
}