fix: read llm_config decoders in the remaining FPM paths

crates/core/src/engine/kv_manager/grouped.rs

@@ -56,13 +56,22 @@ impl GroupedKvPool {
         Some(projected)
     }
 
+    /// Bytes that allocate() reserves for this forward, computed without building the lease.
     pub(crate) fn required_bytes(
         &self,
         computed_before: usize,
         computed_after: usize,
     ) -> Option<u64> {
-        self.project(computed_before, computed_after)
-            .map(|lease| lease.bytes)
+        if computed_after < computed_before {
+            return None;
+        }
+        self.groups.iter().try_fold(0u64, |total, group| {
+            let range = group
+                .block_range(computed_before as u64, computed_after as u64)
+                .ok()?;
+            let bytes = (range.end - range.start).checked_mul(group.page_size_bytes)?;
+            total.checked_add(bytes)
+        })
     }
 
     /// Reserve every group atomically. Expired blocks become reusable in this
@@ -113,6 +122,77 @@ mod tests {
     use super::*;
     use crate::perfmodel::FpmCacheKind;
 
+    fn group(kind: FpmCacheKind, block: u32, page: u64, window: Option<u32>) -> FpmCacheGroup {
+        FpmCacheGroup {
+            name: "group".into(),
+            kind,
+            num_layers: 1,
+            block_size_tokens: block,
+            page_size_bytes: page,
+            sliding_window: window,
+        }
+    }
+
+    #[test]
+    fn required_bytes_equals_the_projected_lease_bytes() {
+        let full = group(FpmCacheKind::Attention, 4, 8, None);
+        let window = group(FpmCacheKind::Attention, 2, 6, Some(3));
+        let convolution = group(FpmCacheKind::Convolution, 2, 2, Some(2));
+        let largest_page = 1 << 53; // The largest page_size_bytes that FpmCacheGroup::validate accepts.
+        let pools = [
+            GroupedKvPool::new(vec![full.clone(), window.clone(), convolution.clone()], 0),
+            GroupedKvPool::new(vec![full], 0),
+            GroupedKvPool::new(vec![window], 0),
+            GroupedKvPool::new(vec![convolution], 0),
+            GroupedKvPool::new(
+                vec![group(FpmCacheKind::Attention, 1, largest_page, None)],
+                0,
+            ),
+            GroupedKvPool::new(
+                vec![group(FpmCacheKind::Attention, 1, largest_page, None); 2],
+                0,
+            ),
+            // FpmCacheGroup::validate rejects a convolution group without a window.
+            GroupedKvPool::new(vec![group(FpmCacheKind::Convolution, 2, 2, None)], 0),
+            // No group exists to reject a shrinking target; only the guard in required_bytes does.
+            GroupedKvPool::new(vec![], 0),
+        ];
+        let mut spans: Vec<(usize, usize)> = (0..=12)
+            .flat_map(|before| (0..=12).map(move |after| (before, after)))
+            .collect();
+        spans.extend([(0, 1024), (0, 2048), (0, usize::MAX)]);
+        for (index, pool) in pools.iter().enumerate() {
+            for &(before, after) in &spans {
+                assert_eq!(
+                    pool.required_bytes(before, after),
+                    pool.project(before, after).map(|lease| lease.bytes),
+                    "pool {index}, {before}..{after}"
+                );
+            }
+        }
+        // Target 0, inside a block, at a block boundary, then window eviction. A shrinking target projects nothing.
+        for (before, after, bytes) in [
+            (0, 0, Some(0)),
+            (0, 3, Some(24)),
+            (0, 4, Some(24)),
+            (4, 8, Some(40)),
+            (8, 9, Some(40)),
+            (9, 9, Some(38)),
+            (5, 4, None),
+        ] {
+            assert_eq!(
+                pools[0].required_bytes(before, after),
+                bytes,
+                "{before}..{after}"
+            );
+        }
+        // The pool has no 2**53 limit, unlike FpmCacheGroup::bytes_for_forward.
+        assert_eq!(pools[4].required_bytes(0, 2), Some(1 << 54));
+        assert_eq!(pools[4].required_bytes(0, 2048), None); // checked_mul overflows u64.
+        assert_eq!(pools[5].required_bytes(0, 1024), None); // checked_add overflows u64.
+        assert_eq!(pools[6].required_bytes(0, 4), None);
+    }
+
     #[test]
     fn mixed_groups_reserve_atomically_and_reuse_expired_pages() {
         let mut pool = GroupedKvPool::new(

docs/perf-model/fpm-self-service/implementation.md

@@ -699,7 +699,7 @@ Setup reads that file and an adjacent `hf_quant_config.json` ModelOpt sidecar, w
 
 With no parallelism flags, guided setup asks for missing shared precision/layout facts, shows a small topology shortlist with resource reasons, and lets you choose one worker with `--output`, or several with [`--output-dir`](#onboard-multiple-parallel-configurations). Enter selects only the displayed single default, and only when all required profile inputs are resolved and its estimated bytes fit the budget. Otherwise a candidate selection is required before setup resolves the remaining precision and cache geometry. Any explicit parallelism flag bypasses suggestions and retains the existing topology validation. Each chosen topology then follows the same profile review, edit, accept and cancellation flow below.
 
-When present, `text_config` must be one nonempty decoder configuration object. Only its geometry is used; wrapper and encoder dimensions are never merged into it. Otherwise, setup reads the flat decoder fields even when vision or audio metadata is present. Nested decoder architecture/model-type declarations select supported decoder validation and resource estimates. The profile preserves the declared architecture used by collection: nested `architectures` when present, otherwise the wrapper architecture. Provenance records both identities when they differ. A wrapper architecture never establishes an unknown nested decoder's resource layout. Shared outer dtype and quantization metadata are inherited only when the text section does not declare that metadata. The checkpoint identity hint and SHA-256 remain those of the original document. `model_max_length` is accepted as a context-limit alias; conflicting context declarations fail explicitly.
+When present, `text_config` must be one nonempty decoder configuration object. Only its geometry is used; wrapper and encoder dimensions are never merged into it. A nonempty `llm_config` object is read the same way. It takes precedence over `text_config` only for wrappers that AISimulate maps to `llm_config`, such as `NemotronH_Omni_Reasoning_V3`. For those wrappers, `llm_config` is the decoder key whenever the config holds it. The FPM collector and the profile architecture check choose the decoder key the same way. Setup, the FPM collector and the profile architecture check refuse a decoder object that holds a `text_config` key or a nonempty `llm_config` object. Such a config nests two decoders. When the chosen decoder key does not hold one nonempty object, the three refuse it the same way. When no decoder key is chosen, setup reads the flat decoder fields even when vision or audio metadata is present. Nested decoder architecture/model-type declarations select supported decoder validation and resource estimates. The profile preserves the declared architecture used by collection: nested `architectures` when present, otherwise the wrapper architecture. Provenance records both identities when they differ. A wrapper architecture never establishes an unknown nested decoder's resource layout. Shared outer dtype and quantization metadata are inherited only when the text section does not declare that metadata. The checkpoint identity hint and SHA-256 remain those of the original document. `model_max_length` is accepted as a context-limit alias; conflicting context declarations fail explicitly.
 
 Scripted intake and interactive final review state that FPM models the text decoder only. Config-derived estimates exclude multimodal encoders, projectors, preprocessing and other non-text components. Observed runtime cache capacity accounts for all components actually loaded by the worker, while timing describes the text decoder only. The saved profile preserves that notice in provenance through planning and generated prediction/recommendation configurations. Text-decoder cache layouts that `config.json` cannot establish, such as linear-attention, recurrent state or an architecture that intake does not recognize, are not rejected: their [cache geometry stays pending the runtime probe](#cache-geometry-pending-the-runtime-probe). Other known incompatible layouts remain rejected. MLA/DSA layouts and configs that name no architecture require explicit resource accounting.
 

python/aisimulate/collector/fpm_forward/database.py

@@ -19,6 +19,7 @@
 import yaml
 
 from aisimulate.fpm_contract import FPM_RESOLVED_CONFIG_GLOB
+from aisimulate_core.decoder_config import decoder_config_key
 from aisimulate_core.sdk.fpm_identity import EXECUTION_COLUMNS, LEGACY_EXECUTION_IDENTITY
 
 from .capabilities import GENERIC_RUNTIME_TEMPLATE_ID
@@ -142,13 +143,13 @@ def _dense_synthetic_kv_eligible(
     ):
         return False
     source = plan.capability.model_config.payload
-    text = source.get("text_config")
+    decoder = source.get(decoder_config_key(source))
     # Dynamo checks both configs for experts before considering recurrent state:
     # https://github.com/ai-dynamo/dynamo/blob/92cf45e2e6707ac32b1fee275f949270fdd61200/components/src/dynamo/vllm/instrumented_scheduler.py#L5585
     # Its skip reason alone therefore does not establish dense full attention.
     if any(
         config.get(key)
-        for config in (source, text)
+        for config in (source, decoder)
         if isinstance(config, dict)
         for key in ("num_local_experts", "num_experts", "n_routed_experts", "moe_num_experts")
     ):

python/aisimulate/collector/fpm_forward/model_capability.py

@@ -21,7 +21,12 @@
     _get_model_config_path,
     _parse_hf_config_json,
 )
-from aisimulate_core.sdk.fpm_model_metadata import attach_quantization_sidecar, inherit_decoder_precision
+from aisimulate_core.decoder_config import decoder_config_problem
+from aisimulate_core.sdk.fpm_model_metadata import (
+    attach_quantization_sidecar,
+    decoder_config_key,
+    inherit_decoder_precision,
+)
 
 _ATTENTION_SOURCE_OPS = {
     "dsa_module": frozenset({"dsa_context_module", "dsa_generation_module"}),
@@ -66,18 +71,25 @@ def payload(self) -> dict[str, Any]:
 
     @property
     def effective_payload(self) -> dict[str, Any]:
-        """Return decoder metadata, inheriting only identity and shared precision."""
+        """Return decoder metadata, inheriting only identity and shared precision.
+
+        A decoder config that is not one nonempty object, or that holds a decoder key, raises ValueError, as config
+        intake refuses it. Only a config with no decoder key is read flat.
+        """
 
         payload = self.payload
-        text_config = payload.get("text_config")
-        if isinstance(text_config, dict):
-            effective, _ = inherit_decoder_precision(payload, text_config)
+        key = decoder_config_key(payload)
+        decoder = payload.get(key)
+        if key is not None and (problem := decoder_config_problem(key, decoder)) is not None:
+            raise ValueError(f"{problem}; supply one text decoder config")
+        if isinstance(decoder, dict):
+            effective, _ = inherit_decoder_precision(payload, decoder)
             if effective.get("architectures") is None and payload.get("architectures") is not None:
                 effective["architectures"] = payload["architectures"]
         else:
             effective = payload
         sidecar = effective.get("hf_quant_config")
-        if sidecar is None and isinstance(text_config, dict) and isinstance(payload.get("hf_quant_config"), dict):
+        if sidecar is None and isinstance(decoder, dict) and isinstance(payload.get("hf_quant_config"), dict):
             try:
                 return attach_quantization_sidecar(effective, payload["hf_quant_config"])
             except ValueError:
@@ -155,11 +167,12 @@ def _load_json_mapping(path: Path, *, description: str) -> tuple[dict[str, Any],
 
 
 def _attach_modelopt_sidecar(payload: dict[str, Any], sidecar: dict[str, Any]) -> dict[str, Any]:
-    text_config = payload.get("text_config")
+    key = decoder_config_key(payload)
+    decoder = payload.get(key)
     if payload.get("hf_quant_config") is not None and payload["hf_quant_config"] != sidecar:
         raise ValueError("conflicting inline hf_quant_config and adjacent hf_quant_config.json")
-    if isinstance(text_config, dict):
-        decoder, _ = inherit_decoder_precision(payload, text_config)
+    if isinstance(decoder, dict):
+        decoder, _ = inherit_decoder_precision(payload, decoder)
         attach_quantization_sidecar(decoder, sidecar)  # Reject contradictory source declarations before planning.
     else:
         attach_quantization_sidecar(payload, sidecar)

python/aisimulate/src/aisimulate/capacity.py

@@ -595,16 +595,18 @@ def estimate_kv_bytes_per_token(
 def resolve_model_context_length(model_name: str) -> int:
     """Resolve ``context_length: max`` from a local, cached, or HF config."""
 
+    from aisimulate_core.decoder_config import decoder_config_key, decoder_config_problem
     from aisimulate_core.sdk.memory import NaiveKVCacheEstimator
 
     # Both modules ship in the same distribution; reuse the canonical loader.
     config = NaiveKVCacheEstimator._load_config(model_name, allow_hf_config_download=True)
     if not isinstance(config, dict):
         raise ValueError(f"could not load Hugging Face config for {model_name!r}")
-    text_config = config.get("text_config")
-    mappings = [config]
-    if isinstance(text_config, dict):
-        mappings.insert(0, text_config)
+    key = decoder_config_key(config)
+    decoder = config.get(key)
+    if key is not None and (problem := decoder_config_problem(key, decoder)) is not None:
+        raise ValueError(f"Hugging Face config for {model_name!r}: {problem}")
+    mappings = [decoder, config] if decoder is not None else [config]
     for mapping in mappings:
         for name in (
             "max_position_embeddings",

python/aisimulate/src/aisimulate/support/config_profile.py

@@ -604,6 +604,10 @@ def _validate_architecture(raw: Mapping[str, Any], architecture: str | None, exp
 
 def load_model_config(path: str | Path) -> ModelConfig:
     """Load local config/ModelOpt metadata with separate byte-level provenance."""
+    # Imported here, not at module level: tests/unit/test_fpm_config_imports.py and the repository-root
+    # tests/test_resource_supervision.py fail when aisimulate_core.decoder_config loads before a config is read.
+    from aisimulate_core.decoder_config import decoder_config_key, nested_decoder_ambiguity
+
     path = Path(path).expanduser().absolute()
     payload = path.read_bytes()
     # JSON Infinity and NaN parse as floats: fields intake reads reject them, and other fields are never read.
@@ -619,17 +623,18 @@ def load_model_config(path: str | Path) -> ModelConfig:
             "for all components actually loaded by the worker; full multimodal latency is not modeled."
         )
     }
-    if "text_config" in document:
-        text_config = document["text_config"]
-        if not isinstance(text_config, dict) or not text_config:
+    decoder_key = decoder_config_key(document)
+    if decoder_key is not None:
+        decoder = document[decoder_key]
+        if not isinstance(decoder, dict) or not decoder:
             raise ValueError(
-                "text_config must be a nonempty object describing one text decoder; "
+                f"{decoder_key} must be a nonempty object describing one text decoder; "
                 "supply its decoder config or a complete --fpm-profile"
             )
-        if "text_config" in text_config:
-            raise ValueError("ambiguous nested text_config; supply one text decoder config or a complete --fpm-profile")
-        raw = dict(text_config)
-        notes["decoder_config"] = "config text_config; outer model and encoder geometry are excluded"
+        if (ambiguity := nested_decoder_ambiguity(decoder_key, decoder)) is not None:
+            raise ValueError(f"{ambiguity}; supply one text decoder config or a complete --fpm-profile")
+        raw = dict(decoder)
+        notes["decoder_config"] = f"config {decoder_key}; outer model and encoder geometry are excluded"
         inherited = []
         for keys, inferred_keys in (
             (("dtype", "torch_dtype"), ()),
@@ -647,7 +652,7 @@ def load_model_config(path: str | Path) -> ModelConfig:
                     inherited.append(key)
         if inherited:
             notes["shared_metadata"] = (
-                f"config-level {', '.join(inherited)} inherited because text_config does not declare that metadata"
+                f"config-level {', '.join(inherited)} inherited because {decoder_key} does not declare that metadata"
             )
     for key in _UNTYPED_READ_FIELDS:
         _require_finite(raw.get(key), key)
@@ -682,9 +687,10 @@ def load_model_config(path: str | Path) -> ModelConfig:
                 f"config wrapper architecture={architecture} retained for checkpoint/collector identity"
             )
         notes["decoder_architecture"] = (
-            f"text_config decoder architecture={decoder_architecture}; used for decoder validation/resource estimates"
+            f"{decoder_key} decoder architecture={decoder_architecture}; used for decoder validation/resource estimates"
             if decoder_architecture
-            else "text_config does not identify a decoder architecture; decoder resource bounds require explicit input"
+            else f"{decoder_key} does not identify a decoder architecture; "
+            "decoder resource bounds require explicit input"
         )
     inkling = architecture == _INKLING_ARCHITECTURE
     # A runtime-resolved layout may use zero or absent ordinary attention dimensions.