feat: decouple FPM workflows from analytical model classes

.github/codeowners/areas.yaml

@@ -33,6 +33,8 @@ areas:
       - python/aisimulate/
       - python/aisimulate/src/aisimulate/capacity.py
       - tests/test_aic.py
+      - tests/test_fpm_profile_workflow.py
+      - tests/test_fpm_query_evidence.py
   - label: sweeper
     github_team: "@ai-dynamo/aisimulate-sweeper-codeowners"
     path_globs:
@@ -158,6 +160,10 @@ shared:
     owners: [forward-pass-engine, maintainers]
   - glob: tests/test_aic.py
     owners: [forward-pass-engine, maintainers]
+  - glob: tests/test_fpm_profile_workflow.py
+    owners: [forward-pass-engine, maintainers]
+  - glob: tests/test_fpm_query_evidence.py
+    owners: [forward-pass-engine, replay, maintainers]
   - glob: python/aisimulate/src/aisimulate/sweeper/
     owners: [sweeper, maintainers]
   - glob: docs/sweeper/

CODEOWNERS

@@ -45,6 +45,8 @@
 /crates/tests/public-api/                                @ai-dynamo/aisimulate-forward-pass-engine-codeowners
 /crates/core/src/perfmodel/                              @ai-dynamo/aisimulate-forward-pass-engine-codeowners
 /crates/core/tests/perfmodel/                            @ai-dynamo/aisimulate-forward-pass-engine-codeowners
+/tests/test_fpm_query_evidence.py                        @ai-dynamo/aisimulate-forward-pass-engine-codeowners
+/tests/test_fpm_profile_workflow.py                      @ai-dynamo/aisimulate-forward-pass-engine-codeowners
 /crates/core/parity_tests/perfmodel/                     @ai-dynamo/aisimulate-forward-pass-engine-codeowners
 /python/aisimulate/src/aisimulate/capacity.py            @ai-dynamo/aisimulate-forward-pass-engine-codeowners
 
@@ -179,12 +181,14 @@
 /crates/core/src/engine/timing.rs                        @ai-dynamo/aisimulate-mocker-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/.gitattributes                        @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/pyproject.toml                        @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-mocker-codeowners @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
+/tests/test_fpm_query_evidence.py                        @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_resource_scheduler.py                        @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_state_cache_config.py                        @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-mocker-codeowners @ai-dynamo/access-aisimulate-maintain
 /crates/tests/public-api/Cargo.toml                      @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/.coderabbit.yaml                      @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/eslint.config.js                      @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_e2e_accuracy_nightly.py                      @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
+/tests/test_fpm_profile_workflow.py                      @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_resource_supervision.py                      @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/access-aisimulate-maintain
 /crates/core/parity_tests/perfmodel/                     @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_ci_workflow_contracts.py                     @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain

crates/core/src/engine/timing.rs

@@ -208,6 +208,11 @@ impl TimingOperationEvidence {
                         left.fallbacks.push(fallback);
                     }
                 }
+                crate::perfmodel::fpm::accumulate_fpm_estimates(
+                    &mut left.fpm_estimates,
+                    right.fpm_estimates,
+                )
+                .map_err(anyhow::Error::msg)?;
                 Some(left)
             }
             (Some(mut details), None) | (None, Some(mut details)) => {
@@ -734,6 +739,12 @@ pub trait TimingModel: Send + Sync {
         None
     }
 
+    /// Latency-only whole-model FPM evidence is separate from energy evidence.
+    /// It must not change the provider's existing power publication contract.
+    fn fpm_evidence_summary(&self) -> Option<TimingEvidenceSummary> {
+        None
+    }
+
     /// Start a new measurement epoch without changing timing predictions or
     /// provider caches. Called only after preparation work has settled.
     ///
@@ -912,6 +923,7 @@ mod tests {
                             sol: None,
                             sol_unavailable_reason: Some("unsupported".into()),
                             fallbacks: Vec::new(),
+                            fpm_estimates: Vec::new(),
                         })
                 }
             }
@@ -1021,6 +1033,7 @@ mod tests {
                 TimingOperationEvidence::new("test", 2.0, None, TimingEvidenceSource::Estimated)
                     .unwrap();
             op.details = Some(OperationDetails {
+                fpm_estimates: Vec::new(),
                 sol: has_sol.then_some(SolDiagnostics {
                     latency_ms: 1.0,
                     math_ms: 0.0,
@@ -1079,6 +1092,7 @@ mod tests {
             )
             .unwrap();
             operation.details = Some(OperationDetails {
+                fpm_estimates: Vec::new(),
                 sol: None,
                 sol_unavailable_reason: Some("unsupported operation".into()),
                 fallbacks: vec![valid.clone(), fallback],
@@ -1100,6 +1114,7 @@ mod tests {
                 .unwrap();
         let mut detailed = plain.clone();
         detailed.details = Some(OperationDetails {
+            fpm_estimates: Vec::new(),
             sol: Some(SolDiagnostics {
                 latency_ms: 1.0,
                 math_ms: 0.0,
@@ -1455,6 +1470,7 @@ mod tests {
                     }),
                     sol_unavailable_reason: None,
                     fallbacks: Vec::new(),
+                    fpm_estimates: Vec::new(),
                 });
             }
             let incoming = ValidatedTimingPhase::from_operations(vec![op]).unwrap();

crates/core/src/lib.rs

@@ -48,8 +48,8 @@ pub use perfmodel::{
     SpeculativeConfig,
 };
 pub use perfmodel::{
-    CorrectionFactorBounds, CorrectionFeatureSpace, FpmInterpolationConfig, OpLevelConfig,
-    RegressionFitConfig, RegressionFitKind, UnrecordedFpmQuantMode,
+    CorrectionFactorBounds, CorrectionFeatureSpace, FpmInterpolationConfig, FpmInterpolationMethod,
+    OpLevelConfig, RegressionFitConfig, RegressionFitKind, UnrecordedFpmQuantMode,
 };
 
 #[cfg(feature = "python")]
@@ -75,3 +75,8 @@ pub(crate) use perfmodel::{
 #[cfg(feature = "python")]
 #[allow(unused_imports)]
 pub(crate) use perfmodel::{py, py_ops};
+
+pub use perfmodel::{
+    DirectFpmQueryEvidence, ForwardPassEstimate, FpmCoordinates, FpmEstimateEvidence,
+    FpmMeasurementSupport, FpmQueryResolution, FpmRankEstimate,
+};

crates/core/src/perfmodel/config.rs

@@ -119,7 +119,10 @@ pub const ENGINE_CONFIG_SCHEMA_VERSION: u32 = 1;
 //   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;
+// - 25 (FPM decoupling): FpmForwardOp also carries the SOL/direct interpolation
+//   selector. The decoupling branch independently used 21 for this positional
+//   field; the combined layout differs from every prior schema, including 24.
+pub const ENGINE_SPEC_SCHEMA_VERSION: u32 = 25;
 
 /// Static engine identity and setup information carried by an
 /// [`crate::perfmodel::engine::spec::EngineSpec`].

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

@@ -27,6 +27,9 @@ pub struct OperationDetails {
     pub sol_unavailable_reason: Option<String>,
     /// Executed measurement substitutions; empty means no recorded substitution.
     pub fallbacks: Vec<ExecutedFallback>,
+    /// Whole-forward estimates, counted by identical result and replay scale.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub fpm_estimates: Vec<crate::FpmEstimateEvidence>,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]

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

@@ -10,6 +10,7 @@ use crate::AicError;
 use crate::common::enums::{DatabaseMode, GemmQuantMode, MoeQuantMode, TransferKind};
 use crate::config::PerfSource;
 use crate::operators::Op;
+use crate::operators::fpm_forward::{FpmForwardOp, FpmInterpolation, FpmPhase};
 use crate::perf_database::moe::MoeKernel;
 use crate::perf_database::{PerfDatabase, kernel_source_ok, parquet_loader::PerfReader};
 
@@ -27,6 +28,28 @@ pub(super) fn validate<'a>(
     Ok(())
 }
 
+pub(super) fn validate_fpm(db: &PerfDatabase, fpm: &FpmForwardOp) -> Result<(), AicError> {
+    let cell = db
+        .fpm_forward
+        .select_cell(&fpm.match_identity, &fpm.model_path, fpm.dcp_size)?;
+    if fpm.interpolation == FpmInterpolation::Direct
+        && match fpm.phase {
+            FpmPhase::Prefill => cell.direct_prefill.is_empty(),
+            FpmPhase::Decode => cell.direct_decode.is_empty(),
+        }
+    {
+        let phase = fpm.phase.as_str();
+        return Err(AicError::PerfDatabase(format!(
+            "direct FPM {phase} interpolation has no genuine measurements for {:?} at {}. \
+             Collect genuine {phase} FPM rows for this model and identity, or provide a \
+             systems root containing them.",
+            fpm.model_path,
+            db.data_root.display(),
+        )));
+    }
+    Ok(())
+}
+
 struct Availability<'a> {
     db: &'a PerfDatabase,
     tables: HashMap<String, bool>,
@@ -162,14 +185,7 @@ impl Availability<'_> {
                 }
             }
             TokenScale(scale) => return self.op(&scale.op),
-            FpmForward(fpm) => {
-                self.db.fpm_forward.select_cell(
-                    &fpm.match_identity,
-                    &fpm.model_path,
-                    fpm.dcp_size,
-                )?;
-                return Ok(());
-            }
+            FpmForward(fpm) => return validate_fpm(self.db, fpm),
             // These families have no analytic/empirical implementation.
             MoeAllToAll(_) | MoeExpertCompute(_) | Dsv4MegaMoe(_)
                 if !matches!(
@@ -446,6 +462,62 @@ mod tests {
         write_parquet(path, &[Col::Str("kernel_source", vec!["fixture"])]);
     }
 
+    #[test]
+    fn direct_fpm_readiness_requires_genuine_rows_only_for_requested_phases() {
+        use crate::perf_database::fpm_forward::tests::{RowSpec, default_identity, write_pair};
+
+        for missing_phase in [FpmPhase::Prefill, FpmPhase::Decode] {
+            for fake_only in [false, true] {
+                let root = systems();
+                let data = root.path().join("data/b200_sxm/vllm/0.25.1");
+                std::fs::create_dir_all(&data).unwrap();
+                let present_phase = match missing_phase {
+                    FpmPhase::Prefill => FpmPhase::Decode,
+                    FpmPhase::Decode => FpmPhase::Prefill,
+                };
+                let row = |phase: FpmPhase, regime| RowSpec {
+                    workload_kind: phase.as_str(),
+                    total_prefill_tokens: if phase == FpmPhase::Prefill { 8 } else { 0 },
+                    kv_seed_regime: Some(regime),
+                    ..RowSpec::default()
+                };
+                let mut rows = vec![row(present_phase, "real_kv")];
+                if fake_only {
+                    rows.push(row(missing_phase, "fake_fallback"));
+                }
+                write_pair(&data, &rows);
+                let db = PerfDatabase::load(root.path(), "b200_sxm", "vllm", "0.25.1").unwrap();
+                let op = |phase: FpmPhase| {
+                    Op::FpmForward(FpmForwardOp {
+                        dcp_size: None,
+                        name: format!("fpm_forward_{}", phase.as_str()),
+                        phase,
+                        model_path: "org/model-a".into(),
+                        match_identity: default_identity(4),
+                        weight_bytes: 0.0,
+                        verify_width: 1,
+                        interpolation: FpmInterpolation::Direct,
+                        sol_ops: vec![],
+                        original_fmha_quant_mode: None,
+                    })
+                };
+                let present = op(present_phase);
+                let missing = op(missing_phase);
+                // A single-phase operation requires only its own measurements.
+                validate(&db, [&present].into_iter()).unwrap();
+                for requested in [vec![&missing], vec![&present, &missing]] {
+                    let error = validate(&db, requested.into_iter())
+                        .unwrap_err()
+                        .to_string();
+                    assert!(error.contains(missing_phase.as_str()), "{error}");
+                    assert!(error.contains("no genuine measurements"), "{error}");
+                    assert!(error.contains("Collect genuine"), "{error}");
+                    assert!(error.contains(data.to_str().unwrap()), "{error}");
+                }
+            }
+        }
+    }
+
     #[test]
     fn dsv41_stage_requires_measured_module_for_silicon() {
         use crate::operators::dsv41::{Dsv41LinearOp, Dsv41StageOp};