perf: use recursive fits with configurable rebuilds

crates/core/src/perfmodel/fpm/estimator.rs

@@ -133,13 +133,18 @@ pub enum RegressionFitKind {
 pub struct RegressionFitConfig {
     pub kind: RegressionFitKind,
     pub singular_ridge_scale: f64,
+    /// Refresh centered statistics after this many accepted insertions and
+    /// actual evictions per workload store. The default `None` disables periodic
+    /// refreshes; numerical recovery and batch-fit fallbacks remain enabled.
+    pub rebuild_interval: Option<usize>,
 }
 
 impl Default for RegressionFitConfig {
     fn default() -> Self {
         Self {
             kind: RegressionFitKind::StandardizedNnls,
             singular_ridge_scale: 1e-9,
+            rebuild_interval: None,
         }
     }
 }
@@ -244,6 +249,7 @@ impl EstimatorConfig {
     }
 
     pub(crate) fn validate(&self) -> Result<(), AicError> {
+        validate_rebuild_interval(self.fpm_regression.fit.rebuild_interval)?;
         crate::config::validate_fpm_parquet_path(
             self.fpm_interpolation.fpm_parquet_path.as_deref(),
             true,
@@ -301,3 +307,56 @@ impl EstimatorConfig {
         })
     }
 }
+
+pub(super) fn validate_rebuild_interval(interval: Option<usize>) -> Result<(), AicError> {
+    if interval == Some(0) {
+        return Err(AicError::InvalidEngineConfig(
+            "estimator_config.fpm_regression.fit.rebuild_interval must be positive or null".into(),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rebuild_interval_defaults_and_explicit_null_survive_serde() {
+        for (json, expected) in [
+            ("{}", None),
+            (r#"{"rebuild_interval":7}"#, Some(7)),
+            (r#"{"rebuild_interval":4096}"#, Some(4096)),
+            (r#"{"rebuild_interval":null}"#, None),
+        ] {
+            let fit: RegressionFitConfig = serde_json::from_str(json).unwrap();
+            assert_eq!(fit.rebuild_interval, expected);
+            let encoded = serde_json::to_value(&fit).unwrap();
+            assert_eq!(encoded["rebuild_interval"], serde_json::json!(expected));
+            let decoded: RegressionFitConfig = serde_json::from_value(encoded).unwrap();
+            assert_eq!(decoded, fit);
+        }
+    }
+
+    #[test]
+    fn rebuild_interval_is_validated_before_estimator_selection() {
+        let mut config = EstimatorConfig::default();
+        for interval in [Some(1), Some(4096), Some(usize::MAX), None] {
+            config.fpm_regression.fit.rebuild_interval = interval;
+            config.validate().unwrap();
+        }
+        config.fpm_regression.fit.rebuild_interval = Some(0);
+        assert!(config.validate().unwrap_err().to_string().contains(
+            "estimator_config.fpm_regression.fit.rebuild_interval must be positive or null"
+        ));
+    }
+
+    #[test]
+    fn legacy_conversion_uses_the_canonical_rebuild_default() {
+        let config = EstimatorConfig::from_legacy(ForwardPassPerfOptions::default()).unwrap();
+        assert_eq!(
+            config.fpm_regression.fit.rebuild_interval,
+            RegressionFitConfig::default().rebuild_interval
+        );
+    }
+}

crates/core/src/perfmodel/fpm/model.rs

@@ -128,7 +128,11 @@ struct RegressionStores {
 }
 
 impl RegressionStores {
-    fn new(worker_type: ForwardPassWorkerType, options: &ForwardPassPerfOptions) -> Self {
+    fn new(
+        worker_type: ForwardPassWorkerType,
+        options: &ForwardPassPerfOptions,
+        rebuild_interval: Option<usize>,
+    ) -> Self {
         use ForwardPassRegressionWorkloadKind::*;
         let kinds: &[ForwardPassRegressionWorkloadKind] = match worker_type {
             ForwardPassWorkerType::Prefill => &[PurePrefill],
@@ -143,7 +147,7 @@ impl RegressionStores {
         Self {
             stores: kinds
                 .iter()
-                .map(|kind| (*kind, BucketedRegression::new(options)))
+                .map(|kind| (*kind, BucketedRegression::new(options, rebuild_interval)))
                 .collect(),
         }
     }
@@ -281,10 +285,7 @@ impl ForwardPassPerfModel {
         }
     }
 
-    /// API:
-    /// `ForwardPassPerfModel::from_regression(worker_type, options) -> Result<Self, AicError>`
-    ///
-    /// Description: create a regression-only forward-pass model.
+    /// Create a regression-only forward-pass model after resolving configuration.
     ///
     /// This mode is for native-AIC-unsupported models. It returns `None` from
     /// `estimate_forward_pass_time_ms` for non-empty iterations until the
@@ -294,12 +295,14 @@ impl ForwardPassPerfModel {
     pub(crate) fn from_regression(
         worker_type: ForwardPassWorkerType,
         options: ForwardPassPerfOptions,
+        rebuild_interval: Option<usize>,
     ) -> Result<Self, AicError> {
         validate_regression_options(&options)?;
+        super::estimator::validate_rebuild_interval(rebuild_interval)?;
         Ok(Self {
             mode: ForwardPassPerfMode::Regression {
                 worker_type,
-                regression: RegressionStores::new(worker_type, &options),
+                regression: RegressionStores::new(worker_type, &options, rebuild_interval),
             },
             options,
             last_warning: None,
@@ -348,7 +351,11 @@ impl ForwardPassPerfModel {
             }
             if mode == EstimationMode::FpmRegression {
                 let options = config.estimator_config.regression_options();
-                let mut model = Self::from_regression(config.worker_type, options)?;
+                let mut model = Self::from_regression(
+                    config.worker_type,
+                    options,
+                    config.estimator_config.fpm_regression.fit.rebuild_interval,
+                )?;
                 let mut resolved = config.clone();
                 resolved.estimation_mode = mode;
                 resolved.fallback_policy = ForwardPassFallbackPolicy::Deny;
@@ -1165,6 +1172,72 @@ fn can_fallback_to_regression(err: &AicError) -> bool {
     )
 }
 
+#[cfg(test)]
+mod rebuild_tests {
+    use super::*;
+    use ForwardPassRegressionWorkloadKind::*;
+
+    #[test]
+    fn workload_stores_and_clones_have_independent_rebuild_clocks() {
+        let options = ForwardPassPerfOptions::default();
+        let mut stores =
+            RegressionStores::new(ForwardPassWorkerType::Aggregated, &options, Some(3));
+        assert!(
+            stores
+                .store_mut(PurePrefill)
+                .add_observation([1.0, 2.0], 3.0)
+        );
+        for i in 1..=2 {
+            assert!(
+                stores
+                    .store_mut(PureDecode)
+                    .add_observation([i as f64, 1.0], 4.0)
+            );
+        }
+        let mut cloned = stores.clone();
+        assert!(
+            stores
+                .store_mut(PureDecode)
+                .add_observation([3.0, 4.0], 5.0)
+        );
+        assert_eq!(stores.store(PureDecode).mutations_since_rebuild(), 0);
+        assert_eq!(stores.store(PurePrefill).mutations_since_rebuild(), 1);
+        assert_eq!(
+            stores.store(ContainsLocallyMixed).mutations_since_rebuild(),
+            0
+        );
+        assert_eq!(
+            stores.store(CrossRankAggregated).mutations_since_rebuild(),
+            0
+        );
+        assert_eq!(cloned.store(PureDecode).mutations_since_rebuild(), 2);
+        assert!(
+            cloned
+                .store_mut(PurePrefill)
+                .add_observation([2.0, 3.0], 4.0)
+        );
+        assert_eq!(cloned.store(PurePrefill).mutations_since_rebuild(), 2);
+        assert_eq!(stores.store(PurePrefill).mutations_since_rebuild(), 1);
+    }
+
+    #[test]
+    fn default_rebuild_interval_disables_periodic_refresh_for_every_workload_store() {
+        let mut stores = RegressionStores::new(
+            ForwardPassWorkerType::Aggregated,
+            &ForwardPassPerfOptions::default(),
+            super::super::estimator::RegressionFitConfig::default().rebuild_interval,
+        );
+        // Constant features avoid fit work without numerical damage. A change
+        // back to a 4096-operation default would reset each clock at 4096.
+        for (_, store) in &mut stores.stores {
+            for _ in 0..2081 {
+                assert!(store.add_observation([1.0, 2.0], 3.0));
+            }
+            assert_eq!(store.mutations_since_rebuild(), 4098);
+        }
+    }
+}
+
 #[cfg(test)]
 mod fallback_errors {
     use super::*;