Skip to main content

gam_models/
marginal_slope_shared.rs

1//! Shared kernels and outer-evaluation infrastructure for the
2//! marginal-slope family of GAMs (BMS, survival, latent survival).
3//!
4//! # Outer-row subsampling
5//!
6//! At large scale (n ≥ tens of thousands) the outer rho-gradient is
7//! a sum-over-rows trace whose per-row cost is dominated by the cubic
8//! cell-moment kernel. The pieces here — [`AutoOuterSubsampleOptions`],
9//! [`auto_outer_score_subsample`], [`maybe_install_auto_outer_subsample`],
10//! and [`build_outer_score_subsample`] — implement a stratified
11//! Horvitz–Thompson estimator that replaces the full row sum with an
12//! unbiased sample, gated by
13//! [`crate::custom_family::BlockwiseFitOptions::auto_outer_subsample`]
14//! and enabled by default for large marginal-slope fits.
15//!
16//! `maybe_install_auto_outer_subsample` is the entry point family
17//! impls call: it consults the per-family phase counter and the
18//! per-family last-ρ mutex (used to detect distinct outer steps),
19//! installs a stratified mask for the first
20//! `BMS_AUTO_SUBSAMPLE_PHASE1_BUDGET` (or family analog) outer
21//! evaluations, and reverts to full data afterward so the BFGS/ARC
22//! convergence target `outer_tol` is reached on exact gradients
23//! rather than chasing the stochastic noise floor.
24//!
25//! This subsampling is **complementary** to the trace-estimator tier
26//! system documented at the top of `solver::reml::reml_outer_engine` (exact /
27//! Hutchinson multi-target / Hutch++ single-target). They operate on
28//! orthogonal axes — the trace estimators reduce work *within* the
29//! Hessian structure for a fixed row set; subsampling reduces the row
30//! set itself for the family-specific row-trace path.
31
32use crate::cubic_cell_kernel::{self, DenestedPartitionCell, LocalSpanCubic};
33use crate::custom_family::{CustomFamilyBlockPsiDerivative, ParameterBlockSpec};
34use crate::outer_subsample::{OuterScoreSubsample, WeightedOuterRow};
35use gam_math::jet_scalar::{JetScalar, OneSeed, Order2, TwoSeed};
36use gam_math::nested_dual::JetField;
37use ndarray::{Array1, Array2, Axis};
38use std::ops::Range;
39use std::sync::Arc;
40
41/// Canonical inner-cache `beta_seed` validator passed to the generic
42/// outer-engine (`optimize_spatial_length_scale_exact_joint`).
43///
44/// The outer solver hands back the converged inner `beta` at each accepted
45/// ρ-step so the next inner solve can warm-start from it. This guards that
46/// cached vector for non-finite entries (which would poison the warm start)
47/// and, when clean, stashes it into the caller's `pending` cell.
48///
49/// This is the single source of truth for the seed callback: every family
50/// that wires up the exact-joint outer engine (survival location-scale,
51/// bernoulli marginal-slope, survival marginal-slope) routes through here
52/// instead of re-deriving the identical closure, which previously drifted in
53/// error construction (`EstimationError::InvalidInput(...)` vs
54/// `bail_invalid_estim!`).
55pub fn make_beta_seed_validator(
56    pending: &std::cell::RefCell<Option<Array1<f64>>>,
57) -> impl FnMut(
58    &Array1<f64>,
59) -> Result<gam_solve::rho_optimizer::SeedOutcome, crate::model_types::EstimationError>
60+ '_ {
61    move |beta: &Array1<f64>| {
62        bail_if_cached_beta_non_finite(beta)?;
63        // Stage the seed for promotion at the next eval, where the freshly
64        // built per-block widths are known. A width mismatch is reconciled
65        // there (the eval's `from_cached_beta` logs and falls back to a cold
66        // β for that step) — never an error that aborts the fit. Staging a
67        // finite β always succeeds, so the contract reply is `Installed`.
68        pending.replace(Some(beta.clone()));
69        Ok(gam_solve::rho_optimizer::SeedOutcome::Installed)
70    }
71}
72
73/// Canonical non-finite guard on a cached inner `beta`.
74///
75/// Single source of truth for the `"cached inner beta contains non-finite
76/// entries"` check + error: the full seed closure
77/// ([`make_beta_seed_validator`]) and the bare warm-start length-then-finite
78/// guards in `custom_family` all route through this so the predicate and the
79/// error construction (`EstimationError::InvalidInput`) never drift apart.
80pub use gam_problem::bail_if_cached_beta_non_finite;
81
82#[inline]
83pub const fn eval_coeff4_at(coefficients: &[f64; 4], z: f64) -> f64 {
84    ((coefficients[3] * z + coefficients[2]) * z + coefficients[1]) * z + coefficients[0]
85}
86
87#[inline]
88pub fn add_scaled_coeff4(target: &mut [f64; 4], source: &[f64; 4], scale: f64) {
89    for j in 0..4 {
90        target[j] += scale * source[j];
91    }
92}
93
94#[inline]
95fn coeff4_dot(left: &[f64; 4], right: &[f64; 4]) -> f64 {
96    left[0] * right[0] + left[1] * right[1] + left[2] * right[2] + left[3] * right[3]
97}
98
99#[inline]
100pub const fn scale_coeff4(source: [f64; 4], scale: f64) -> [f64; 4] {
101    [
102        source[0] * scale,
103        source[1] * scale,
104        source[2] * scale,
105        source[3] * scale,
106    ]
107}
108
109pub fn probit_frailty_scale(gaussian_frailty_sd: Option<f64>) -> f64 {
110    let sigma = gaussian_frailty_sd.unwrap_or(0.0);
111    if sigma <= 0.0 {
112        1.0
113    } else {
114        crate::survival::lognormal_kernel::ProbitFrailtyScaleJet::from_log_sigma(sigma.ln()).s
115    }
116}
117
118/// One auxiliary-parameter channel's objective, full primary gradient, and
119/// symmetric primary Hessian.
120pub(crate) struct DirectionalPrimaryTerms {
121    pub(crate) objective: f64,
122    pub(crate) grad: Array1<f64>,
123    pub(crate) hess: Array2<f64>,
124}
125
126fn order2_primary_terms<const K: usize>(channel: Order2<K>) -> DirectionalPrimaryTerms {
127    let gradient = channel.g();
128    let hessian = channel.h();
129    DirectionalPrimaryTerms {
130        objective: channel.value(),
131        grad: Array1::from_vec(gradient.to_vec()),
132        hess: Array2::from_shape_fn((K, K), |(row, column)| hessian[row][column]),
133    }
134}
135
136/// Differentiate a generic order-two primary program once with respect to one
137/// auxiliary scalar parameter.
138///
139/// The primary variables carry their ordinary value/gradient/Hessian seeds in
140/// the [`OneSeed::base`] channel while `parameter` carries only the auxiliary
141/// derivative in [`OneSeed::eps`]. The resulting epsilon `Order2` therefore is
142/// the auxiliary derivative of the objective, its complete primary gradient,
143/// and its complete primary Hessian. This replaces the old
144/// `1 + K + K(K+1)/2` separate bitmask-jet evaluations with one evaluation of
145/// the same row expression.
146pub(crate) fn first_parameter_order2_terms<const K: usize, Eval>(
147    primaries: [f64; K],
148    parameter: f64,
149    parameter_first: f64,
150    eval: Eval,
151) -> Result<DirectionalPrimaryTerms, String>
152where
153    Eval: FnOnce(&[OneSeed<K>; K], &OneSeed<K>) -> Result<OneSeed<K>, String>,
154{
155    let zero = <Order2<K> as JetScalar<K>>::constant(0.0);
156    let variables = std::array::from_fn(|axis| OneSeed {
157        base: <Order2<K> as JetScalar<K>>::variable(primaries[axis], axis),
158        eps: zero,
159    });
160    let parameter_jet = OneSeed {
161        base: <Order2<K> as JetScalar<K>>::constant(parameter),
162        eps: <Order2<K> as JetScalar<K>>::constant(parameter_first),
163    };
164    eval(&variables, &parameter_jet).map(|output| order2_primary_terms(output.eps))
165}
166
167/// Differentiate a generic order-two primary program once to second order in
168/// one auxiliary scalar parameter.
169///
170/// Both nilpotent directions represent the same external coordinate. Seeding
171/// `eps_del` with the parameter's own second derivative supplies the complete
172/// chain rule, and the output `eps_del` channel contains the second auxiliary
173/// derivative of `(objective, primary gradient, primary Hessian)`.
174pub(crate) fn second_parameter_order2_terms<const K: usize, Eval>(
175    primaries: [f64; K],
176    parameter: f64,
177    parameter_first: f64,
178    parameter_second: f64,
179    eval: Eval,
180) -> Result<DirectionalPrimaryTerms, String>
181where
182    Eval: FnOnce(&[TwoSeed<K>; K], &TwoSeed<K>) -> Result<TwoSeed<K>, String>,
183{
184    let zero = <Order2<K> as JetScalar<K>>::constant(0.0);
185    let variables = std::array::from_fn(|axis| TwoSeed {
186        base: <Order2<K> as JetScalar<K>>::variable(primaries[axis], axis),
187        eps: zero,
188        del: zero,
189        eps_del: zero,
190    });
191    let parameter_jet = TwoSeed {
192        base: <Order2<K> as JetScalar<K>>::constant(parameter),
193        eps: <Order2<K> as JetScalar<K>>::constant(parameter_first),
194        del: <Order2<K> as JetScalar<K>>::constant(parameter_first),
195        eps_del: <Order2<K> as JetScalar<K>>::constant(parameter_second),
196    };
197    eval(&variables, &parameter_jet).map(|output| order2_primary_terms(output.eps_del))
198}
199
200/// Differentiate an order-two primary program once in an auxiliary parameter
201/// and once along a primary-space direction, in one [`TwoSeed`] evaluation.
202/// The mixed `eps_del` `Order2` supplies the directional derivative of the
203/// auxiliary objective/gradient/Hessian without rebuilding the row program for
204/// every unit primary axis.
205pub(crate) fn first_parameter_directional_order2_terms<const K: usize, Eval>(
206    primaries: [f64; K],
207    direction: &[f64; K],
208    parameter: f64,
209    parameter_first: f64,
210    eval: Eval,
211) -> Result<DirectionalPrimaryTerms, String>
212where
213    Eval: FnOnce(&[TwoSeed<K>; K], &TwoSeed<K>) -> Result<TwoSeed<K>, String>,
214{
215    let zero = <Order2<K> as JetScalar<K>>::constant(0.0);
216    let variables = std::array::from_fn(|axis| TwoSeed {
217        base: <Order2<K> as JetScalar<K>>::variable(primaries[axis], axis),
218        eps: zero,
219        del: <Order2<K> as JetScalar<K>>::constant(direction[axis]),
220        eps_del: zero,
221    });
222    let parameter_jet = TwoSeed {
223        base: <Order2<K> as JetScalar<K>>::constant(parameter),
224        eps: <Order2<K> as JetScalar<K>>::constant(parameter_first),
225        del: zero,
226        eps_del: zero,
227    };
228    eval(&variables, &parameter_jet).map(|output| order2_primary_terms(output.eps_del))
229}
230
231fn zero_local_span_cubic() -> LocalSpanCubic {
232    LocalSpanCubic {
233        left: 0.0,
234        right: 1.0,
235        c0: 0.0,
236        c1: 0.0,
237        c2: 0.0,
238        c3: 0.0,
239    }
240}
241
242pub(crate) fn build_denested_partition_cells(
243    a: f64,
244    b: f64,
245    score_warp: Option<&crate::bms::DeviationRuntime>,
246    beta_h: Option<&Array1<f64>>,
247    link_dev: Option<&crate::bms::DeviationRuntime>,
248    beta_w: Option<&Array1<f64>>,
249    scale: f64,
250) -> Result<Vec<DenestedPartitionCell>, String> {
251    let score_breaks = score_warp
252        .map(|runtime| runtime.breakpoints().to_vec())
253        .unwrap_or_default();
254    let link_breaks = link_dev
255        .map(|runtime| runtime.breakpoints().to_vec())
256        .unwrap_or_default();
257
258    let mut cells = cubic_cell_kernel::build_denested_partition_cells_with_tails(
259        a,
260        b,
261        &score_breaks,
262        &link_breaks,
263        |z| {
264            if let (Some(runtime), Some(beta)) = (score_warp, beta_h) {
265                runtime.local_cubic_at(beta.view(), z)
266            } else {
267                Ok(zero_local_span_cubic())
268            }
269        },
270        |u| {
271            if let (Some(runtime), Some(beta)) = (link_dev, beta_w) {
272                runtime.local_cubic_at(beta.view(), u)
273            } else {
274                Ok(zero_local_span_cubic())
275            }
276        },
277    )?;
278    if scale != 1.0 {
279        for partition_cell in &mut cells {
280            partition_cell.cell.c0 *= scale;
281            partition_cell.cell.c1 *= scale;
282            partition_cell.cell.c2 *= scale;
283            partition_cell.cell.c3 *= scale;
284        }
285    }
286    Ok(cells)
287}
288
289pub(crate) struct ObservedDenestedCellPartials {
290    pub(crate) coeff: [f64; 4],
291    pub(crate) dc_da: [f64; 4],
292    pub(crate) dc_db: [f64; 4],
293    pub(crate) dc_daa: [f64; 4],
294    pub(crate) dc_dab: [f64; 4],
295    pub(crate) dc_dbb: [f64; 4],
296    pub(crate) dc_daaa: [f64; 4],
297    pub(crate) dc_daab: [f64; 4],
298    pub(crate) dc_dabb: [f64; 4],
299    pub(crate) dc_dbbb: [f64; 4],
300}
301
302pub(crate) fn observed_denested_cell_partials(
303    z_obs: f64,
304    a: f64,
305    b: f64,
306    score_warp: Option<&crate::bms::DeviationRuntime>,
307    beta_h: Option<&Array1<f64>>,
308    link_dev: Option<&crate::bms::DeviationRuntime>,
309    beta_w: Option<&Array1<f64>>,
310    scale: f64,
311) -> Result<ObservedDenestedCellPartials, String> {
312    let zero_score_span = zero_local_span_cubic();
313    let zero_link_span = zero_local_span_cubic();
314    let u_obs = a + b * z_obs;
315    let score_span_obs = if let (Some(runtime), Some(beta_h)) = (score_warp, beta_h) {
316        runtime.local_cubic_at(beta_h.view(), z_obs)?
317    } else {
318        zero_score_span
319    };
320    let link_span_obs = if let (Some(runtime), Some(beta_w)) = (link_dev, beta_w) {
321        runtime.local_cubic_at(beta_w.view(), u_obs)?
322    } else {
323        zero_link_span
324    };
325    let coeff = scale_coeff4(
326        cubic_cell_kernel::denested_cell_coefficients(score_span_obs, link_span_obs, a, b),
327        scale,
328    );
329    let (dc_da_raw, dc_db_raw) =
330        cubic_cell_kernel::denested_cell_coefficient_partials(score_span_obs, link_span_obs, a, b);
331    let (dc_daa_raw, dc_dab_raw, dc_dbb_raw) =
332        cubic_cell_kernel::denested_cell_second_partials(score_span_obs, link_span_obs, a, b);
333    let (dc_daaa, dc_daab, dc_dabb, dc_dbbb) =
334        cubic_cell_kernel::denested_cell_third_partials(link_span_obs);
335    Ok(ObservedDenestedCellPartials {
336        coeff,
337        dc_da: scale_coeff4(dc_da_raw, scale),
338        dc_db: scale_coeff4(dc_db_raw, scale),
339        dc_daa: scale_coeff4(dc_daa_raw, scale),
340        dc_dab: scale_coeff4(dc_dab_raw, scale),
341        dc_dbb: scale_coeff4(dc_dbb_raw, scale),
342        dc_daaa: scale_coeff4(dc_daaa, scale),
343        dc_daab: scale_coeff4(dc_daab, scale),
344        dc_dabb: scale_coeff4(dc_dabb, scale),
345        dc_dbbb: scale_coeff4(dc_dbbb, scale),
346    })
347}
348
349pub(crate) fn add_two_surface_psi_outer(
350    block_i: usize,
351    psi_row_i: &Array1<f64>,
352    block_j: usize,
353    psi_row_j: &Array1<f64>,
354    alpha: f64,
355    marginal_block: usize,
356    logslope_block: usize,
357    h_mm: &mut Array2<f64>,
358    h_gg: &mut Array2<f64>,
359    h_mg: &mut Array2<f64>,
360) {
361    if alpha == 0.0 {
362        return;
363    }
364    let col_i = psi_row_i.view().insert_axis(Axis(1));
365    let row_j = psi_row_j.view().insert_axis(Axis(0));
366
367    if block_i == block_j {
368        let col_j = psi_row_j.view().insert_axis(Axis(1));
369        let row_i = psi_row_i.view().insert_axis(Axis(0));
370        let target = match block_i {
371            b if b == marginal_block => h_mm,
372            b if b == logslope_block => h_gg,
373            _ => return,
374        };
375        ndarray::linalg::general_mat_mul(alpha, &col_i, &row_j, 1.0, target);
376        ndarray::linalg::general_mat_mul(alpha, &col_j, &row_i, 1.0, target);
377    } else {
378        let (marginal_row, logslope_row) = if block_i == marginal_block {
379            (psi_row_i, psi_row_j)
380        } else {
381            (psi_row_j, psi_row_i)
382        };
383        let m_col = marginal_row.view().insert_axis(Axis(1));
384        let g_row = logslope_row.view().insert_axis(Axis(0));
385        ndarray::linalg::general_mat_mul(alpha, &m_col, &g_row, 1.0, h_mg);
386    }
387}
388
389pub(crate) fn add_optional_vector(left: &mut Option<Array1<f64>>, right: &Option<Array1<f64>>) {
390    if let (Some(left), Some(right)) = (left.as_mut(), right.as_ref()) {
391        *left += right;
392    }
393}
394
395pub(crate) fn add_optional_matrix(left: &mut Option<Array2<f64>>, right: &Option<Array2<f64>>) {
396    if let (Some(left), Some(right)) = (left.as_mut(), right.as_ref()) {
397        *left += right;
398    }
399}
400
401pub(crate) fn psi_derivative_location(
402    derivative_blocks: &[Vec<CustomFamilyBlockPsiDerivative>],
403    psi_index: usize,
404) -> Option<(usize, usize)> {
405    let mut cursor = 0usize;
406    for (block_idx, block) in derivative_blocks.iter().enumerate() {
407        if psi_index < cursor + block.len() {
408            return Some((block_idx, psi_index - cursor));
409        }
410        cursor += block.len();
411    }
412    None
413}
414
415/// Predicate used by every marginal-slope family's persistent-warm-start
416/// fingerprint guard: the caller's parameter blocks must each have row count
417/// matching the family's `n`, and the list must be non-empty.
418#[inline]
419pub(crate) fn parameter_block_specs_match_rows(
420    specs: &[ParameterBlockSpec],
421    expected_n: usize,
422) -> bool {
423    !specs.is_empty()
424        && specs
425            .iter()
426            .all(|spec| spec.design.nrows() == expected_n && spec.offset.len() == expected_n)
427}
428
429#[derive(Clone, Copy)]
430pub(crate) struct CoeffSupport {
431    pub(crate) include_primary: bool,
432    pub(crate) include_h: bool,
433    pub(crate) include_w: bool,
434}
435
436impl CoeffSupport {
437    #[inline]
438    pub(crate) fn without_primary(self) -> Self {
439        Self {
440            include_primary: false,
441            ..self
442        }
443    }
444}
445
446pub(crate) struct SparsePrimaryCoeffJetView<'a> {
447    primary_index: usize,
448    h_range: Option<Range<usize>>,
449    w_range: Option<Range<usize>>,
450    pub(crate) first: &'a [[f64; 4]],
451    pub(crate) a_first: &'a [[f64; 4]],
452    pub(crate) b_first: &'a [[f64; 4]],
453    pub(crate) aa_first: &'a [[f64; 4]],
454    pub(crate) ab_first: &'a [[f64; 4]],
455    pub(crate) bb_first: &'a [[f64; 4]],
456    pub(crate) aaa_first: &'a [[f64; 4]],
457    pub(crate) aab_first: &'a [[f64; 4]],
458    pub(crate) abb_first: &'a [[f64; 4]],
459    pub(crate) bbb_first: &'a [[f64; 4]],
460}
461
462impl<'a> SparsePrimaryCoeffJetView<'a> {
463    pub(crate) fn new(
464        primary_index: usize,
465        h_range: Option<&Range<usize>>,
466        w_range: Option<&Range<usize>>,
467        first: &'a [[f64; 4]],
468        a_first: &'a [[f64; 4]],
469        b_first: &'a [[f64; 4]],
470        aa_first: &'a [[f64; 4]],
471        ab_first: &'a [[f64; 4]],
472        bb_first: &'a [[f64; 4]],
473        aaa_first: &'a [[f64; 4]],
474        aab_first: &'a [[f64; 4]],
475        abb_first: &'a [[f64; 4]],
476        bbb_first: &'a [[f64; 4]],
477    ) -> Self {
478        Self {
479            primary_index,
480            h_range: h_range.cloned(),
481            w_range: w_range.cloned(),
482            first,
483            a_first,
484            b_first,
485            aa_first,
486            ab_first,
487            bb_first,
488            aaa_first,
489            aab_first,
490            abb_first,
491            bbb_first,
492        }
493    }
494
495    #[inline]
496    fn in_h_range(&self, idx: usize) -> bool {
497        self.h_range
498            .as_ref()
499            .map(|range| range.contains(&idx))
500            .unwrap_or(false)
501    }
502
503    #[inline]
504    fn in_w_range(&self, idx: usize) -> bool {
505        self.w_range
506            .as_ref()
507            .map(|range| range.contains(&idx))
508            .unwrap_or(false)
509    }
510
511    #[inline]
512    fn param_supported(&self, idx: usize, support: CoeffSupport) -> bool {
513        (support.include_primary && idx == self.primary_index)
514            || (support.include_h && self.in_h_range(idx))
515            || (support.include_w && self.in_w_range(idx))
516    }
517
518    pub(crate) fn directional_family(
519        &self,
520        family: &[[f64; 4]],
521        dir: &Array1<f64>,
522        support: CoeffSupport,
523    ) -> [f64; 4] {
524        let mut out = [0.0; 4];
525        if support.include_primary {
526            add_scaled_coeff4(
527                &mut out,
528                &family[self.primary_index],
529                dir[self.primary_index],
530            );
531        }
532        if support.include_h
533            && let Some(h_range) = self.h_range.as_ref()
534        {
535            for idx in h_range.clone() {
536                add_scaled_coeff4(&mut out, &family[idx], dir[idx]);
537            }
538        }
539        if support.include_w
540            && let Some(w_range) = self.w_range.as_ref()
541        {
542            for idx in w_range.clone() {
543                add_scaled_coeff4(&mut out, &family[idx], dir[idx]);
544            }
545        }
546        out
547    }
548
549    pub(crate) fn add_directional_family_adjoint(
550        &self,
551        family: &[[f64; 4]],
552        coeff_adjoint: &[f64; 4],
553        support: CoeffSupport,
554        direction_adjoint: &mut [f64],
555    ) {
556        assert!(direction_adjoint.len() > self.primary_index);
557        if support.include_primary {
558            direction_adjoint[self.primary_index] +=
559                coeff4_dot(coeff_adjoint, &family[self.primary_index]);
560        }
561        if support.include_h
562            && let Some(h_range) = self.h_range.as_ref()
563        {
564            for idx in h_range.clone() {
565                direction_adjoint[idx] += coeff4_dot(coeff_adjoint, &family[idx]);
566            }
567        }
568        if support.include_w
569            && let Some(w_range) = self.w_range.as_ref()
570        {
571            for idx in w_range.clone() {
572                direction_adjoint[idx] += coeff4_dot(coeff_adjoint, &family[idx]);
573            }
574        }
575    }
576
577    pub(crate) fn mixed_directional_from_b_family(
578        &self,
579        family: &[[f64; 4]],
580        dir_u: &Array1<f64>,
581        dir_v: &Array1<f64>,
582        support: CoeffSupport,
583    ) -> [f64; 4] {
584        let mut out = [0.0; 4];
585        let dir_u_primary = dir_u[self.primary_index];
586        let dir_v_primary = dir_v[self.primary_index];
587        if support.include_primary {
588            add_scaled_coeff4(
589                &mut out,
590                &family[self.primary_index],
591                dir_u_primary * dir_v_primary,
592            );
593        }
594        if support.include_h
595            && let Some(h_range) = self.h_range.as_ref()
596        {
597            for idx in h_range.clone() {
598                add_scaled_coeff4(
599                    &mut out,
600                    &family[idx],
601                    dir_u_primary * dir_v[idx] + dir_v_primary * dir_u[idx],
602                );
603            }
604        }
605        if support.include_w
606            && let Some(w_range) = self.w_range.as_ref()
607        {
608            for idx in w_range.clone() {
609                add_scaled_coeff4(
610                    &mut out,
611                    &family[idx],
612                    dir_u_primary * dir_v[idx] + dir_v_primary * dir_u[idx],
613                );
614            }
615        }
616        out
617    }
618
619    pub(crate) fn param_directional_from_b_family(
620        &self,
621        family: &[[f64; 4]],
622        param: usize,
623        dir: &Array1<f64>,
624        support: CoeffSupport,
625    ) -> [f64; 4] {
626        if param == self.primary_index {
627            return self.directional_family(family, dir, support);
628        }
629        if self.param_supported(param, support.without_primary()) {
630            let mut out = [0.0; 4];
631            add_scaled_coeff4(&mut out, &family[param], dir[self.primary_index]);
632            return out;
633        }
634        [0.0; 4]
635    }
636
637    pub(crate) fn add_param_directional_from_b_family_adjoint(
638        &self,
639        family: &[[f64; 4]],
640        param: usize,
641        coeff_adjoint: &[f64; 4],
642        support: CoeffSupport,
643        direction_adjoint: &mut [f64],
644    ) {
645        assert!(direction_adjoint.len() > self.primary_index);
646        if param == self.primary_index {
647            self.add_directional_family_adjoint(family, coeff_adjoint, support, direction_adjoint);
648        } else if self.param_supported(param, support.without_primary()) {
649            direction_adjoint[self.primary_index] += coeff4_dot(coeff_adjoint, &family[param]);
650        }
651    }
652
653    pub(crate) fn param_mixed_from_bb_family(
654        &self,
655        family: &[[f64; 4]],
656        param: usize,
657        dir_u: &Array1<f64>,
658        dir_v: &Array1<f64>,
659        support: CoeffSupport,
660    ) -> [f64; 4] {
661        if param == self.primary_index {
662            return self.mixed_directional_from_b_family(family, dir_u, dir_v, support);
663        }
664        if self.param_supported(param, support.without_primary()) {
665            let mut out = [0.0; 4];
666            add_scaled_coeff4(
667                &mut out,
668                &family[param],
669                dir_u[self.primary_index] * dir_v[self.primary_index],
670            );
671            return out;
672        }
673        [0.0; 4]
674    }
675
676    pub(crate) fn pair_from_b_family(
677        &self,
678        family: &[[f64; 4]],
679        u: usize,
680        v: usize,
681        support: CoeffSupport,
682    ) -> [f64; 4] {
683        if u == self.primary_index && v == self.primary_index {
684            if support.include_primary {
685                return family[self.primary_index];
686            }
687            return [0.0; 4];
688        }
689        if u == self.primary_index && self.param_supported(v, support.without_primary()) {
690            return family[v];
691        }
692        if v == self.primary_index && self.param_supported(u, support.without_primary()) {
693            return family[u];
694        }
695        [0.0; 4]
696    }
697
698    pub(crate) fn pair_directional_from_bb_family(
699        &self,
700        family: &[[f64; 4]],
701        u: usize,
702        v: usize,
703        dir: &Array1<f64>,
704        support: CoeffSupport,
705    ) -> [f64; 4] {
706        if u == self.primary_index && v == self.primary_index {
707            return self.directional_family(family, dir, support);
708        }
709        if u == self.primary_index && self.param_supported(v, support.without_primary()) {
710            let mut out = [0.0; 4];
711            add_scaled_coeff4(&mut out, &family[v], dir[self.primary_index]);
712            return out;
713        }
714        if v == self.primary_index && self.param_supported(u, support.without_primary()) {
715            let mut out = [0.0; 4];
716            add_scaled_coeff4(&mut out, &family[u], dir[self.primary_index]);
717            return out;
718        }
719        [0.0; 4]
720    }
721
722    pub(crate) fn add_pair_directional_from_bb_family_adjoint(
723        &self,
724        family: &[[f64; 4]],
725        u: usize,
726        v: usize,
727        coeff_adjoint: &[f64; 4],
728        support: CoeffSupport,
729        direction_adjoint: &mut [f64],
730    ) {
731        assert!(direction_adjoint.len() > self.primary_index);
732        if u == self.primary_index && v == self.primary_index {
733            self.add_directional_family_adjoint(family, coeff_adjoint, support, direction_adjoint);
734        } else if u == self.primary_index && self.param_supported(v, support.without_primary()) {
735            direction_adjoint[self.primary_index] += coeff4_dot(coeff_adjoint, &family[v]);
736        } else if v == self.primary_index && self.param_supported(u, support.without_primary()) {
737            direction_adjoint[self.primary_index] += coeff4_dot(coeff_adjoint, &family[u]);
738        }
739    }
740
741    pub(crate) fn pair_mixed_from_bbb_family(
742        &self,
743        family: &[[f64; 4]],
744        u: usize,
745        v: usize,
746        dir_u: &Array1<f64>,
747        dir_v: &Array1<f64>,
748        support: CoeffSupport,
749    ) -> [f64; 4] {
750        if u == self.primary_index && v == self.primary_index {
751            return self.mixed_directional_from_b_family(family, dir_u, dir_v, support);
752        }
753        if u == self.primary_index && self.param_supported(v, support.without_primary()) {
754            let mut out = [0.0; 4];
755            add_scaled_coeff4(
756                &mut out,
757                &family[v],
758                dir_u[self.primary_index] * dir_v[self.primary_index],
759            );
760            return out;
761        }
762        if v == self.primary_index && self.param_supported(u, support.without_primary()) {
763            let mut out = [0.0; 4];
764            add_scaled_coeff4(
765                &mut out,
766                &family[u],
767                dir_u[self.primary_index] * dir_v[self.primary_index],
768            );
769            return out;
770        }
771        [0.0; 4]
772    }
773}
774
775// ---------------------------------------------------------------------------
776// Outer-only stratified row subsample (Phase 1 scaffolding).
777//
778// The large-scale outer-loop score/gradient passes do O(n) work per outer
779// evaluation, which dominates wall-clock once n grows past ~10^5. To keep
780// outer-loop iterations tractable while leaving the inner PIRLS solve and the
781// final covariance assembly untouched, outer-only hot loops can be redirected
782// to iterate over a small stratified subsample with a constant rescaling
783// factor, sampled once per fit and shared via `Arc`. The subsample is
784// stratified by event/outcome × z-deciles (≤ 200 strata) so that the rescaled
785// estimator inherits the same support coverage as the full-data estimator.
786//
787// This module defines only the types and helpers; Phase 2 wires them into
788// per-row hot loops. Default state (`outer_score_subsample = None`) keeps the
789// legacy full-data behavior bit-for-bit.
790
791/// Splitmix64: deterministic single-u64 expansion. Thin wrapper over the
792/// canonical implementation in [`gam_linalg::utils::splitmix64`].
793#[inline]
794const fn splitmix64(state: &mut u64) -> u64 {
795    gam_linalg::utils::splitmix64(state)
796}
797
798/// Configuration for the automatic outer-score subsampler.
799///
800/// At large scale (n ≥ tens of thousands) the marginal-slope outer
801/// rho-gradient computes a sum-over-rows trace
802/// `tr(F Fᵀ M_k) = Σ_i row_i(k)` whose per-row work is dominated by
803/// the cell-moment kernel. Stratified Horvitz–Thompson subsampling
804/// replaces the full sum with an unbiased estimator using `K` of `N`
805/// rows; the trace cost drops from `O(N · cell_work)` to
806/// `O(K · cell_work)`.
807///
808/// # Math
809///
810/// Estimator `T̂ = Σ_{i∈S} w_i · row_i` with HT weights
811/// `w_i = N_h / K_h` (per-stratum) is unbiased: `E[T̂] = T`.
812///
813/// Variance under stratified SRS without replacement:
814/// `Var(T̂) = Σ_h N_h² (1 − K_h/N_h) S_h² / K_h`
815/// where `S_h²` is the within-stratum variance of per-row contributions.
816/// With proportional allocation `K_h = K · N_h/N`, the standard deviation
817/// of `T̂` relative to `T` is roughly
818/// `σ(T̂)/T ≈ (1/√K) · √(1 − K/N) · cv_within`
819/// where `cv_within` is the within-stratum coefficient of variation.
820///
821/// The defaults are tuned so that the relative gradient-noise σ stays
822/// below ≈ 1 % across realistic `n` ∈ [30 000, 300 000+], assuming
823/// `cv_within ≲ 1` (which holds for marginal-slope contributions
824/// because the z-decile stratification absorbs the dominant
825/// inhomogeneity).
826#[derive(Clone, Debug)]
827pub struct AutoOuterSubsampleOptions {
828    /// Below this `n`, the auto-subsampler always returns `None` (use
829    /// full data). Default 30 000.
830    pub min_n_for_auto: usize,
831    /// Floor on `K`, so the relative gradient noise stays bounded
832    /// even when the target fraction would round to a smaller `K`.
833    /// `K = max(min_k, round(n · target_fraction))`. Default 10 000
834    /// gives `σ/T ≤ 1 %` for cv_within ≤ 1 and any `n ≥ min_n_for_auto`.
835    pub min_k: usize,
836    /// Target ratio `K / n` once `n ≫ min_k`. Default 0.10.
837    pub target_fraction: f64,
838    /// RNG seed for stratified mask construction. Default
839    /// `0xA075_8AMP_LE_5UB5` (deterministic across runs at the same
840    /// `n`, so CRN holds across BFGS iterations).
841    pub seed: u64,
842    /// Family-supplied **per-unit-of-K** outer-derivative work cost.
843    ///
844    /// Despite the historical name, this is *not* a per-row quantity.
845    /// It is `predicted_outer_gradient_work / K` evaluated at the
846    /// family's reference operating point — i.e. how many work units
847    /// each additional row in the K-subsample contributes summed over
848    /// all n. The auto schedule caps `K` by
849    /// `K_work = AUTO_OUTER_WORK_BUDGET / outer_work_per_k_unit`,
850    /// guaranteeing a single outer evaluation never exceeds
851    /// [`AUTO_OUTER_WORK_BUDGET`] work units regardless of the
852    /// noise-only target. Default `1` (no effective work cap beyond
853    /// `K ≤ n`); families with measurable per-K cost (survival
854    /// marginal-slope, BMS) overwrite at the call site.
855    ///
856    /// Calibration recipe: from a profiled run,
857    ///     outer_work_per_k_unit = predicted_gradient_work / K.
858    /// For the large-scale survival marginal-slope reference
859    /// (predicted_gradient_work ≈ 4.33×10⁹ at K=19_661), this gives
860    /// ~220_000; we use 250_000 as a conservative upper bound. With
861    /// `AUTO_OUTER_WORK_BUDGET = 5×10⁸` that caps K at ~2_000.
862    pub outer_work_per_k_unit: u64,
863    /// Absolute floor on the chosen K after the noise/work caps are combined.
864    /// Default [`AUTO_OUTER_MIN_K_FLOOR`].
865    pub min_k_floor: usize,
866}
867
868/// Half-billion outer-derivative work units per evaluation. Picked so the
869/// rigid survival marginal-slope pilot Newton cycle (which previously ran
870/// ~57 min at n≈2e5 with `K=19_661`) finishes in a minute or two on
871/// commodity hardware once `K` is capped by this budget.
872pub const AUTO_OUTER_WORK_BUDGET: u64 = 500_000_000;
873
874/// Absolute floor on `K` chosen by the auto schedule. Even when the work
875/// budget would drive `K` to a handful of rows the stratified mask cannot
876/// usefully shrink below `MIN_K_FLOOR` without collapsing entire deciles
877/// of `z`-strata. Set so the resulting gradient noise (~3 %) is still
878/// usable for BFGS Phase 1 progress when the family is very expensive.
879pub const AUTO_OUTER_MIN_K_FLOOR: usize = 1_000;
880
881/// L2 distance below which two outer ρ keys are treated as the *same* outer
882/// step (a line-search retry, not a fresh outer iteration). Well below any
883/// meaningful BFGS step on log-scale ρ, well above float-noise from cloning
884/// the ρ vector. Used to keep the phase-1 budget counting outer iterations
885/// rather than per-step function evaluations.
886const AUTO_OUTER_DISTINCT_STEP_L2_TOL: f64 = 1e-10;
887
888/// Reason the auto schedule chose the reported `K`. Used by the
889/// `[family auto-subsample]` log line so operators can tell whether the
890/// noise model, the work budget, the `MIN_K_FLOOR`, or `n` itself
891/// determined the subsample size.
892#[derive(Clone, Copy, Debug, PartialEq, Eq)]
893pub enum AutoOuterCapReason {
894    Noise,
895    Work,
896    Floor,
897    NFull,
898}
899
900impl AutoOuterCapReason {
901    pub fn as_str(self) -> &'static str {
902        match self {
903            AutoOuterCapReason::Noise => "noise",
904            AutoOuterCapReason::Work => "work",
905            AutoOuterCapReason::Floor => "floor",
906            AutoOuterCapReason::NFull => "n",
907        }
908    }
909}
910
911impl Default for AutoOuterSubsampleOptions {
912    fn default() -> Self {
913        Self {
914            min_n_for_auto: 30_000,
915            min_k: 10_000,
916            target_fraction: 0.10,
917            seed: 0xA075_8A8B_1ED5_5B5C,
918            outer_work_per_k_unit: 1,
919            min_k_floor: AUTO_OUTER_MIN_K_FLOOR,
920        }
921    }
922}
923
924/// Outcome of [`AutoOuterSubsampleOptions::target_k_detailed`]: the
925/// chosen `K`, the underlying noise-only choice, the work-budget cap,
926/// and which constraint won.
927#[derive(Clone, Copy, Debug)]
928pub struct AutoOuterKChoice {
929    pub k: usize,
930    pub k_noise: usize,
931    pub k_work: usize,
932    pub cap_reason: AutoOuterCapReason,
933}
934
935impl AutoOuterSubsampleOptions {
936    /// Compute the K that this configuration would pick for a given n.
937    /// Returns `None` if `n < min_n_for_auto` (caller should not subsample).
938    pub fn target_k(&self, n: usize) -> Option<usize> {
939        self.target_k_detailed(n).map(|choice| choice.k)
940    }
941
942    /// Same as [`target_k`] but also reports the noise-only `K`, the
943    /// work-budget cap, and which constraint set the final value. Used by
944    /// [`maybe_install_auto_outer_subsample`] to surface a `cap_reason`
945    /// in the auto-subsample log line.
946    pub fn target_k_detailed(&self, n: usize) -> Option<AutoOuterKChoice> {
947        if n < self.min_n_for_auto {
948            return None;
949        }
950        let k_noise_raw = ((n as f64) * self.target_fraction).round() as usize;
951        let k_noise = k_noise_raw.max(self.min_k);
952        // Work-budget cap. `outer_work_per_k_unit == 1` is the
953        // default-1-work-unit signal that the family has not measured
954        // its per-K cost, in which case the work cap is `WORK_BUDGET`
955        // and typically dominated by `n`.
956        let work_per_k = self.outer_work_per_k_unit.max(1);
957        let k_work_u64 = AUTO_OUTER_WORK_BUDGET / work_per_k;
958        let k_work = usize::try_from(k_work_u64).unwrap_or(usize::MAX);
959        // Combine noise + work + n + floor in a single comparison so we
960        // can attribute the binding constraint exactly once.
961        let mut k = k_noise.min(k_work);
962        let mut cap_reason = if k_work < k_noise {
963            AutoOuterCapReason::Work
964        } else {
965            AutoOuterCapReason::Noise
966        };
967        if k < self.min_k_floor {
968            k = self.min_k_floor;
969            cap_reason = AutoOuterCapReason::Floor;
970        }
971        if k > n {
972            k = n;
973            cap_reason = AutoOuterCapReason::NFull;
974        }
975        if k >= n {
976            // Borderline: the auto schedule would cover the whole
977            // dataset. Subsampling buys nothing.
978            return None;
979        }
980        Some(AutoOuterKChoice {
981            k,
982            k_noise,
983            k_work,
984            cap_reason,
985        })
986    }
987}
988
989/// Build a stratified outer-score subsample automatically from problem
990/// characteristics. Returns `None` for problems too small to benefit
991/// (the caller should fall back to the full-data path).
992///
993/// Stratification matches `build_outer_score_subsample`: 100 z-deciles
994/// × the supplied secondary stratum (typically the {0, 1} response
995/// indicator). When `stratum_secondary` is `None` the secondary
996/// dimension collapses to a single bin.
997///
998/// The returned mask carries proper Horvitz–Thompson weights so that
999/// `Σ_{i ∈ mask} weight_i · row_i` is an unbiased estimate of the
1000/// full row sum.
1001pub fn auto_outer_score_subsample(
1002    z: &[f64],
1003    stratum_secondary: Option<&[u8]>,
1004    options: &AutoOuterSubsampleOptions,
1005) -> Option<OuterScoreSubsample> {
1006    let n = z.len();
1007    let k = options.target_k(n)?;
1008    let secondary_storage;
1009    let secondary: &[u8] = if let Some(s) = stratum_secondary {
1010        if s.len() != n {
1011            // Caller error; fall through to no-subsample rather than panic.
1012            return None;
1013        }
1014        s
1015    } else {
1016        secondary_storage = vec![0u8; n];
1017        &secondary_storage
1018    };
1019    Some(build_outer_score_subsample(z, secondary, k, options.seed))
1020}
1021
1022/// Two-phase auto-subsample guard shared across marginal-slope families.
1023///
1024/// Returns `Some(cloned_options)` carrying a freshly stratified
1025/// Horvitz-Thompson mask when `options.auto_outer_subsample` is enabled, the
1026/// caller has not already supplied a mask, and
1027/// the per-family phase counter is below `phase1_budget`. Returns `None`
1028/// when the caller's options should be used unchanged (either subsample
1029/// is disabled / pre-installed, the budget is exhausted, or the problem
1030/// is too small for `auto_outer_score_subsample` to find a benefit).
1031///
1032/// The `phase_counter` and `last_rho` pair together implement
1033/// distinct-step detection: line searches re-call the family at the
1034/// same ρ during step-size retries, but the budget is meant to count
1035/// outer iterations, not function evaluations. The counter only ticks
1036/// when the incoming ρ differs from the last observed ρ in L2 by
1037/// > 1e-10 — well below any meaningful BFGS step on log-scale ρ, well
1038/// > above float-noise from cloning. The mutex around `last_rho` is the
1039/// > minimal coordination needed: `(counter, last_rho)` must update
1040/// > together so two threads cannot both decide "new ρ" and double-bump.
1041///
1042/// The transition at `phase_idx == phase1_budget` is logged exactly
1043/// once via `log::info!` with the supplied `family_label`. Each phase-1
1044/// install also logs the planned mask size and predicted gradient
1045/// noise. Callers running with auto-subsample disabled see no logging.
1046pub fn maybe_install_auto_outer_subsample(
1047    options: &crate::custom_family::BlockwiseFitOptions,
1048    z: &[f64],
1049    stratum_secondary: Option<&[u8]>,
1050    outer_rho_key: &[f64],
1051    phase_counter: &Arc<std::sync::atomic::AtomicUsize>,
1052    last_rho: &Arc<std::sync::Mutex<Option<Array1<f64>>>>,
1053    phase1_budget: usize,
1054    family_label: &'static str,
1055    outer_work_per_k_unit: u64,
1056    min_n_for_auto: usize,
1057    min_k: usize,
1058    min_k_floor: usize,
1059) -> Option<crate::custom_family::BlockwiseFitOptions> {
1060    if options.outer_score_subsample.is_some() || !options.auto_outer_subsample {
1061        return None;
1062    }
1063    // Establish that this problem will actually use a row sample before
1064    // advancing the pilot counter.  The exact-polish lifecycle treats a zero
1065    // counter as proof that no approximate derivative measure ran; counting a
1066    // small-n no-op here would otherwise force a redundant second optimization.
1067    let auto_options = AutoOuterSubsampleOptions {
1068        min_n_for_auto,
1069        min_k,
1070        min_k_floor,
1071        outer_work_per_k_unit: outer_work_per_k_unit.max(1),
1072        ..AutoOuterSubsampleOptions::default()
1073    };
1074    let choice = auto_options.target_k_detailed(z.len())?;
1075    let phase_idx = {
1076        let mut guard = last_rho
1077            .lock()
1078            .expect("auto_subsample_last_rho mutex poisoned");
1079        let new_step = match guard.as_ref() {
1080            None => true,
1081            Some(prev) if prev.len() != outer_rho_key.len() => true,
1082            Some(prev) => {
1083                let mut sq = 0.0_f64;
1084                for (a, b) in outer_rho_key.iter().zip(prev.iter()) {
1085                    let d = a - b;
1086                    sq += d * d;
1087                }
1088                sq.sqrt() > AUTO_OUTER_DISTINCT_STEP_L2_TOL
1089            }
1090        };
1091        if new_step {
1092            *guard = Some(Array1::from(outer_rho_key.to_vec()));
1093            phase_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
1094        } else {
1095            let current = phase_counter.load(std::sync::atomic::Ordering::SeqCst);
1096            // The generic runner can promote an early-stopped pilot directly
1097            // to `phase1_budget` while remaining at the same rho checkpoint.
1098            // Preserve that exact-phase marker; subtract one only while the
1099            // counter still denotes an ordinary repeated Phase-1 evaluation.
1100            if current >= phase1_budget {
1101                current
1102            } else {
1103                current.saturating_sub(1)
1104            }
1105        }
1106    };
1107    if phase_idx >= phase1_budget {
1108        // Mark the exact phase explicitly. A raw counter equal to the budget
1109        // can also mean "the last sampled evaluation just completed"; the
1110        // post-budget sentinel lets the generic runner distinguish that state
1111        // from a full-data evaluation that has already occurred.
1112        phase_counter.fetch_max(
1113            phase1_budget.saturating_add(1),
1114            std::sync::atomic::Ordering::SeqCst,
1115        );
1116        if phase_idx == phase1_budget {
1117            log::info!(
1118                "[{family_label} auto-subsample] Phase 1 budget exhausted after {} evals; \
1119                 Phase 2 (full data) for remaining iterations",
1120                phase1_budget
1121            );
1122        }
1123        return None;
1124    }
1125    let mask = auto_outer_score_subsample(z, stratum_secondary, &auto_options)?;
1126    let n_full = mask.n_full;
1127    let k = mask.len();
1128    log::info!(
1129        "[{family_label} auto-subsample] phase=1 eval={}/{} n={} K={} fraction={:.3} expected_grad_noise={:.2}% work_per_k_unit={} k_noise={} k_work={} cap_reason={}",
1130        phase_idx + 1,
1131        phase1_budget,
1132        n_full,
1133        k,
1134        k as f64 / n_full.max(1) as f64,
1135        100.0 * (1.0 / (k as f64).sqrt()) * (1.0 - k as f64 / n_full.max(1) as f64).sqrt(),
1136        outer_work_per_k_unit,
1137        choice.k_noise,
1138        choice.k_work,
1139        choice.cap_reason.as_str(),
1140    );
1141    let mut cloned = options.clone();
1142    cloned.outer_score_subsample = Some(Arc::new(mask));
1143    Some(cloned)
1144}
1145
1146/// Build a deterministic stratified row subsample of size ≥ `k` from
1147/// `(z, stratum_secondary)`.
1148///
1149/// Stratification: 100 z-deciles × distinct values of `stratum_secondary`
1150/// (typically the {0,1} event/outcome indicator, giving ≤ 200 strata).
1151/// Each non-empty stratum contributes `ceil(k * stratum_size / n)` rows
1152/// drawn via a splitmix64-keyed Fisher-Yates partial shuffle so the result
1153/// is reproducible from `(seed, stratum_id)`.
1154///
1155/// The returned mask is sorted, deduplicated, and never empty when `n > 0`.
1156/// Per-row weights `w_i = N_h / k_h` (Horvitz-Thompson inverse-inclusion
1157/// weights for the stratum the row came from) are assigned to
1158/// `OuterScoreSubsample::rows`, and `weight_scale` is reported as the mean
1159/// of those weights for diagnostics only.
1160///
1161/// Panics if `z.len() != stratum_secondary.len()`.
1162pub fn build_outer_score_subsample(
1163    z: &[f64],
1164    stratum_secondary: &[u8],
1165    k: usize,
1166    seed: u64,
1167) -> OuterScoreSubsample {
1168    let n = z.len();
1169    assert_eq!(
1170        n,
1171        stratum_secondary.len(),
1172        "build_outer_score_subsample: z and stratum_secondary must have equal length",
1173    );
1174
1175    if n == 0 {
1176        return OuterScoreSubsample::with_uniform_weight(Vec::new(), 0, seed, 1.0);
1177    }
1178
1179    // If the requested subsample covers the full dataset (or more), short-
1180    // circuit to the full row set with weight 1.0 — this is a no-op
1181    // relative to the legacy full-data path.
1182    if k >= n {
1183        let mask: Vec<usize> = (0..n).collect();
1184        return OuterScoreSubsample::with_uniform_weight(mask, n, seed, 1.0);
1185    }
1186
1187    // Q = 100 z-deciles. Sort indices by z and split into Q ~equal chunks.
1188    const Q: usize = 100;
1189    let mut z_order: Vec<usize> = (0..n).collect();
1190    z_order.sort_by(|&a, &b| z[a].partial_cmp(&z[b]).unwrap_or(std::cmp::Ordering::Equal));
1191    // decile[i] = bin index in 0..Q for row i
1192    let mut decile = vec![0u16; n];
1193    for (rank, &row) in z_order.iter().enumerate() {
1194        // Map rank in 0..n to bin in 0..Q. Using floor((rank * Q) / n)
1195        // keeps bin sizes within ±1 row of n/Q.
1196        let bin = (rank * Q) / n;
1197        let bin = bin.min(Q - 1);
1198        decile[row] = bin as u16;
1199    }
1200
1201    // Distinct secondary values (the canonical use case is {0,1}, but the
1202    // general u8 alphabet is supported transparently).
1203    let mut distinct_secondary: Vec<u8> = stratum_secondary.to_vec();
1204    distinct_secondary.sort_unstable();
1205    distinct_secondary.dedup();
1206    // stratum index = secondary_rank * Q + decile, where secondary_rank is
1207    // the position of the row's secondary value in `distinct_secondary`.
1208    let mut secondary_rank = vec![0u16; 256];
1209    for (rank, &val) in distinct_secondary.iter().enumerate() {
1210        secondary_rank[val as usize] = rank as u16;
1211    }
1212    let n_strata = distinct_secondary.len() * Q;
1213
1214    // Bucket rows by stratum.
1215    let mut strata: Vec<Vec<usize>> = vec![Vec::new(); n_strata];
1216    for i in 0..n {
1217        let s = secondary_rank[stratum_secondary[i] as usize] as usize * Q + decile[i] as usize;
1218        strata[s].push(i);
1219    }
1220
1221    // For each non-empty stratum, draw ceil(k * stratum_size / n) rows and
1222    // tag each retained row with its HT weight w_h = N_h / k_h.
1223    let mut picked: Vec<WeightedOuterRow> = Vec::with_capacity(k + n_strata);
1224    for (stratum_id, rows) in strata.iter().enumerate() {
1225        if rows.is_empty() {
1226            continue;
1227        }
1228        let take = (k as u128 * rows.len() as u128).div_ceil(n as u128) as usize;
1229        let take = take.max(1).min(rows.len());
1230        // HT inverse-inclusion weight for this stratum: w_h = N_h / k_h.
1231        // Identical for every row drawn from `stratum_id`.
1232        let w_h = rows.len() as f64 / take as f64;
1233        let stratum_tag = stratum_id as u32;
1234
1235        // Deterministic key from (seed, stratum_id).
1236        let mut state = seed ^ (stratum_id as u64).wrapping_mul(0x9E3779B97F4A7C15);
1237        // Mix once so even seed=0, stratum_id=0 produces a non-trivial state.
1238        splitmix64(&mut state);
1239
1240        if take == rows.len() {
1241            for &index in rows.iter() {
1242                picked.push(WeightedOuterRow {
1243                    index,
1244                    weight: w_h,
1245                    stratum: stratum_tag,
1246                });
1247            }
1248        } else {
1249            // Fisher-Yates partial shuffle: produce `take` distinct rows.
1250            let mut buf: Vec<usize> = rows.clone();
1251            let m = buf.len();
1252            for i in 0..take {
1253                let r = splitmix64(&mut state);
1254                let j = i + (r as usize) % (m - i);
1255                buf.swap(i, j);
1256            }
1257            for &index in &buf[..take] {
1258                picked.push(WeightedOuterRow {
1259                    index,
1260                    weight: w_h,
1261                    stratum: stratum_tag,
1262                });
1263            }
1264        }
1265    }
1266
1267    // `from_weighted_rows` sorts + dedups by index. Strata are disjoint by
1268    // construction so dedup is a no-op, but we route through the constructor
1269    // so the OuterScoreSubsample contract stays in one place.
1270    OuterScoreSubsample::from_weighted_rows(picked, n, seed)
1271}
1272
1273// ---------------------------------------------------------------------------
1274// Outer-row iteration helpers.
1275//
1276// These wrap the choice between "iterate 0..n" (default) and "iterate
1277// `subsample.mask`" so per-row hot loops in Phase 2 can call a single helper
1278// rather than branch by hand. We expose both an enum that callers can match
1279// on directly (cheap path: a `Range` plus a `Arc<Vec<usize>>`) and a
1280// `Vec<usize>`-returning convenience that satisfies
1281// `IntoParallelIterator<Item = usize>` via `Vec`'s rayon impl.
1282
1283/// Row-index iteration choice for outer-only score/gradient passes.
1284#[derive(Debug, Clone)]
1285pub enum OuterRowIter {
1286    /// Full data: iterate `0..n`.
1287    All { n: usize },
1288    /// Subsample: iterate `subsample.mask`.
1289    Subset { mask: Arc<Vec<usize>> },
1290}
1291
1292impl OuterRowIter {
1293    /// Number of rows this iterator covers.
1294    #[inline]
1295    pub fn len(&self) -> usize {
1296        match self {
1297            OuterRowIter::All { n } => *n,
1298            OuterRowIter::Subset { mask } => mask.len(),
1299        }
1300    }
1301
1302    #[inline]
1303    pub fn is_empty(&self) -> bool {
1304        self.len() == 0
1305    }
1306
1307    /// Materialize the row indices as a `Vec<usize>`. Useful for callers
1308    /// that want a `IntoParallelIterator<Item = usize>` source — `Vec<usize>`
1309    /// satisfies that trait via rayon's blanket impl.
1310    pub fn to_vec(&self) -> Vec<usize> {
1311        match self {
1312            OuterRowIter::All { n } => (0..*n).collect(),
1313            OuterRowIter::Subset { mask } => mask.as_ref().clone(),
1314        }
1315    }
1316}
1317
1318/// Choose the row-iteration strategy for an outer-only pass. When
1319/// `opts.outer_score_subsample` is `Some`, returns the subsample mask;
1320/// otherwise returns the full range `0..n`.
1321///
1322/// Callers using this helper iterate over row indices and must additionally
1323/// consult [`outer_row_weights_by_index`] (or [`outer_weighted_rows`]) for
1324/// per-row HT weights — a single global rescale is biased under stratified
1325/// sampling and is no longer exposed.
1326pub fn outer_row_indices(
1327    opts: &crate::custom_family::BlockwiseFitOptions,
1328    n: usize,
1329) -> OuterRowIter {
1330    match opts.outer_score_subsample.as_ref() {
1331        Some(s) => OuterRowIter::Subset {
1332            mask: Arc::clone(&s.mask),
1333        },
1334        None => OuterRowIter::All { n },
1335    }
1336}
1337
1338/// Per-row HT-weighted iteration: returns one `WeightedOuterRow` per
1339/// retained row when a subsample is active; otherwise returns
1340/// `(index, weight = 1.0, stratum = 0)` for every row in `0..n`.
1341pub fn outer_weighted_rows(
1342    opts: &crate::custom_family::BlockwiseFitOptions,
1343    n: usize,
1344) -> Vec<WeightedOuterRow> {
1345    match opts.outer_score_subsample.as_ref() {
1346        Some(s) => s.rows.as_ref().clone(),
1347        None => (0..n)
1348            .map(|index| WeightedOuterRow {
1349                index,
1350                weight: 1.0,
1351                stratum: 0,
1352            })
1353            .collect(),
1354    }
1355}
1356
1357/// Dense-by-row HT weights of length `n`. Masked rows carry their HT
1358/// weight; unmasked rows default to 1.0 so that callers who index by row
1359/// regardless of subsampling still get a valid scalar (the consumer is
1360/// expected to iterate only over `outer_row_indices`).
1361pub fn outer_row_weights_by_index(
1362    opts: &crate::custom_family::BlockwiseFitOptions,
1363    n: usize,
1364) -> Vec<f64> {
1365    match opts.outer_score_subsample.as_ref() {
1366        Some(s) => {
1367            let mut weights = vec![1.0; n];
1368            for r in s.rows.iter() {
1369                if r.index < n {
1370                    weights[r.index] = r.weight;
1371                }
1372            }
1373            weights
1374        }
1375        None => vec![1.0; n],
1376    }
1377}
1378
1379/// Shared monotonicity line-search safeguard for time-block linear inequality
1380/// constraints `A·beta >= b`.
1381///
1382/// Both survival families (location-scale and marginal-slope) clamp a Newton
1383/// step `beta + alpha·delta` to the largest feasible fraction `alpha ∈ [0, 1]`
1384/// such that no constraint row is driven below its bound, then back off by the
1385/// fixed `0.995` safeguard whenever the boundary is reached. The slack/drift
1386/// arithmetic and the `0.995` factor live here once; each family supplies only
1387/// its own error type by mapping the dimension-mismatch and constraint-violation
1388/// conditions into `E` via the two closures (preserving family-specific message
1389/// text and error variants).
1390///
1391/// `map_dim_err` is called with `(beta_len, delta_len, expected_ncols)` when the
1392/// step dimensions disagree with the constraint matrix. `map_violation_err` is
1393/// called with `(row, slack)` when the current `beta` already violates a
1394/// constraint row (slack below `-1e-10`).
1395pub fn feasible_step_fraction<E>(
1396    constraints: &gam_problem::LinearInequalityConstraints,
1397    beta: &Array1<f64>,
1398    direction: &Array1<f64>,
1399    map_dim_err: impl Fn(usize, usize, usize) -> E,
1400    map_violation_err: impl Fn(usize, f64) -> E,
1401) -> Result<f64, E> {
1402    if beta.len() != constraints.a.ncols() || direction.len() != constraints.a.ncols() {
1403        return Err(map_dim_err(
1404            beta.len(),
1405            direction.len(),
1406            constraints.a.ncols(),
1407        ));
1408    }
1409    // Feasibility-violation tolerance for the *current* iterate, kept consistent
1410    // with the QP entry gate `check_linear_feasibility` (called at 1e-8) and the
1411    // residual left by `project_onto_linear_constraints` (per-row violation <= 1e-10
1412    // on the working vector, accumulating up to O(1e-9) on the final beta through
1413    // its sequential Dykstra corrections). Rejecting at -1e-10 here re-classified a
1414    // beta the QP had already accepted as feasible as a hard error (gam#797: the
1415    // projected survival time-block seed lands at slack ~ -1.1e-9 on a binding
1416    // derivative-guard row, so every trust-region attempt errored out before any
1417    // step). A slack within this band is numerically AT the boundary; treat it as
1418    // active (slack = 0) rather than a violation.
1419    const FEASIBLE_STEP_VIOLATION_TOL: f64 = 1e-8;
1420    // Multiplicative backoff applied when the step is clipped by a binding
1421    // constraint, keeping the new iterate strictly interior (slack > 0) so the
1422    // next iteration's feasibility gate cannot reject a point that landed
1423    // exactly on the boundary through round-off.
1424    const FEASIBLE_STEP_BOUNDARY_BACKOFF: f64 = 0.995;
1425    let mut alpha = 1.0f64;
1426    for row in 0..constraints.a.nrows() {
1427        let a_row = constraints.a.row(row);
1428        let raw_slack = a_row.dot(beta) - constraints.b[row];
1429        if raw_slack < -FEASIBLE_STEP_VIOLATION_TOL {
1430            return Err(map_violation_err(row, raw_slack));
1431        }
1432        // Clamp boundary round-off to the boundary so a tiny negative slack cannot
1433        // produce a spurious negative/zero step fraction below.
1434        let slack = raw_slack.max(0.0);
1435        let drift = a_row.dot(direction);
1436        if drift < 0.0 {
1437            alpha = alpha.min((slack / -drift).clamp(0.0, 1.0));
1438        }
1439    }
1440    if alpha >= 1.0 {
1441        Ok(1.0)
1442    } else {
1443        Ok((FEASIBLE_STEP_BOUNDARY_BACKOFF * alpha).clamp(0.0, 1.0))
1444    }
1445}
1446
1447/// Family-specific ψ-calculus hooks for the shared exact-Newton joint-ψ
1448/// workspace.
1449///
1450/// The two marginal-slope families (Bernoulli marginal-slope and survival
1451/// marginal-slope) build an [`ExactNewtonJointPsiWorkspace`] whose four methods
1452/// share a single skeleton: a σ-auxiliary (log-σ frailty) dispatch branch on
1453/// top of a family-specific non-σ row pass. The skeleton lives once in
1454/// [`MarginalSlopeExactNewtonPsiWorkspace`]; each family supplies only the
1455/// resolved per-call operations here, holding its own block states, specs,
1456/// derivative blocks, cache and outer-subsample options internally.
1457///
1458/// Implementors own all workspace state, so every hook takes only the ψ index /
1459/// pair / direction. The two genuine per-family policy differences in the
1460/// second-order σ-aux branch are encoded as
1461/// [`both_sigma_aux_second_order`](Self::both_sigma_aux_second_order) (which
1462/// pure-σ pairs are admissible) and
1463/// [`mixed_sigma_aux_second_order`](Self::mixed_sigma_aux_second_order) (how a
1464/// mixed σ / non-σ pair is handled) rather than being harmonized away.
1465pub trait MarginalSlopePsiFamily: Send + Sync {
1466    /// True when ψ index `psi_index` addresses the log-σ frailty auxiliary
1467    /// parameter rather than a spatial / spline derivative axis.
1468    fn is_sigma_aux(&self, psi_index: usize) -> bool;
1469
1470    /// First-order joint-ψ terms for the σ-auxiliary parameter.
1471    fn sigma_first_order_terms(
1472        &self,
1473    ) -> Result<Option<gam_problem::ExactNewtonJointPsiTerms>, String>;
1474
1475    /// First-order joint-ψ terms for a non-σ derivative axis `psi_index`.
1476    fn psi_first_order_terms(
1477        &self,
1478        psi_index: usize,
1479    ) -> Result<Option<gam_problem::ExactNewtonJointPsiTerms>, String>;
1480
1481    /// Batched first-order joint-ψ terms over all derivative axes (used by the
1482    /// outer score sweep). Returns `Ok(None)` when the batched fast path is
1483    /// unavailable for the current configuration so the caller falls back to
1484    /// per-axis evaluation.
1485    fn psi_first_order_terms_all(
1486        &self,
1487    ) -> Result<Option<Vec<gam_problem::ExactNewtonJointPsiTerms>>, String>;
1488
1489    /// Whether the σ-aux second-order branch should treat `(psi_i, psi_j)` as a
1490    /// pure-σ pair (dispatching to [`sigma_second_order_terms`](Self::sigma_second_order_terms)).
1491    /// Any σ-touching pair that is not pure-σ routes through
1492    /// [`mixed_sigma_aux_second_order`](Self::mixed_sigma_aux_second_order).
1493    fn both_sigma_aux_second_order(&self, psi_i: usize, psi_j: usize) -> bool;
1494
1495    /// Second-order joint-ψ terms for a pure σ / σ pair.
1496    fn sigma_second_order_terms(
1497        &self,
1498    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderTerms>, String>;
1499
1500    /// Per-family policy for a mixed σ / non-σ second-order pair: one family
1501    /// rejects it (no cross auxiliary terms available), the other returns
1502    /// `Ok(None)`.
1503    fn mixed_sigma_aux_second_order(
1504        &self,
1505    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderTerms>, String>;
1506
1507    /// Second-order joint-ψ terms for a non-σ derivative-axis pair.
1508    fn psi_second_order_terms(
1509        &self,
1510        psi_i: usize,
1511        psi_j: usize,
1512    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderTerms>, String>;
1513
1514    /// Direction-contracted second-order ψ terms over the non-σ derivative
1515    /// axes (#740). `alpha_psi` is the full ψ-block weight vector; the
1516    /// contraction is against the combined non-σ direction
1517    /// `ψ(α) = Σ_j alpha_psi[j] · ψ_j`, streaming the family's rows ONCE so the
1518    /// profiled θ-HVP operator applies one combined-direction n-pass per matvec
1519    /// instead of `K²` per-pair [`Self::psi_second_order_terms`] passes.
1520    ///
1521    /// Default `None` keeps the family on the exact per-pair path. The generic
1522    /// workspace only calls this when no σ-auxiliary axis carries weight (a σ
1523    /// term routes the whole direction back to the per-pair fallback), so an
1524    /// override only handles the pure non-σ derivative axes — the same domain
1525    /// as [`Self::psi_second_order_terms`].
1526    fn psi_second_order_terms_contracted(
1527        &self,
1528        _: &[f64],
1529    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderContracted>, String> {
1530        // Default implementation ignores this parameter.
1531        Ok(None)
1532    }
1533
1534    /// Hessian directional derivative for the σ-auxiliary parameter, returned
1535    /// as a dense matrix (the generic wraps it into
1536    /// [`DriftDerivResult::Dense`](gam_problem::DriftDerivResult::Dense)).
1537    fn sigma_hessian_directional_derivative(
1538        &self,
1539        d_beta_flat: &Array1<f64>,
1540    ) -> Result<Option<Array2<f64>>, String>;
1541
1542    /// Hessian directional derivative for a non-σ derivative axis, returned as
1543    /// a hyper-operator (the generic wraps it into
1544    /// [`DriftDerivResult::Operator`](gam_problem::DriftDerivResult::Operator)).
1545    fn psi_hessian_directional_derivative(
1546        &self,
1547        psi_index: usize,
1548        d_beta_flat: &Array1<f64>,
1549    ) -> Result<Option<Arc<dyn gam_problem::HyperOperator>>, String>;
1550}
1551
1552/// Generic exact-Newton joint-ψ workspace shared by the marginal-slope
1553/// families. Owns the σ-auxiliary dispatch skeleton and delegates every
1554/// family-specific operation to its [`MarginalSlopePsiFamily`] impl.
1555pub struct MarginalSlopeExactNewtonPsiWorkspace<F: MarginalSlopePsiFamily> {
1556    family: F,
1557}
1558
1559impl<F: MarginalSlopePsiFamily> MarginalSlopeExactNewtonPsiWorkspace<F> {
1560    pub fn new(family: F) -> Self {
1561        Self { family }
1562    }
1563}
1564
1565impl<F: MarginalSlopePsiFamily> gam_problem::ExactNewtonJointPsiWorkspace
1566    for MarginalSlopeExactNewtonPsiWorkspace<F>
1567{
1568    fn first_order_terms(
1569        &self,
1570        psi_index: usize,
1571    ) -> Result<Option<gam_problem::ExactNewtonJointPsiTerms>, String> {
1572        if self.family.is_sigma_aux(psi_index) {
1573            return self.family.sigma_first_order_terms();
1574        }
1575        self.family.psi_first_order_terms(psi_index)
1576    }
1577
1578    fn first_order_terms_all(
1579        &self,
1580    ) -> Result<Option<Vec<gam_problem::ExactNewtonJointPsiTerms>>, String> {
1581        self.family.psi_first_order_terms_all()
1582    }
1583
1584    fn second_order_terms(
1585        &self,
1586        psi_i: usize,
1587        psi_j: usize,
1588    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderTerms>, String> {
1589        if self.family.is_sigma_aux(psi_i) || self.family.is_sigma_aux(psi_j) {
1590            if self.family.both_sigma_aux_second_order(psi_i, psi_j) {
1591                return self.family.sigma_second_order_terms();
1592            }
1593            return self.family.mixed_sigma_aux_second_order();
1594        }
1595        self.family.psi_second_order_terms(psi_i, psi_j)
1596    }
1597
1598    fn second_order_terms_contracted(
1599        &self,
1600        alpha_psi: &[f64],
1601    ) -> Result<Option<gam_problem::ExactNewtonJointPsiSecondOrderContracted>, String> {
1602        // The σ-auxiliary axes do not participate in the family's combined
1603        // non-σ row stream (their second-order terms come from a separate
1604        // σ/σ and mixed-σ path with no directional row kernel). If any
1605        // σ-aux axis carries weight in this applied direction, decline the
1606        // contracted fast path entirely so the caller keeps the exact
1607        // per-pair assembly — the contracted hook is a representation/cost
1608        // choice, never an approximation, so falling back is the correct
1609        // behaviour rather than dropping the σ contribution.
1610        for (j, &weight) in alpha_psi.iter().enumerate() {
1611            if weight != 0.0 && self.family.is_sigma_aux(j) {
1612                return Ok(None);
1613            }
1614        }
1615        self.family.psi_second_order_terms_contracted(alpha_psi)
1616    }
1617
1618    fn hessian_directional_derivative(
1619        &self,
1620        psi_index: usize,
1621        d_beta_flat: &Array1<f64>,
1622    ) -> Result<Option<gam_problem::DriftDerivResult>, String> {
1623        if self.family.is_sigma_aux(psi_index) {
1624            return self
1625                .family
1626                .sigma_hessian_directional_derivative(d_beta_flat)
1627                .map(|result| result.map(gam_problem::DriftDerivResult::Dense));
1628        }
1629        self.family
1630            .psi_hessian_directional_derivative(psi_index, d_beta_flat)
1631            .map(|result| result.map(gam_problem::DriftDerivResult::Operator))
1632    }
1633}
1634
1635/// Process-stable worker-count estimate used for **chunk-boundary sizing only**.
1636///
1637/// Reproducibility contract (#1045): the boundaries of the row-reduction chunks
1638/// — and therefore the floating-point association of the per-chunk sums that
1639/// feed the marginal-slope REML optimum — must NOT depend on the size of the
1640/// rayon worker pool that happens to be executing the fit. Sizing chunks to the
1641/// live `rayon::current_num_threads()` broke that contract: installing a
1642/// narrower worker pool (exactly the #1045 perf lever — shrink the pool so the
1643/// per-fit `crossbeam_epoch` bookkeeping stops dominating small-`n` loops)
1644/// regrouped the per-chunk row sums and, through the iterative REML optimizer
1645/// near a flat optimum, steered the fit to a different `(ρ, λ)`. That is a
1646/// reproducibility defect, not a perf win.
1647///
1648/// This returns the machine parallelism captured once for the process — a fixed
1649/// deployment property, independent of how many workers a scoped
1650/// `ThreadPool::install` exposes — so shrinking (or widening) the executing
1651/// pool leaves the chunk boundaries, the reduction tree, and hence the fit
1652/// unchanged. rayon still fans the fixed chunks across whatever workers are
1653/// present, so parallelism is fully preserved. In production, where gam owns a
1654/// single global pool sized to the machine, this equals the previous
1655/// `current_num_threads()` value, so the fit's numerics are preserved.
1656pub(crate) fn reproducible_chunk_parallelism() -> usize {
1657    use std::sync::OnceLock;
1658    static CACHED: OnceLock<usize> = OnceLock::new();
1659    *CACHED.get_or_init(|| {
1660        std::thread::available_parallelism()
1661            .map(|n| n.get())
1662            .unwrap_or(1)
1663            .max(1)
1664    })
1665}
1666
1667/// Deterministic-order parallel reduction over a row-index slice.
1668///
1669/// Splits `rows` into contiguous chunks sized to saturate the rayon pool
1670/// (several chunks per worker, floored so small `n` stays coarse), processes
1671/// each chunk sequentially in parallel via `process_row`, and combines the
1672/// per-chunk accumulators in chunk-index order via `combine` on the calling
1673/// thread. The chunk count is a pure function of `(rows.len(),
1674/// reproducible_chunk_parallelism())` — the latter a process constant, NOT the
1675/// live scoped-pool worker count — so the reduction tree is fixed across calls
1676/// and across pool sizes regardless of rayon's work-stealing decisions.
1677///
1678/// `try_fold/try_reduce` over `rows.into_par_iter()` does **not** have
1679/// this property: rayon's adaptive splitter sets chunk boundaries based
1680/// on `current_num_threads()` and runtime work-stealing, so two calls
1681/// with identical inputs can return ULP-different floating-point sums
1682/// when the rayon pool has different concurrent activity. Tests that
1683/// compare two reductions and rely on bit-for-bit equality flake under
1684/// load with that pattern. This primitive is the per-family deterministic
1685/// row-reduction that the bernoulli / survival sigma-ψ paths funnel
1686/// through; their per-row contributions are the dominant non-deterministic
1687/// source in the marginal-slope outer-loop score / Hessian sums.
1688pub(crate) fn chunked_row_reduction<Item, Acc, Init, Process, Combine>(
1689    rows: &[Item],
1690    init: Init,
1691    process_row: Process,
1692    mut combine: Combine,
1693) -> Result<Acc, String>
1694where
1695    Item: Sync + Copy,
1696    Acc: Send,
1697    Init: Fn() -> Acc + Sync,
1698    Process: Fn(Item, &mut Acc) -> Result<(), String> + Sync,
1699    Combine: FnMut(&mut Acc, Acc),
1700{
1701    use rayon::iter::{IntoParallelIterator, ParallelIterator};
1702    let n = rows.len();
1703    if n == 0 {
1704        return Ok(init());
1705    }
1706    // The chunk count is sized so the heavy reduction phases actually saturate
1707    // the rayon pool: a fixed `32` left half of a 64-core box idle whenever the
1708    // pool had more than 32 workers, capping utilization at ~50% on the biobank
1709    // coord-corrections / row-stream phases. Targeting several chunks per worker
1710    // keeps load balanced across an uneven row-cost tail (work-stealing still
1711    // moves whole chunks, never partial sums) without flooding the sequential
1712    // `combine` with tiny partials. The count is a pure function of
1713    // `(rows.len(), reproducible_chunk_parallelism())`; the latter is the
1714    // process-stable machine parallelism (NOT the live scoped-pool worker
1715    // count), so chunk boundaries are invariant to the executing pool size and
1716    // the ordered `Vec` collect + sequential `combine` keep the reduction
1717    // bit-for-bit deterministic regardless of pool size or work-stealing.
1718    const CHUNKS_PER_WORKER: usize = 4;
1719    const MIN_CHUNK_COUNT: usize = 32;
1720    const MIN_ROWS_PER_CHUNK: usize = 64;
1721    let workers = reproducible_chunk_parallelism();
1722    let target_chunk_count = workers
1723        .saturating_mul(CHUNKS_PER_WORKER)
1724        .max(MIN_CHUNK_COUNT);
1725    // Never carve chunks below `MIN_ROWS_PER_CHUNK` rows: for small `n` the
1726    // scheduler/partial-accumulator overhead would dominate the row arithmetic.
1727    let chunk_count = target_chunk_count
1728        .min(n.div_ceil(MIN_ROWS_PER_CHUNK))
1729        .max(1);
1730    let chunk_size = n.div_ceil(chunk_count).max(1);
1731    let n_chunks = n.div_ceil(chunk_size);
1732    // `(0..n_chunks).into_par_iter()` is `IndexedParallelIterator`, so the
1733    // `.collect::<Vec<_>>()` below preserves chunk-index order regardless
1734    // of work-stealing. That ordered `Vec` is what makes the sequential
1735    // `combine` deterministic.
1736    let chunk_states: Vec<Acc> = (0..n_chunks)
1737        .into_par_iter()
1738        .map(|chunk_idx| -> Result<Acc, String> {
1739            let start = chunk_idx * chunk_size;
1740            let end = (start + chunk_size).min(n);
1741            let mut acc = init();
1742            for &item in &rows[start..end] {
1743                process_row(item, &mut acc)?;
1744            }
1745            Ok(acc)
1746        })
1747        .collect::<Result<Vec<Acc>, String>>()?;
1748    let mut total = init();
1749    for chunk in chunk_states {
1750        combine(&mut total, chunk);
1751    }
1752    Ok(total)
1753}
1754
1755#[cfg(test)]
1756mod tests {
1757    use super::*;
1758
1759    #[test]
1760    fn auto_outer_score_subsample_skips_small_problems() {
1761        let n = 1000;
1762        let z: Vec<f64> = (0..n).map(|i| i as f64).collect();
1763        let opts = AutoOuterSubsampleOptions::default();
1764        assert!(
1765            auto_outer_score_subsample(&z, None, &opts).is_none(),
1766            "n={n} below default min_n_for_auto=30000 should not subsample"
1767        );
1768    }
1769
1770    #[test]
1771    fn auto_outer_score_subsample_returns_target_k_above_threshold() {
1772        let n = 60_000;
1773        let z: Vec<f64> = (0..n).map(|i| (i as f64).sin()).collect();
1774        let opts = AutoOuterSubsampleOptions::default();
1775        let mask = auto_outer_score_subsample(&z, None, &opts)
1776            .expect("n=60000 should auto-subsample with default options");
1777        // Default target_fraction=0.10 and min_k=10000 → K = max(10000, 6000) = 10000.
1778        assert_eq!(mask.n_full, n);
1779        assert!(
1780            mask.len() >= 9_900 && mask.len() <= 10_200,
1781            "expected K≈10_000, got {}",
1782            mask.len()
1783        );
1784        // HT weights should reconstruct n_full in expectation: sum of
1785        // per-row weights ≈ n_full (allowing for small allocation rounding).
1786        let weight_sum: f64 = mask.rows.iter().map(|r| r.weight).sum();
1787        let rel_err = (weight_sum - n as f64).abs() / n as f64;
1788        assert!(
1789            rel_err < 0.02,
1790            "HT weight sum {weight_sum:.3} should ≈ n_full={n}, rel_err={rel_err:.4}"
1791        );
1792    }
1793
1794    #[test]
1795    fn sampled_outer_schedule_promotes_same_checkpoint_to_exact_measure_979() {
1796        let options = crate::custom_family::BlockwiseFitOptions::default();
1797        let phase_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
1798        let last_rho = Arc::new(std::sync::Mutex::new(None));
1799        let phase_budget = 12;
1800        let rho = [0.25, -0.5];
1801
1802        // A small problem never installs a sample and therefore must not ask
1803        // the generic runner for a redundant exact-polish solve.
1804        let small_z: Vec<f64> = (0..1_000).map(|i| i as f64).collect();
1805        assert!(
1806            maybe_install_auto_outer_subsample(
1807                &options,
1808                &small_z,
1809                None,
1810                &rho,
1811                &phase_counter,
1812                &last_rho,
1813                phase_budget,
1814                "test-small",
1815                1,
1816                30_000,
1817                10_000,
1818                1_000,
1819            )
1820            .is_none()
1821        );
1822        let schedule = crate::custom_family::OuterDerivativePilotSchedule::new(
1823            Arc::clone(&phase_counter),
1824            phase_budget,
1825        );
1826        assert!(!schedule.enter_exact_phase());
1827        assert_eq!(phase_counter.load(std::sync::atomic::Ordering::SeqCst), 0);
1828
1829        // Once a large problem actually installs its sampled measure, the
1830        // transition is single-shot and the SAME checkpoint rho immediately
1831        // evaluates on full data (no one-evaluation sampled leak into polish).
1832        let large_z: Vec<f64> = (0..40_000).map(|i| (i as f64).sin()).collect();
1833        assert!(
1834            maybe_install_auto_outer_subsample(
1835                &options,
1836                &large_z,
1837                None,
1838                &rho,
1839                &phase_counter,
1840                &last_rho,
1841                phase_budget,
1842                "test-large",
1843                1,
1844                30_000,
1845                10_000,
1846                1_000,
1847            )
1848            .is_some()
1849        );
1850        assert!(schedule.enter_exact_phase());
1851        assert!(!schedule.enter_exact_phase());
1852        assert_eq!(
1853            phase_counter.load(std::sync::atomic::Ordering::SeqCst),
1854            phase_budget + 1
1855        );
1856
1857        // Exact boundary regression: `counter == budget` is still the state
1858        // immediately after the last sampled point, not proof that a full-data
1859        // derivative has run. It must request one exact-polish continuation.
1860        let boundary_counter = Arc::new(std::sync::atomic::AtomicUsize::new(phase_budget));
1861        let boundary_schedule = crate::custom_family::OuterDerivativePilotSchedule::new(
1862            Arc::clone(&boundary_counter),
1863            phase_budget,
1864        );
1865        assert!(boundary_schedule.enter_exact_phase());
1866        assert_eq!(
1867            boundary_counter.load(std::sync::atomic::Ordering::SeqCst),
1868            phase_budget + 1
1869        );
1870        assert!(!boundary_schedule.enter_exact_phase());
1871        assert!(
1872            maybe_install_auto_outer_subsample(
1873                &options,
1874                &large_z,
1875                None,
1876                &rho,
1877                &phase_counter,
1878                &last_rho,
1879                phase_budget,
1880                "test-large",
1881                1,
1882                30_000,
1883                10_000,
1884                1_000,
1885            )
1886            .is_none(),
1887            "the first exact-polish evaluation at the pilot checkpoint must use full data",
1888        );
1889    }
1890
1891    #[test]
1892    fn auto_outer_score_subsample_horvitz_thompson_unbiased() {
1893        // On a synthetic per-row contribution `t_i = z_i² + 1`, verify
1894        // the HT-weighted sum over the auto-mask matches the full sum
1895        // within 3 standard deviations of the predicted estimator
1896        // variance. This guards against silent regressions in either
1897        // the stratified mask construction or the weight assignment.
1898        let n = 50_000;
1899        let z: Vec<f64> = (0..n)
1900            .map(|i| ((i as f64) / n as f64) * 2.0 - 1.0)
1901            .collect();
1902        let stratum: Vec<u8> = (0..n).map(|i| if i % 3 == 0 { 1 } else { 0 }).collect();
1903        let opts = AutoOuterSubsampleOptions {
1904            seed: 0xC0FFEE,
1905            ..AutoOuterSubsampleOptions::default()
1906        };
1907        let t: Vec<f64> = z.iter().map(|zi| zi * zi + 1.0).collect();
1908        let exact: f64 = t.iter().sum();
1909        let mask = auto_outer_score_subsample(&z, Some(&stratum), &opts)
1910            .expect("n=50000 should auto-subsample");
1911        let estimate: f64 = mask.rows.iter().map(|r| r.weight * t[r.index]).sum();
1912        // Predicted standard error: σ ≈ (1/√K) · √(1 − K/N) · cv · |T|.
1913        // For t_i ∈ [1, 2], cv ≲ 0.4. Be generous (factor 5) to keep
1914        // the test robust against PRNG-dependent allocation jitter.
1915        let k = mask.len();
1916        let predicted_se =
1917            exact * 0.4 * (1.0 / (k as f64).sqrt()) * (1.0 - k as f64 / n as f64).sqrt();
1918        let observed_err = (estimate - exact).abs();
1919        assert!(
1920            observed_err < 5.0 * predicted_se.max(1.0),
1921            "HT estimate {estimate:.3} vs exact {exact:.3}: err={observed_err:.3} exceeds 5×predicted_se={:.3}",
1922            predicted_se
1923        );
1924    }
1925
1926    #[test]
1927    fn subsample_full_n_equals_no_subsample() {
1928        // mask = (0..n) — the all-rows subsample should have weight_scale 1.0
1929        // and outer_row_indices should yield the same sorted set in both
1930        // Some(mask=full) and None modes.
1931        let n: usize = 1024;
1932        let z: Vec<f64> = (0..n).map(|i| i as f64).collect();
1933        let secondary: Vec<u8> = (0..n).map(|i| (i % 2) as u8).collect();
1934        let s = build_outer_score_subsample(&z, &secondary, n, 0xDEADBEEF);
1935        assert_eq!(s.len(), n);
1936        assert!((s.weight_scale - 1.0).abs() < 1e-12);
1937
1938        let mut full = crate::custom_family::BlockwiseFitOptions::default();
1939        let from_none = outer_row_indices(&full, n).to_vec();
1940        full.outer_score_subsample = Some(Arc::new(s));
1941        let from_some = outer_row_indices(&full, n).to_vec();
1942
1943        let mut a = from_none.clone();
1944        let mut b = from_some.clone();
1945        a.sort_unstable();
1946        b.sort_unstable();
1947        assert_eq!(a, b);
1948        assert_eq!(a, (0..n).collect::<Vec<_>>());
1949    }
1950
1951    #[test]
1952    fn stratification_covers_all_strata() {
1953        // Synthetic with 2 secondary classes × 100 z-deciles. Every
1954        // non-empty (secondary, decile) stratum must contribute ≥ 1 row.
1955        let n: usize = 20_000;
1956        let z: Vec<f64> = (0..n).map(|i| (i as f64) * 0.001).collect();
1957        let secondary: Vec<u8> = (0..n).map(|i| (i % 2) as u8).collect();
1958        let k = 2_000;
1959        let s = build_outer_score_subsample(&z, &secondary, k, 12345);
1960        assert!(s.len() >= k, "subsample size {} < k {}", s.len(), k);
1961
1962        // Recompute deciles to label rows.
1963        let mut order: Vec<usize> = (0..n).collect();
1964        order.sort_by(|&a, &b| z[a].partial_cmp(&z[b]).unwrap());
1965        let mut decile = vec![0usize; n];
1966        for (rank, &row) in order.iter().enumerate() {
1967            decile[row] = ((rank * 100) / n).min(99);
1968        }
1969        // For each (sec, dec), is there at least one row in mask?
1970        let mut covered = [false; 200];
1971        for &row in s.mask.iter() {
1972            let stratum = secondary[row] as usize * 100 + decile[row];
1973            covered[stratum] = true;
1974        }
1975        // All 200 strata are non-empty in this synthetic, so all must be
1976        // covered.
1977        for (stratum, &c) in covered.iter().enumerate() {
1978            assert!(c, "stratum {} uncovered", stratum);
1979        }
1980    }
1981
1982    #[test]
1983    fn deterministic_seed() {
1984        // Same inputs + seed must produce identical masks; different seeds
1985        // produce different masks (with overwhelming probability for these
1986        // sizes).
1987        let n: usize = 5_000;
1988        let z: Vec<f64> = (0..n).map(|i| (i as f64).sin()).collect();
1989        let secondary: Vec<u8> = (0..n).map(|i| (i % 2) as u8).collect();
1990        let k = 800;
1991        let a = build_outer_score_subsample(&z, &secondary, k, 0xABCDEF);
1992        let b = build_outer_score_subsample(&z, &secondary, k, 0xABCDEF);
1993        let c = build_outer_score_subsample(&z, &secondary, k, 0xFEDCBA);
1994        assert_eq!(a.mask.as_ref(), b.mask.as_ref());
1995        assert_ne!(a.mask.as_ref(), c.mask.as_ref());
1996    }
1997
1998    #[test]
1999    fn weight_scale_correct() {
2000        // n=10000, k=2000 → weight_scale ≈ 5.0 (allow small overshoot from
2001        // ceil(k * stratum_size / n) summed across strata).
2002        let n: usize = 10_000;
2003        let z: Vec<f64> = (0..n).map(|i| i as f64).collect();
2004        let secondary: Vec<u8> = (0..n).map(|i| (i % 2) as u8).collect();
2005        let k = 2_000;
2006        let s = build_outer_score_subsample(&z, &secondary, k, 7);
2007        assert!(s.len() >= k);
2008        // overshoot bounded by number of strata (one extra row per stratum
2009        // from the ceil); for 2 × 100 = 200 strata, overshoot ≤ 200.
2010        assert!(
2011            s.len() <= k + 200,
2012            "subsample {} much larger than expected",
2013            s.len()
2014        );
2015        let scale = s.weight_scale;
2016        // expected ≈ 5.0; allow ±10% for the ceiling overshoot.
2017        assert!(
2018            (scale - 5.0).abs() < 0.5,
2019            "weight_scale {} not near 5.0",
2020            scale
2021        );
2022    }
2023}