feat: load DCP self-benchmark profiles

README.md

@@ -265,7 +265,7 @@ APIs:
 
 - [Estimator/FPE Python and Rust SDK](docs/core-api.md)
 - [AIC-compatible modeled-power contract (semantics only)](docs/power-model.md)
-- [FPM collection-to-prediction workflow](python/aisimulate/docs/fpm/end-to-end-workflow.md)
+- [Self-benchmarking and FPM onboarding guide](python/aisimulate/docs/fpm/self-benchmarking-and-onboarding.md)
 - [Replay SDK and artifact contract](crates/core/src/replay/README.md)
 - [Sweeper SDK](docs/sweeper/overview.md)
 - [Legacy CLI reference](docs/cli/legacy-aic-user-guide.md)

crates/core/src/lib.rs

@@ -49,7 +49,7 @@ pub use perfmodel::{
 };
 pub use perfmodel::{
     CorrectionFactorBounds, CorrectionFeatureSpace, FpmInterpolationConfig, OpLevelConfig,
-    RegressionFitConfig, RegressionFitKind,
+    RegressionFitConfig, RegressionFitKind, UnrecordedFpmQuantMode,
 };
 
 #[cfg(feature = "python")]

crates/core/src/perfmodel/config.rs

@@ -114,7 +114,12 @@ pub const ENGINE_CONFIG_SCHEMA_VERSION: u32 = 1;
 // - 22 (GLM-5.2 VR200 pilot): exact observed-MoE selection, prefill graph
 //   identity and two appended composite operators extend the schema-21 layout.
 //   The pilot and AIC-1781 concurrently claimed 21; reject both older layouts.
-pub const ENGINE_SPEC_SCHEMA_VERSION: u32 = 22;
+// - 23 (DCP identity): ParallelMapping gained optional recorded dcp_size.
+//   DCP and the pilot concurrently claimed 22; reject both older layouts.
+//   JSON defaults preserve unrecorded DCP.
+// - 24 (typed FPM DCP): FpmForwardOp carries dcp_size separately from the
+//   base matching identity, so control flow never parses the string tuple.
+pub const ENGINE_SPEC_SCHEMA_VERSION: u32 = 24;
 
 /// Static engine identity and setup information carried by an
 /// [`crate::perfmodel::engine::spec::EngineSpec`].
@@ -261,6 +266,8 @@ pub struct ParallelMapping {
     /// re-derived from this field.
     #[serde(default)]
     pub cp_size: Option<u32>,
+    #[serde(default)]
+    pub dcp_size: Option<u32>,
 }
 
 /// Precision/quantization dtypes. Flattened into [`EngineConfig`]. Field

crates/core/src/perfmodel/engine/readiness.rs

@@ -163,9 +163,11 @@ impl Availability<'_> {
             }
             TokenScale(scale) => return self.op(&scale.op),
             FpmForward(fpm) => {
-                self.db
-                    .fpm_forward
-                    .select_cell(&fpm.match_identity, &fpm.model_path)?;
+                self.db.fpm_forward.select_cell(
+                    &fpm.match_identity,
+                    &fpm.model_path,
+                    fpm.dcp_size,
+                )?;
                 return Ok(());
             }
             // These families have no analytic/empirical implementation.

crates/core/src/perfmodel/engine/runtime.rs

@@ -2485,6 +2485,7 @@ mod tests {
             moe_kernel_source: None,
             kv_block_size: None,
             parallel: ParallelMapping {
+                dcp_size: None,
                 tp_size: 8,
                 pp_size: 1,
                 attention_dp_size: Some(1),
@@ -2555,6 +2556,7 @@ mod tests {
         config.backend = BackendKind::Sglang;
         config.backend_version = Some("0.5.18+nvinternal.rubin.0.8full.66997102".into());
         config.parallel = ParallelMapping {
+            dcp_size: None,
             tp_size: 4,
             pp_size: 1,
             attention_dp_size: Some(1),
@@ -3530,6 +3532,7 @@ mod tests {
         config.systems_path = Some(tmp.to_path_buf());
         let fpm_op = |phase: FpmPhase| {
             Op::FpmForward(FpmForwardOp {
+                dcp_size: None,
                 name: format!("fpm_forward_{}", phase.as_str()),
                 phase,
                 model_path: "org/model-a".to_string(),
@@ -3601,6 +3604,7 @@ mod tests {
         use crate::perf_database::fpm_forward::tests::default_identity;
         let db = PerfDatabase::load(&systems_root(), "b200_sxm", "vllm", "0.24.0").unwrap();
         let fpm_op = Op::FpmForward(FpmForwardOp {
+            dcp_size: None,
             name: "fpm_forward_prefill".into(),
             phase: FpmPhase::Prefill,
             model_path: "org/model-a".into(),
@@ -3659,6 +3663,7 @@ mod tests {
         ));
         let fpm_op = |phase: FpmPhase| {
             Op::FpmForward(FpmForwardOp {
+                dcp_size: None,
                 name: format!("fpm_forward_{}", phase.as_str()),
                 phase,
                 model_path: "org/model-a".to_string(),
@@ -4204,6 +4209,7 @@ mod tests {
         let hidden = Op::Overlap(crate::operators::OverlapOp::new(
             "hidden",
             vec![Op::FpmForward(FpmForwardOp {
+                dcp_size: None,
                 name: "fpm_forward_prefill".into(),
                 phase: FpmPhase::Prefill,
                 model_path: "org/model-a".into(),
@@ -4522,6 +4528,7 @@ mod tests {
             "0.25.1",
         ));
         let mut op = FpmForwardOp {
+            dcp_size: None,
             name: "fpm_forward_decode".into(),
             phase: FpmPhase::Decode,
             model_path: "org/model-a".into(),
@@ -4568,6 +4575,7 @@ mod tests {
         ));
         let fpm_op = |phase: FpmPhase, width: u32| {
             Op::FpmForward(FpmForwardOp {
+                dcp_size: None,
                 name: format!("fpm_forward_{}", phase.as_str()),
                 phase,
                 model_path: "org/model-a".to_string(),

crates/core/src/perfmodel/engine/spec.rs

@@ -625,6 +625,7 @@ mod tests {
         // Recursive like Overlap/Fallback: sol_ops carries the model's
         // original granular list, so the round-trip must preserve nesting.
         crate::operators::FpmForwardOp {
+            dcp_size: None,
             name: "fpm_forward_prefill".into(),
             phase: crate::operators::FpmPhase::Prefill,
             model_path: "org/model-a".into(),
@@ -850,6 +851,7 @@ mod tests {
             moe_kernel_source: None,
             kv_block_size: Some(64),
             parallel: ParallelMapping {
+                dcp_size: None,
                 tp_size: 8,
                 pp_size: 1,
                 attention_dp_size: Some(8),
@@ -1045,7 +1047,7 @@ mod tests {
             vec![OpSpec::FpmForward(fpm_forward())],
             vec![OpSpec::Moe(moe())],
         );
-        assert_eq!(spec.schema_version, 22);
+        assert_eq!(spec.schema_version, 24);
         let mut bytes = spec.to_bincode().unwrap();
         assert_eq!(EngineSpec::from_bincode(&bytes).unwrap(), spec);
 
@@ -1350,8 +1352,11 @@ mod tests {
     fn merged_schema_preserves_fpm_selectors_and_pilot_payloads() {
         let mut fpm_config = sample_engine_config();
         fpm_config.forward_model = Some("fpm".into());
+        fpm_config.parallel.dcp_size = Some(2);
         fpm_config.quantization.fpm_fmha_dtype = Some(DataType::Fp8);
-        let fpm_spec = EngineSpec::new(fpm_config, vec![OpSpec::FpmForward(fpm_forward())], vec![]);
+        let mut fpm_op = fpm_forward();
+        fpm_op.dcp_size = Some(2);
+        let fpm_spec = EngineSpec::new(fpm_config, vec![OpSpec::FpmForward(fpm_op)], vec![]);
 
         let mut pilot_config = sample_engine_config();
         pilot_config.prefill_graph_profile =
@@ -1375,18 +1380,18 @@ mod tests {
         for spec in [fpm_spec, pilot_spec] {
             let bytes = spec.to_bincode().unwrap();
             assert_eq!(EngineSpec::from_bincode(&bytes).unwrap(), spec);
-            for previous_version in [20u32, 21] {
+            for previous_version in [20u32, 21, 22, 23] {
                 let mut stale = bytes.clone();
                 stale[..4].copy_from_slice(&previous_version.to_le_bytes());
-                // Both branches claimed 21 after distinct schema-20 layouts.
-                // Reject either version before decoding even a missing payload.
+                // DCP and the pilot claimed 22 for different layouts.
+                // Reject all older versions before decoding even a missing payload.
                 for input in [stale.as_slice(), &stale[..4]] {
                     assert!(matches!(
                         EngineSpec::from_bincode(input),
                         Err(AicError::UnsupportedSchemaVersion {
                             kind: "EngineSpec",
                             got,
-                            expected: 22,
+                            expected: 24,
                         }) if got == previous_version
                     ));
                 }
@@ -1412,9 +1417,25 @@ mod tests {
             Err(AicError::UnsupportedSchemaVersion {
                 got: 19,
                 expected: ENGINE_SPEC_SCHEMA_VERSION,
-
                 ..
             })
         ));
     }
+    #[test]
+    fn recorded_dcp_and_legacy_absence_round_trip() {
+        let mut spec = handshake_spec();
+        spec.engine.parallel.dcp_size = Some(2);
+        let decoded = EngineSpec::from_bincode(&spec.to_bincode().unwrap()).unwrap();
+        assert_eq!(decoded.engine.parallel.dcp_size, Some(2));
+        let mut legacy = serde_json::to_value(&spec.engine).unwrap();
+        legacy.as_object_mut().unwrap().remove("dcp_size");
+        let legacy: EngineConfig = serde_json::from_value(legacy).unwrap();
+        assert_eq!(legacy.parallel.dcp_size, None);
+        let mut previous = spec.to_bincode().unwrap();
+        previous[..4].copy_from_slice(&23u32.to_le_bytes());
+        assert!(matches!(
+            EngineSpec::from_bincode(&previous),
+            Err(AicError::UnsupportedSchemaVersion { got: 23, .. })
+        ));
+    }
 }

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

@@ -94,6 +94,9 @@ pub struct ForwardPassPerfModelConfig {
     pub pp: u32,
     #[serde(default = "one", alias = "attention_dp_size")]
     pub attention_dp: u32,
+    /// Recorded decode context parallelism within the TP group.
+    #[serde(default)]
+    pub dcp: Option<u32>,
     #[serde(default)]
     pub moe_tp_size: Option<u32>,
     #[serde(default)]
@@ -171,6 +174,7 @@ impl ForwardPassPerfModelConfig {
             tp: 1,
             pp: 1,
             attention_dp: 1,
+            dcp: None,
             moe_tp_size: None,
             moe_ep_size: None,
             gemm_quant_mode: None,
@@ -277,6 +281,17 @@ impl ForwardPassPerfModelConfig {
         if self.tp == 0 || self.pp == 0 || self.attention_dp == 0 {
             return Err(invalid_config("tp, pp, and attention_dp must be positive"));
         }
+        if self.dcp.is_some_and(|dcp| dcp == 0 || self.tp % dcp != 0) {
+            return Err(invalid_config("dcp must be positive and divide tp"));
+        }
+        self.estimator_config
+            .fpm_interpolation
+            .validate_quant_modes(
+                self.fmha_quant_mode
+                    .as_deref()
+                    .or(self.fpm_fmha_quant_mode.as_deref()),
+                self.comm_quant_mode.as_deref(),
+            )?;
         if self.moe_tp_size.is_some() != self.moe_ep_size.is_some() {
             return Err(invalid_config(
                 "moe_tp_size and moe_ep_size must be configured together",
@@ -430,7 +445,7 @@ impl ForwardPassPerfModelConfig {
     }
 }
 
-fn invalid_config(message: impl Into<String>) -> AicError {
+pub(super) fn invalid_config(message: impl Into<String>) -> AicError {
     AicError::InvalidEngineConfig(format!(
         "invalid forward pass perf model config: {}",
         message.into()
@@ -661,6 +676,32 @@ mod tests {
             .is_err()
         );
     }
+    #[test]
+    fn recorded_dcp_is_validated_and_survives_round_trip() {
+        let cfg =
+            config(serde_json::json!({"tp": 8, "dcp": 8, "moe_tp_size": 8, "moe_ep_size": 1}));
+        cfg.validate().unwrap();
+        let restored: ForwardPassPerfModelConfig =
+            serde_json::from_str(&serde_json::to_string(&cfg).unwrap()).unwrap();
+        assert_eq!(restored.dcp, Some(8));
+        assert_eq!(config(serde_json::json!({})).dcp, None);
+        for value in [0, 3, 16] {
+            assert!(
+                config(serde_json::json!({"tp": 8, "dcp": value}))
+                    .validate()
+                    .is_err()
+            );
+        }
+    }
+
+    #[test]
+    fn unrecorded_profile_quantization_rejects_explicit_overrides() {
+        let mut cfg = config(serde_json::json!({"estimation_mode": "fpm_interpolation",
+            "estimator_config": {"fpm_interpolation": {"text_only": true, "unrecorded_quant_modes": ["fmha", "comm"]}}}));
+        cfg.validate().unwrap();
+        cfg.fmha_quant_mode = Some("bfloat16".into());
+        assert!(cfg.validate().is_err());
+    }
 
     #[test]
     fn graph_profile_pins_identity_and_rejects_unqualified_selection() {

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

@@ -37,9 +37,47 @@ pub struct OpLevelConfig {
 #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
 #[serde(default, deny_unknown_fields)]
 pub struct FpmInterpolationConfig {
+    /// The profile covers text prefill/decode; encoder weights remain resident.
+    #[serde(skip_serializing_if = "is_false")]
+    pub text_only: bool,
     /// External parquet and its same-stem metadata sidecar.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub fpm_parquet_path: Option<PathBuf>,
+    /// Match null profile identities only for these unspecified quant modes.
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub unrecorded_quant_modes: Vec<UnrecordedFpmQuantMode>,
+}
+
+fn is_false(value: &bool) -> bool {
+    !*value
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum UnrecordedFpmQuantMode {
+    Fmha,
+    Comm,
+}
+
+impl FpmInterpolationConfig {
+    pub(crate) fn validate_quant_modes(
+        &self,
+        fmha: Option<&str>,
+        comm: Option<&str>,
+    ) -> Result<(), AicError> {
+        for mode in &self.unrecorded_quant_modes {
+            let explicit = match mode {
+                UnrecordedFpmQuantMode::Fmha => fmha,
+                UnrecordedFpmQuantMode::Comm => comm,
+            };
+            if explicit.is_some() {
+                return Err(super::config::invalid_config(
+                    "an unrecorded FPM quant mode cannot have an explicit quantization override",
+                ));
+            }
+        }
+        Ok(())
+    }
 }
 
 /// These weights currently affect regression. Native correction keeps its

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

@@ -328,6 +328,17 @@ impl ForwardPassPerfModel {
         let mut failures = Vec::new();
         let mut last_error = None;
         for mode in config.candidate_modes() {
+            if config.dcp.is_some_and(|dcp| dcp > 1)
+                && (mode != EstimationMode::FpmInterpolation
+                    || config.backend != crate::BackendKind::Vllm)
+            {
+                let error = AicError::UnsupportedModel(
+                    "DCP timing requires measured vLLM FPM interpolation".into(),
+                );
+                failures.push(format!("{mode:?}: {error}"));
+                last_error = Some(error);
+                continue;
+            }
             if config.speculation.is_some() && mode != EstimationMode::OpLevel {
                 let error =
                     AicError::UnsupportedModel("ngram speculation requires op_level timing".into());

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

@@ -33,6 +33,26 @@ fn systems_root() -> PathBuf {
 
 const TEST_MODEL: &str = "MiniMaxAI/MiniMax-M2.5";
 
+#[test]
+fn dcp_cannot_construct_explicit_regression() {
+    let mut config = crate::ForwardPassPerfModelConfig::new(
+        TEST_MODEL,
+        "b200_sxm",
+        BackendKind::Vllm,
+        ForwardPassWorkerType::Aggregated,
+    );
+    config.tp = 8;
+    config.dcp = Some(8);
+    config.estimation_mode = crate::EstimationMode::FpmRegression;
+    let error = ForwardPassPerfModel::best_available(config).unwrap_err();
+    assert!(matches!(error, AicError::UnsupportedModel(_)));
+    assert!(
+        error
+            .to_string()
+            .contains("measured vLLM FPM interpolation")
+    );
+}
+
 /// Hand-built context op list against the b200_sxm/vllm/0.24.0 perf tables
 /// (same fixture pattern as `engine/runtime.rs` and `py.rs` tests).
 fn context_ops() -> Vec<Op> {
@@ -114,6 +134,7 @@ fn fixture_engine_config() -> EngineConfig {
         moe_kernel_source: None,
         kv_block_size: None,
         parallel: ParallelMapping {
+            dcp_size: None,
             tp_size: 8,
             pp_size: 1,
             attention_dp_size: Some(1),

crates/core/src/perfmodel/memory.rs

@@ -515,6 +515,7 @@ mod tests {
                 moe_kernel_source: None,
                 kv_block_size: None,
                 parallel: ParallelMapping {
+                    dcp_size: None,
                     tp_size: 1,
                     pp_size: 1,
                     attention_dp_size: Some(1),