fix(fpm kvwarm): hybrid (KDA/Mamba) fixes for the real-KV decode warm-up, validated on GLM-5.3-Flash

components/src/dynamo/vllm/args.py

@@ -437,9 +437,11 @@ def update_engine_config_with_dynamo(
         benchmark_config: Dict[str, Any] = {
             "mode": dynamo_config.benchmark_mode,
             "randomize_kda_state": dynamo_config.benchmark_randomize_kda_state,
+            "hybrid_live_state": dynamo_config.benchmark_hybrid_live_state,
             "warmup_iterations": dynamo_config.benchmark_warmup_iterations,
             "output_path": dynamo_config.benchmark_output_path,
             "timeout": dynamo_config.benchmark_timeout,
+            "max_batch_size": dynamo_config.benchmark_max_batch_size,
             "collect_imbalanced": dynamo_config.benchmark_collect_imbalanced,
         }
         explicit_points = dynamo_config._benchmark_points

components/src/dynamo/vllm/backend_args.py

@@ -480,6 +480,20 @@ def add_arguments(self, parser) -> None:
                 "vLLM GPU worker. These are synthetic states, not real context history."
             ),
         )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--benchmark-hybrid-live-state",
+            env_var="DYN_BENCHMARK_HYBRID_LIVE_STATE",
+            default=False,
+            help=(
+                "Hybrid (KDA/Mamba) models: let the real-KV decode warm-up run with "
+                "recurrent-state groups forked from the parked chain's live state block "
+                "instead of skipping the warm-up. Attention KV is the chain's real prefix; "
+                "the recurrent state is a valid but deeper-context state (shallow points "
+                "read a few percent fast). Mutually exclusive with "
+                "--benchmark-randomize-kda-state."
+            ),
+        )
         add_argument(
             g,
             flag_name="--benchmark-warmup-iterations",
@@ -524,6 +538,21 @@ def add_arguments(self, parser) -> None:
                 "A bounded cleanup grace still fails closed if no result is written."
             ),
         )
+        add_argument(
+            g,
+            flag_name="--benchmark-max-batch-size",
+            env_var="DYN_BENCHMARK_MAX_BATCH_SIZE",
+            default=None,
+            arg_type=int,
+            help=(
+                "Cap the decode benchmark batch-size axis without changing the "
+                "engine's own limits. Points above the cap are never generated "
+                "(not measured, not skipped). Decoupled from --max-num-seqs: keep "
+                "the engine at its deployment concurrency while sweeping only the "
+                "batch range the performance model needs; setting --max-num-seqs "
+                "to at least twice this cap keeps every swept rung warmable."
+            ),
+        )
 
 
 # @dataclass()
@@ -571,9 +600,11 @@ class DynamoVllmConfig(ConfigBase):
     benchmark_mode: Optional[BenchmarkMode] = None
     benchmark_points_file: Optional[str] = None
     benchmark_randomize_kda_state: bool = False
+    benchmark_hybrid_live_state: bool = False
     benchmark_warmup_iterations: int = 5
     benchmark_output_path: str = "/tmp/benchmark_results.json"
     benchmark_timeout: int = 900
+    benchmark_max_batch_size: Optional[int] = None
     prefill_max_new_token_samples: int = 64
     prefill_max_kv_read_token_samples: int = 16
     decode_max_kv_read_token_samples: int = 128
@@ -693,6 +724,18 @@ def _resolve_legacy_benchmark_sampling(self) -> None:
             setattr(self, replacement_name, mapped_value)
 
     def _validate_benchmark_sampling(self) -> None:
+        if self.benchmark_hybrid_live_state and self.benchmark_randomize_kda_state:
+            raise ValueError(
+                "--benchmark-hybrid-live-state and --benchmark-randomize-kda-state "
+                "are mutually exclusive"
+            )
+        if self.benchmark_hybrid_live_state and self.benchmark_mode not in (
+            "decode",
+            "agg",
+        ):
+            raise ValueError(
+                "--benchmark-hybrid-live-state requires --benchmark-mode decode or agg"
+            )
         if self.benchmark_randomize_kda_state and self.benchmark_mode not in (
             "decode",
             "agg",