feat: consume measured SD rounds in profile-backed replay

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

@@ -297,7 +297,9 @@ impl ForwardPassPerfModelConfig {
                     "fpm_profile requires a literal backend_version",
                 ));
             }
-            if self.nextn != 0 || self.speculation.is_some() {
+            if (self.nextn != 0 && self.validate_fpm_execution()?.is_none())
+                || self.speculation.is_some()
+            {
                 return Err(invalid_config(
                     "FPM profiles support plain autoregressive decoder-only execution",
                 ));
@@ -466,7 +468,11 @@ impl ForwardPassPerfModelConfig {
                 ));
             }
         }
-        if self.estimation_mode == EstimationMode::FpmInterpolation && self.nextn != 0 {
+        let execution = self.validate_fpm_execution()?;
+        if self.estimation_mode == EstimationMode::FpmInterpolation
+            && self.nextn != 0
+            && execution.is_none()
+        {
             return Err(invalid_config(
                 "estimation_mode='fpm_interpolation' does not support MTP speculative decoding",
             ));
@@ -497,6 +503,42 @@ impl ForwardPassPerfModelConfig {
         }
         self.estimator_config.validate()
     }
+
+    pub(crate) fn validate_fpm_execution(
+        &self,
+    ) -> Result<Option<super::FpmExecutionBinding>, AicError> {
+        let Some(binding) = self.fpm_execution_binding()? else {
+            return Ok(None);
+        };
+        if self.estimation_mode != EstimationMode::FpmInterpolation
+            || self.estimator_config.fpm_interpolation.method
+                != super::FpmInterpolationMethod::Direct
+            || self.fallback_policy != ForwardPassFallbackPolicy::Deny
+            || self.database_mode != DatabaseMode::Silicon
+        {
+            return Err(invalid_config(
+                "speculative-round FPM requires explicit fpm_interpolation, direct method, SILICON data and fallback_policy='deny'",
+            ));
+        }
+        if self.nextn != binding.execution_profile.num_draft_tokens || self.speculation.is_some() {
+            return Err(invalid_config(
+                "speculative-round FPM nextn must match num_draft_tokens and cannot use generic speculation",
+            ));
+        }
+        if self.backend != BackendKind::Vllm
+            || self.pp != 1
+            || self.attention_dp != 1
+            || self.cp_size.is_some_and(|cp| cp != 1)
+            || self.dcp.is_some()
+            || self.moe_ep_size.is_some_and(|ep| ep != 1)
+            || self.moe_tp_size.is_some_and(|tp| tp != 1 && tp != self.tp)
+        {
+            return Err(invalid_config(
+                "speculative-round FPM supports vLLM pure TP with PP1/DP1/CP1 and no recorded DCP",
+            ));
+        }
+        Ok(Some(binding))
+    }
 }
 
 pub(super) fn invalid_config(message: impl Into<String>) -> AicError {

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

@@ -1111,7 +1111,10 @@ fn build_native_candidate(
             "FPM interpolation does not support EPLB, slots, moe_backend or moe_kernel_source overrides".into(),
         ));
     }
-    if config.estimation_mode == EstimationMode::FpmInterpolation && config.nextn != 0 {
+    if config.estimation_mode == EstimationMode::FpmInterpolation
+        && config.nextn != 0
+        && config.validate_fpm_execution()?.is_none()
+    {
         return Err(AicError::UnsupportedModel(
             "FPM interpolation does not support MTP".into(),
         ));

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

@@ -5,7 +5,9 @@
 
 use serde::{Deserialize, Serialize};
 
-use super::{ForwardPassPerfModel, ForwardPassPerfModelConfig, ForwardPassWorkerType};
+use super::{
+    ForwardPassPerfModel, ForwardPassPerfModelConfig, ForwardPassWorkerType, FpmExecutionBinding,
+};
 use crate::{AicError, BackendKind, MemoryBreakdown};
 
 const MAX_EXACT_BYTES: u64 = 1 << 53;
@@ -338,15 +340,86 @@ impl ForwardPassPerfModelConfig {
         if profile.get("model").and_then(serde_json::Value::as_str) != Some(self.model.as_str()) {
             return Err(invalid("FPM profile model identity mismatch"));
         }
+        let execution = self.validate_fpm_execution()?;
         if self.backend != BackendKind::Vllm
             || self.pp != 1
-            || self.nextn != 0
+            || (self.nextn != 0 && execution.is_none())
             || self.speculation.is_some()
         {
             return Err(invalid(
                 "FPM resources require vLLM PP1 plain autoregressive execution",
             ));
         }
+        let deployment = self
+            .fpm_deployment()?
+            .ok_or_else(|| invalid("FPM deployment requires a profile"))?;
+        let resources: FpmResourceConfig = serde_json::from_value(
+            deployment
+                .get("resources")
+                .cloned()
+                .ok_or_else(|| invalid("FPM deployment requires resources"))?,
+        )
+        .map_err(|error| invalid(format!("FPM resource profile: {error}")))?;
+        resources.validate()?;
+        if execution.is_some()
+            && (resources.cache_layout != FpmCacheLayout::Linear
+                || resources.runtime_memory.is_none())
+        {
+            return Err(invalid(
+                "speculative-round FPM resources require observed runtime memory and a linear cache",
+            ));
+        }
+        Ok(Some(resources))
+    }
+
+    pub(crate) fn fpm_execution_binding(&self) -> Result<Option<FpmExecutionBinding>, AicError> {
+        let Some(profile) = &self.fpm_profile else {
+            return Ok(None);
+        };
+        // Do not change validation order for historical profiles without the
+        // optional execution identity.
+        let has_execution = profile
+            .get("deployments")
+            .and_then(serde_json::Value::as_array)
+            .is_some_and(|deployments| {
+                deployments.iter().any(|deployment| {
+                    deployment
+                        .get("execution_profile")
+                        .is_some_and(|value| !value.is_null())
+                })
+            });
+        if !has_execution {
+            return Ok(None);
+        }
+        let deployment = self
+            .fpm_deployment()?
+            .ok_or_else(|| invalid("FPM execution identity requires a deployment"))?;
+        let Some(execution) = deployment
+            .get("execution_profile")
+            .filter(|value| !value.is_null())
+        else {
+            return Ok(None);
+        };
+        let binding = FpmExecutionBinding {
+            model_revision: profile
+                .get("model_revision")
+                .and_then(serde_json::Value::as_str)
+                .ok_or_else(|| invalid("FPM execution identity requires model_revision"))?
+                .to_owned(),
+            execution_profile: serde_json::from_value(execution.clone())
+                .map_err(|error| invalid(format!("FPM execution identity: {error}")))?,
+        };
+        binding.validate()?;
+        Ok(Some(binding))
+    }
+
+    fn fpm_deployment(&self) -> Result<Option<&serde_json::Value>, AicError> {
+        let Some(profile) = &self.fpm_profile else {
+            return Ok(None);
+        };
+        if profile.get("model").and_then(serde_json::Value::as_str) != Some(self.model.as_str()) {
+            return Err(invalid("FPM profile model identity mismatch"));
+        }
         let deployments = profile
             .get("deployments")
             .and_then(serde_json::Value::as_array)
@@ -406,23 +479,7 @@ impl ForwardPassPerfModelConfig {
         if matching.next().is_some() {
             return Err(invalid("duplicate FPM deployment resources"));
         }
-        if deployment
-            .get("execution_profile")
-            .is_some_and(|execution| !execution.is_null())
-        {
-            return Err(invalid(
-                "speculative-round FPM profiles are identity-only; execution is not enabled",
-            ));
-        }
-        let resources: FpmResourceConfig = serde_json::from_value(
-            deployment
-                .get("resources")
-                .cloned()
-                .ok_or_else(|| invalid("FPM deployment requires resources"))?,
-        )
-        .map_err(|error| invalid(format!("FPM resource profile: {error}")))?;
-        resources.validate()?;
-        Ok(Some(resources))
+        Ok(Some(deployment))
     }
 }
 
@@ -673,6 +730,87 @@ mod tests {
         config
     }
 
+    fn round_config() -> ForwardPassPerfModelConfig {
+        let mut config = role_config();
+        config.nextn = 3;
+        config.estimation_mode = super::super::EstimationMode::FpmInterpolation;
+        config.estimator_config.fpm_interpolation.method =
+            super::super::FpmInterpolationMethod::Direct;
+        let profile = config.fpm_profile.as_mut().unwrap();
+        profile.insert("model_revision".into(), "1".repeat(40).into());
+        for deployment in profile["deployments"].as_array_mut().unwrap() {
+            deployment["execution_profile"] = serde_json::json!({
+                "kind": "speculative_round", "method": "eagle3", "num_draft_tokens": 3,
+                "draft_model": "test/draft", "draft_revision": "2".repeat(40),
+                "timing_scope": "model_step_cuda"
+            });
+            let resources = deployment["resources"].as_object_mut().unwrap();
+            resources.insert("cache_layout".into(), "linear".into());
+            resources.insert("kv_bytes_per_token".into(), 10.into());
+            resources.remove("cache_groups");
+        }
+        config
+    }
+
+    #[test]
+    fn speculative_round_keeps_observed_capacity_and_identity_after_reload() {
+        let config = round_config();
+        config.validate().unwrap();
+        let reloaded: ForwardPassPerfModelConfig =
+            serde_json::from_str(&serde_json::to_string(&config).unwrap()).unwrap();
+        assert_eq!(reloaded.nextn, 3);
+        assert_eq!(
+            config.fpm_execution_binding().unwrap(),
+            reloaded.fpm_execution_binding().unwrap()
+        );
+        let resources = reloaded.fpm_resources().unwrap().unwrap();
+        let mut query = request();
+        query.cuda_graph_reserved_bytes = 0;
+        query.tolerance_fraction = None;
+        let budget = resources.estimate_budget(&query, 4096).unwrap();
+        // The recorded usable pool is 600 bytes at 10 bytes/token. SD has
+        // already affected that observation; neither draft depth nor TP is a
+        // second capacity multiplier or memory deduction.
+        assert_eq!(budget.total_kv_size_bytes, 600);
+        assert_eq!(budget.total_kv_size_tokens, Some(60));
+        assert_eq!(budget.memory_breakdown, None);
+        query.cuda_graph_reserved_bytes = 20;
+        assert!(resources.estimate_budget(&query, 4096).is_err());
+    }
+
+    #[test]
+    fn speculative_round_requires_matching_depth_direct_selection_and_observed_memory() {
+        let config = round_config();
+        for depth in [0, 2, 4] {
+            let mut invalid = config.clone();
+            invalid.nextn = depth;
+            assert!(invalid.validate().is_err());
+            assert!(invalid.fpm_resources().is_err());
+        }
+        let mut invalid = config.clone();
+        invalid.estimation_mode = super::super::EstimationMode::Auto;
+        assert!(invalid.validate().is_err());
+        invalid = config.clone();
+        invalid.estimator_config.fpm_interpolation.method =
+            super::super::FpmInterpolationMethod::Sol;
+        assert!(invalid.validate().is_err());
+        invalid = config.clone();
+        invalid.fallback_policy = super::super::ForwardPassFallbackPolicy::Allow;
+        assert!(invalid.validate().is_err());
+        for replacement in [resources(), runtime_resources()] {
+            let mut invalid = config.clone();
+            invalid.fpm_profile.as_mut().unwrap()["deployments"][0]["resources"] =
+                serde_json::to_value(replacement).unwrap();
+            assert!(invalid.fpm_resources().is_err());
+        }
+        let mut invalid = config;
+        let resources = invalid.fpm_profile.as_mut().unwrap()["deployments"][0]["resources"]
+            .as_object_mut()
+            .unwrap();
+        resources.remove("runtime_memory");
+        assert!(invalid.fpm_resources().is_err());
+    }
+
     #[test]
     fn profile_resources_select_independent_roles_after_round_trip() {
         let config = role_config();

crates/core/src/python.rs

@@ -1038,8 +1038,12 @@ fn aic_capacity_kwargs<'py>(
         "cuda_graph_reserved_bytes",
         config.cuda_graph_reserved_bytes,
     )?;
-    // Capacity intentionally omits NextN until AIC's Eagle memory model no
-    // longer returns negative KV capacity. Timing compilation still uses it.
+    // Legacy analytical capacity still omits NextN. Profile resources have
+    // already been validated by materialize_aic_capacity; nonzero depth here
+    // identifies measured SD memory, whose recorded pool must not be re-priced.
+    if config.fpm_profile.is_some() && config.nextn > 0 {
+        kwargs.set_item("nextn", config.nextn)?;
+    }
     kwargs.set_item("systems_path", config.systems_path.as_deref())?;
     Ok(kwargs)
 }
@@ -4099,6 +4103,40 @@ mod tests {
         }
     }
 
+    #[test]
+    fn measured_profile_capacity_forwards_depth_without_changing_analytical_defaults() {
+        pyo3::prepare_freethreaded_python();
+        let role = aggregated_role(&ReplayEngineConfig::default());
+        let mut config = aic_config();
+        Python::with_gil(|py| {
+            for depth in [0, 3] {
+                config.nextn = depth;
+                let kwargs = aic_capacity_kwargs(py, &config, &role).unwrap();
+                assert!(!kwargs.contains("nextn").unwrap());
+            }
+            // Selection/resource validity is enforced by materialize_aic_capacity.
+            // This unit test isolates only the previously missing kwargs hop.
+            config.fpm_profile = Some(serde_json::Map::new());
+            let kwargs = aic_capacity_kwargs(py, &config, &role).unwrap();
+            assert_eq!(
+                kwargs
+                    .get_item("nextn")
+                    .unwrap()
+                    .unwrap()
+                    .extract::<u32>()
+                    .unwrap(),
+                3
+            );
+            config.nextn = 0;
+            assert!(
+                !aic_capacity_kwargs(py, &config, &role)
+                    .unwrap()
+                    .contains("nextn")
+                    .unwrap()
+            );
+        });
+    }
+
     #[test]
     fn grouped_load_and_timing_domain_preserve_byte_capacity() {
         let mut role = aggregated_role(&ReplayEngineConfig::default());

docs/perf-model/methods/whole-forward.md

@@ -34,9 +34,9 @@ execution_profile:
 The initial identity contract permits vLLM pure TP with DP1/PP1/CP1 and 1–5
 draft tokens. Acceptance length belongs to replay progress, not this timing
 selector. Omitting `execution_profile` preserves existing AR profile JSON and
-cell keys. Identity publication alone does not enable execution: canonical
-construction rejects these profiles with either `nextn: 0` or nonzero `nextn`
-until the corresponding consumer is supported.
+cell keys. Consumption requires explicit `estimation_mode: fpm_interpolation`,
+`method: direct`, `database_mode: SILICON`, `fallback_policy: deny`, and `nextn`
+equal to `num_draft_tokens`. Ordinary AR profiles cannot supply SD-round timing.
 
 Rust produces the exact table `execution_profile` string by prefixing
 `speculative_round_v1:` to its canonical serialization of
@@ -71,6 +71,39 @@ not a valid conversion. Cached prefill and decode must retain the existing
 `real_kv` provenance requirement. Missing or fabricated state cannot be healed
 into these measured-only cells.
 
+### Consuming a measured round
+
+Use the same canonical `ForwardPassPerfModel.best_available(config)` path as
+ordinary FPM profiles. Both prefill and decode cells must carry the identical
+execution binding. The decode lookup uses request count and actual total past
+KV tokens, without multiplying the request count by `nextn + 1`. The measured
+latency already includes target verification, sampling and draft work; no draft
+operation tail is added. Prefill must likewise use the matching measured SD
+prefill cell, not a relabelled AR prefill cell.
+
+Resources must use a linear cache with observed `runtime_memory`. The recorded
+usable KV pool already reflects draft and graph allocations; capacity sizing
+does not subtract them again. Pending or analytically declared memory alone is
+insufficient. Grouped caches, generic `speculation` configurations, recorded
+DCP and non-pure-TP topologies remain unsupported on this route. Use `nextn`
+with the exact recorded `execution_profile`; generic speculation cost graphs
+must not be composed with measurements that already include the draft work.
+
+Supported queries are pure-prefill and pure-decode `ForwardPassMetrics`, prefill
+static/detailed queries, and replay's decode query with actual total KV tokens.
+Raw mixed FPM, static multi-round generation, per-operation and ad-hoc operation
+queries fail explicitly. Ordinary replay retains its existing separate P/D
+phase composition; this is an approximation for mixed iterations, not a measured
+mixed-round model. Partial-width decode rounds still use the configured depth's
+timing. The existing replay prefill callback uses integer per-request mean
+lengths, so ragged batches can lose remainder tokens; raw FPM totals do not have
+that approximation. Enable query coverage to identify holes rather than
+extrapolating.
+
+The SD identity does not expand the guided `onboard` collector or its replay
+validation scope. Prepare and validate the measured table separately, then use
+ordinary `predict` with the [replay SD controls](../../replay/engine/speculation.md#measured-sd-round-fpm).
+
 ## Direct and SOL-assisted methods
 
 - `method: direct` needs a verified resource profile and `database_mode: SILICON`.