diff --git a/crates/core/src/python.rs b/crates/core/src/python.rs
index bfd901e19..ad3088687 100644
--- a/crates/core/src/python.rs
+++ b/crates/core/src/python.rs
@@ -54,6 +54,8 @@ struct RuntimeTraffic {
     #[serde(default)]
     osl: Option<usize>,
     #[serde(default)]
+    cached_prefix_tokens: Option<usize>,
+    #[serde(default)]
     request_count: Option<usize>,
     #[serde(default)]
     turns_per_session: Option<usize>,
@@ -120,6 +122,16 @@ struct AicTimingConfig {
     free_gpu_memory_fraction: Option<f64>,
     #[serde(default)]
     systems_path: Option<String>,
+    #[serde(default)]
+    moe_backend: Option<String>,
+    #[serde(default)]
+    attention_backend: Option<String>,
+    #[serde(default)]
+    enable_wideep: bool,
+    #[serde(default)]
+    enable_eplb: bool,
+    #[serde(default)]
+    wideep_num_slots: Option<u32>,
 }
 
 const fn one() -> u32 {
@@ -231,6 +243,11 @@ impl AicTimingModel {
             kwargs.set_item("nextn", config.nextn)?;
             kwargs.set_item("kv_block_size", config.kv_block_size)?;
             kwargs.set_item("systems_path", config.systems_path.as_deref())?;
+            kwargs.set_item("moe_backend", config.moe_backend.as_deref())?;
+            kwargs.set_item("attention_backend", config.attention_backend.as_deref())?;
+            kwargs.set_item("enable_wideep", config.enable_wideep)?;
+            kwargs.set_item("enable_eplb", config.enable_eplb)?;
+            kwargs.set_item("wideep_num_slots", config.wideep_num_slots)?;
             let spec = sdk.getattr("compile_engine")?.call(
                 (
                     config.model.as_str(),
@@ -660,8 +677,16 @@ fn build_runtime_input(
     } else {
         1
     };
+    let cached_prefix_tokens = traffic.cached_prefix_tokens.unwrap_or(0);
     let trace = Trace::synthetic(SyntheticTraceSpec {
-        block_size: engine_block_size,
+        // A one-token trace block preserves prefixes that are not aligned to
+        // the engine's scheduler block size. The driver still hashes them at
+        // `engine_block_size` when it constructs replay requests.
+        block_size: if cached_prefix_tokens == 0 {
+            engine_block_size
+        } else {
+            1
+        },
         num_sessions: sessions,
         turns_per_session: turns,
         input_tokens: LengthSpec {
@@ -672,6 +697,7 @@ fn build_runtime_input(
             mean: traffic.osl.context("synthetic traffic requires osl")?,
             stddev: 0.0,
         },
+        cached_prefix_tokens,
         shared_prefix_ratio: traffic.shared_prefix_ratio.unwrap_or(0.0),
         num_prefix_groups: traffic.num_prefix_groups.unwrap_or(0),
         first_turn_arrivals: synthetic_arrivals(&traffic)?,
@@ -877,6 +903,11 @@ mod tests {
             mem_fraction_static: None,
             free_gpu_memory_fraction: None,
             systems_path: None,
+            moe_backend: None,
+            attention_backend: None,
+            enable_wideep: false,
+            enable_eplb: false,
+            wideep_num_slots: None,
         }
     }
 
@@ -926,6 +957,28 @@ mod tests {
         );
     }
 
+    #[test]
+    fn aic_timing_config_deserializes_engine_controls() {
+        let config: AicTimingConfig = serde_json::from_value(serde_json::json!({
+            "model": "test-model",
+            "backend": "sglang",
+            "system": "test-system",
+            "tp": 1,
+            "moe_backend": "deepep_moe",
+            "attention_backend": "fa3",
+            "enable_wideep": true,
+            "enable_eplb": true,
+            "wideep_num_slots": 64
+        }))
+        .unwrap();
+
+        assert_eq!(config.moe_backend.as_deref(), Some("deepep_moe"));
+        assert_eq!(config.attention_backend.as_deref(), Some("fa3"));
+        assert!(config.enable_wideep);
+        assert!(config.enable_eplb);
+        assert_eq!(config.wideep_num_slots, Some(64));
+    }
+
     #[test]
     fn json_binding_retains_authored_per_request_correlation() {
         let spec = ReplaySpec {
diff --git a/crates/core/src/replay/loadgen/tests.rs b/crates/core/src/replay/loadgen/tests.rs
index 42133601a..6b525f6ef 100644
--- a/crates/core/src/replay/loadgen/tests.rs
+++ b/crates/core/src/replay/loadgen/tests.rs
@@ -718,6 +718,7 @@ fn test_partition_by_session_round_robin_keeps_sessions_intact() {
             mean: 2,
             stddev: 0.0,
         },
+        cached_prefix_tokens: 0,
         shared_prefix_ratio: 0.5,
         num_prefix_groups: 2,
         first_turn_arrivals: ArrivalSpec::Burst,
@@ -754,6 +755,7 @@ fn test_synthetic_prefix_groups_share_prefixes_within_group() {
             mean: 2,
             stddev: 0.0,
         },
+        cached_prefix_tokens: 0,
         shared_prefix_ratio: 0.5,
         num_prefix_groups: 2,
         first_turn_arrivals: ArrivalSpec::Burst,
@@ -778,6 +780,46 @@ fn test_synthetic_prefix_groups_share_prefixes_within_group() {
     assert_eq!(suffixes.len(), trace.sessions.len());
 }
 
+#[test]
+fn test_synthetic_exact_cached_prefix_shares_only_requested_tokens() {
+    let trace = Trace::synthetic(SyntheticTraceSpec {
+        block_size: 1,
+        num_sessions: 3,
+        turns_per_session: 1,
+        input_tokens: LengthSpec {
+            mean: 8,
+            stddev: 0.0,
+        },
+        output_tokens: LengthSpec {
+            mean: 2,
+            stddev: 0.0,
+        },
+        cached_prefix_tokens: 3,
+        shared_prefix_ratio: 0.0,
+        num_prefix_groups: 0,
+        first_turn_arrivals: ArrivalSpec::Burst,
+        inter_turn_delays: DelaySpec::None,
+        seed: 42,
+        arrival_seed: 42,
+    })
+    .unwrap();
+
+    let prompts = trace
+        .sessions
+        .iter()
+        .map(|session| session.turns[0].synthesize_tokens(1).unwrap())
+        .collect::<Vec<_>>();
+    assert!(prompts.windows(2).all(|pair| pair[0][..3] == pair[1][..3]));
+    assert_eq!(
+        prompts
+            .iter()
+            .map(|tokens| tokens[3..].to_vec())
+            .collect::<HashSet<_>>()
+            .len(),
+        prompts.len()
+    );
+}
+
 #[test]
 fn test_synthetic_arrival_mode_changes_timestamps_only() {
     let build = |first_turn_arrivals, arrival_seed| {
@@ -793,6 +835,7 @@ fn test_synthetic_arrival_mode_changes_timestamps_only() {
                 mean: 4,
                 stddev: 1.0,
             },
+            cached_prefix_tokens: 0,
             shared_prefix_ratio: 0.5,
             num_prefix_groups: 5,
             first_turn_arrivals,
diff --git a/crates/core/src/replay/loadgen/trace.rs b/crates/core/src/replay/loadgen/trace.rs
index 5211342c9..995ee72fe 100644
--- a/crates/core/src/replay/loadgen/trace.rs
+++ b/crates/core/src/replay/loadgen/trace.rs
@@ -784,6 +784,13 @@ impl Trace {
                 spec.shared_prefix_ratio
             );
         }
+        if spec.cached_prefix_tokens % spec.block_size != 0 {
+            bail!(
+                "cached_prefix_tokens {} must align to synthetic trace block_size {}",
+                spec.cached_prefix_tokens,
+                spec.block_size
+            );
+        }
 
         let mut rng = StdRng::seed_from_u64(spec.seed);
         let mut sessions = Vec::with_capacity(spec.num_sessions);
@@ -803,13 +810,24 @@ impl Trace {
             for turn_idx in 0..spec.turns_per_session {
                 let input_length = sample_length(&spec.input_tokens, 1, &mut rng);
                 let max_output_tokens = sample_length(&spec.output_tokens, 1, &mut rng);
+                if spec.cached_prefix_tokens > input_length {
+                    bail!(
+                        "cached_prefix_tokens {} exceeds sampled synthetic input length {}",
+                        spec.cached_prefix_tokens,
+                        input_length
+                    );
+                }
                 let num_blocks = input_length.div_ceil(spec.block_size);
-                let prefix_blocks =
+                let grouped_prefix_blocks =
                     ((num_blocks as f64) * spec.shared_prefix_ratio).round() as usize;
-                let prefix_blocks = prefix_blocks.min(num_blocks);
+                let grouped_prefix_blocks = grouped_prefix_blocks.min(num_blocks);
+                let cached_prefix_blocks = spec.cached_prefix_tokens / spec.block_size;
                 let mut hash_ids = Vec::with_capacity(num_blocks);
 
-                for block_idx in 0..prefix_blocks {
+                for block_idx in 0..cached_prefix_blocks {
+                    hash_ids.push(0xCA_C0_0000_0000_0000 | block_idx as u64);
+                }
+                for block_idx in cached_prefix_blocks..grouped_prefix_blocks {
                     if let Some(group_id) = group_id {
                         hash_ids.push(0xD00D_0000_0000_0000 | (group_id << 32) | block_idx as u64);
                     }
diff --git a/crates/core/src/replay/loadgen/types.rs b/crates/core/src/replay/loadgen/types.rs
index bb9f8a8dc..f85315326 100644
--- a/crates/core/src/replay/loadgen/types.rs
+++ b/crates/core/src/replay/loadgen/types.rs
@@ -335,6 +335,7 @@ pub struct SyntheticTraceSpec {
     pub turns_per_session: usize,
     pub input_tokens: LengthSpec,
     pub output_tokens: LengthSpec,
+    pub cached_prefix_tokens: usize,
     pub shared_prefix_ratio: f64,
     pub num_prefix_groups: usize,
     pub first_turn_arrivals: ArrivalSpec,
diff --git a/docs/cli/migrate-from-aiconfigurator.md b/docs/cli/migrate-from-aiconfigurator.md
index deae8b269..7b66edf27 100644
--- a/docs/cli/migrate-from-aiconfigurator.md
+++ b/docs/cli/migrate-from-aiconfigurator.md
@@ -27,6 +27,114 @@ by the unified command surface. A legacy command becomes an AISimulate recommend
 | `--tpot` | `evaluation.sla.itl_ms` | Per-request goodput uses average ITL; strict mode compares aggregate mean TPOT |
 | `--strict-sla` | `optimization.strict_sla: true` | Reject before scalar ranking or Pareto dominance |
 
+## Engine and request controls
+
+The additional controls below remain flat fields under `engine`. They are concrete, pinned inputs;
+adding them does not create new recommendation dimensions or enlarge the search space.
+
+| AIConfigurator input | AISimulate recommendation | Notes |
+|---|---|---|
+| `--prefix` | `traffic.source.cached_prefix_tokens` | Exact cached tokens shared by synthetic requests; must not exceed the input length |
+| `--max-seq-len` | `engine.context_length` | Search capacity limit; AISimulate also accepts `max`, and the engine runner enforces it directly for vLLM |
+| `--nextn` | `engine.nextn` | Concrete MTP draft depth from 0 through 5 |
+| `--nextn-accepted` | `engine.nextn_accepted` | Required when `nextn > 0`; measured average accepted draft tokens in `[0, nextn]` |
+| `--enable-chunked-prefill` | `engine.enable_chunked_prefill: true` | Enables chunking on aggregated and prefill workers; decode retains its backend default |
+| `--free-gpu-memory-fraction` | `engine.free_gpu_memory_fraction` | Shared override for every active worker role |
+| `--enable-wideep` / `Task.enable_wideep` | `engine.enable_wideep: true` | Requires an MoE model; the legacy CLI flag is deprecated even though the Task field remains available |
+| `Task.enable_eplb` | `engine.enable_eplb: true` | Requires an MoE model |
+| `Task.wideep_num_slots` | `engine.wideep_num_slots` | Positive EPLB slot count |
+| `--moe-backend` / `Task.moe_backend` | `engine.moe_backend` | `deepep_moe` or `megamoe`; SGLang only |
+| `Task.attention_backend` | `engine.attention_backend` | `flashinfer` or `fa3`; SGLang MLA models only |
+| `--gemm-quant-mode` | `engine.gemm_quant_mode` | Same AIConfigurator enum name |
+| `--moe-quant-mode` | `engine.moe_quant_mode` | Same AIConfigurator enum name; requires an MoE model |
+| `--kvcache-quant-mode` | `engine.kvcache_quant_mode` | Same AIConfigurator enum name |
+| `--fmha-quant-mode` | `engine.fmha_quant_mode` | Same AIConfigurator enum name |
+| `--comm-quant-mode` | `engine.comm_quant_mode` | Same AIConfigurator enum name |
+
+`--nextn auto` does not transfer as the string `auto`. Resolve it with AIConfigurator first, then
+copy the resulting integer depth and the measured `nextn_accepted` value. AISimulate does not infer
+an acceptance assumption.
+
+### Example with the new controls
+
+Legacy AIConfigurator command:
+
+```bash
+aiconfigurator cli default \
+  --model-path deepseek-ai/DeepSeek-R1 \
+  --system h200_sxm \
+  --backend sglang \
+  --total-gpus 8 \
+  --isl 4096 \
+  --osl 1024 \
+  --prefix 1024 \
+  --max-seq-len 8192 \
+  --nextn 3 \
+  --nextn-accepted 1.5 \
+  --enable-chunked-prefill \
+  --free-gpu-memory-fraction 0.85 \
+  --moe-backend deepep_moe \
+  --gemm-quant-mode fp8 \
+  --moe-quant-mode fp8 \
+  --kvcache-quant-mode fp8 \
+  --fmha-quant-mode fp8 \
+  --comm-quant-mode fp8
+```
+
+The equivalent pinned controls in `recommendation.yaml` are:
+
+```yaml
+traffic:
+  source:
+    type: synthetic
+    input_tokens: 4096
+    output_tokens: 1024
+    cached_prefix_tokens: 1024
+  load:
+    type: constant_rate
+    requests_per_second: 4
+  stop:
+    requests_per_load_unit: 10
+
+engine:
+  mode: aggregated
+  model: deepseek-ai/DeepSeek-R1
+  hardware: h200_sxm
+  backend: sglang
+  context_length: 8192
+  nextn: 3
+  nextn_accepted: 1.5
+  enable_chunked_prefill: true
+  free_gpu_memory_fraction: 0.85
+  moe_backend: deepep_moe
+  gemm_quant_mode: fp8
+  moe_quant_mode: fp8
+  kvcache_quant_mode: fp8
+  fmha_quant_mode: fp8
+  comm_quant_mode: fp8
+  workers:
+    aggregated: {}
+
+evaluation: {}
+
+optimization:
+  target: throughput
+  constraints:
+    max_candidate_gpus: 8
+```
+
+Run it with:
+
+```bash
+aisimulate recommend --config recommendation.yaml
+```
+
+The traffic rate and stop condition are explicit AISimulate choices; the legacy capacity sweep did
+not define an equivalent offered load. For disaggregated configurations, the shared engine controls
+apply to both roles, except decode retains its backend chunking default. Keep using
+`workers.prefill.kv_cache.capacity.memory_fraction` and
+`workers.decode.kv_cache.capacity.memory_fraction` when the two roles need different memory limits.
+
 ## Illustrative strict SLA translation
 
 The examples below demonstrate how the strict-SLA fields map, but they are not behaviorally
diff --git a/docs/sweeper/configuration.md b/docs/sweeper/configuration.md
index 7480c4273..8f9ed7db4 100644
--- a/docs/sweeper/configuration.md
+++ b/docs/sweeper/configuration.md
@@ -61,13 +61,24 @@ configuration for each candidate.
 | `backend` | `[vllm]` | engine backends to search |
 | `gpu_budget` | `32` | maximum GPUs per candidate |
 | `min_gpu_budget` | `None` | optional lower bound during enumeration |
-| `context_length` | `None` | optional KV-feasibility sequence length |
+| `context_length` | `None` | compatibility alias for `max_seq_len`; both must match if set |
+| `max_seq_len` | model maximum | sequence capacity used by KV feasibility and every engine's `max_model_len` |
 | `parallel_configs` | `[]` | optional pinned parallel configurations |
 | `startup_time` | `None` | optional simulated worker startup time |
 | `aic_nextn` | `None` | optional speculative-decoding depth |
+| `nextn_accepted` | `None` | required explicit expected accepted draft tokens when `aic_nextn` is set |
+| `enable_chunked_prefill` | `false` | enable chunking on aggregated/prefill roles; selected `max_num_batched_tokens` remains the exact context-token budget |
+| `enable_wideep`, `enable_eplb` | `false` | shared MoE WideEP/EPLB controls |
+| `wideep_num_slots` | `None` | positive EPLB slot count |
+| `moe_backend`, `attention_backend` | `None` | explicit supported MoE/MLA kernel backends |
+| `gemm_quant_mode`, `moe_quant_mode`, `kvcache_quant_mode`, `fmha_quant_mode`, `comm_quant_mode` | `None` | shared quantization overrides used by KV feasibility and AIC timing |
+| `free_gpu_memory_fraction` | role default | shared memory fraction, mapped to backend-native total/free-memory semantics |
 
 Each engine role also has lists for `max_num_batched_tokens` and `max_num_seqs`, plus pinned block
 size, GPU-memory-utilization, and prefix-caching fields. A one-item list pins a searched field.
+Shared controls apply to both active roles in a disaggregated candidate. Unsupported model/backend
+combinations are rejected before adapter preparation or replay, and the resolved values are retained
+in `Candidate.config.engine_request` and `ReplaySpec.backend_deployment.engine_request`.
 
 ## Pinned Parallel Configurations
 
diff --git a/docs/sweeper/traffic.md b/docs/sweeper/traffic.md
index 3e9df437d..bff2917b5 100644
--- a/docs/sweeper/traffic.md
+++ b/docs/sweeper/traffic.md
@@ -58,6 +58,7 @@ Every `Workload` field:
 | `num_request_ratio` | `float \| None` | `None` | Synthetic request count relative to the load: `num_requests = round(num_request_ratio * load)`. Required for synthetic. See below. |
 | `random_range_ratio` | `float` | `1.0` | Uniformly sample each synthetic ISL and OSL from `[int(ratio * configured_length), configured_length]`. Must be in `(0.0, 1.0]`; `1.0` preserves fixed lengths. Single-turn only. |
 | `random_seed` | `int` | `0` | Unsigned 64-bit seed for deterministic synthetic ISL/OSL sampling. |
+| `cached_prefix_tokens` | `int` | `0` | Exact common prompt-prefix length materialized as token IDs for KV-aware replay. It must not exceed the shortest randomized ISL and is independent of ratio-based locality controls. |
 | `shared_prefix_ratio` | `float` | `0.0` | Fraction of shared prefix across requests (cache-locality / prefix sharing). |
 | `num_prefix_groups` | `int` | `0` | Number of distinct shared-prefix groups. |
 | `turns_per_session` | `int` | `1` | Turns per multi-turn session. |
@@ -68,7 +69,7 @@ Every `Workload` field:
 | `replay_concurrency` | `int \| None` | `None` | Closed-loop in-flight cap **for a trace** (shape 1c); when set, trace timestamps are ignored. For synthetic closed-loop use `concurrency` instead. |
 
 The synthetic fields are `isl`, `osl`, `request_rate`, `concurrency`, `kv_load_ratio`,
-`num_request_ratio`, `random_range_ratio`, and `random_seed`;
+`num_request_ratio`, `random_range_ratio`, `random_seed`, and `cached_prefix_tokens`;
 `shared_prefix_ratio`, `num_prefix_groups`, `turns_per_session`, `inter_turn_delay_ms` are
 shared synthetic knobs carried by `ReplaySpec.workload`.
 
diff --git a/python/aisimulate/src/aiconfigurator_core/sdk/config_builders.py b/python/aisimulate/src/aiconfigurator_core/sdk/config_builders.py
index 1b32567ae..55add7270 100644
--- a/python/aisimulate/src/aiconfigurator_core/sdk/config_builders.py
+++ b/python/aisimulate/src/aiconfigurator_core/sdk/config_builders.py
@@ -32,6 +32,11 @@ def build_model_config(
     comm_quant_mode: str | None = None,
     forward_model: str | None = None,
     enable_encoder_dp: bool = True,
+    moe_backend: str | None = None,
+    attention_backend: str | None = None,
+    enable_wideep: bool = False,
+    enable_eplb: bool = False,
+    wideep_num_slots: int | None = None,
 ) -> ModelConfig:
     """Build a ModelConfig with optional quant mode overrides."""
     return ModelConfig(
@@ -47,6 +52,11 @@ def build_model_config(
         comm_quant_mode=CommQuantMode[comm_quant_mode] if comm_quant_mode else None,
         forward_model=forward_model or "op_level",
         enable_encoder_dp=enable_encoder_dp,
+        moe_backend=moe_backend,
+        attention_backend=attention_backend or "flashinfer",
+        enable_wideep=enable_wideep,
+        enable_eplb=enable_eplb,
+        wideep_num_slots=wideep_num_slots,
     )
 
 
diff --git a/python/aisimulate/src/aiconfigurator_core/sdk/engine.py b/python/aisimulate/src/aiconfigurator_core/sdk/engine.py
index 4343fd234..0a2e22a83 100644
--- a/python/aisimulate/src/aiconfigurator_core/sdk/engine.py
+++ b/python/aisimulate/src/aiconfigurator_core/sdk/engine.py
@@ -320,6 +320,11 @@ def compile_engine(
     kv_block_size: int | None = None,
     systems_path: str | None = None,
     forward_model: str | None = None,
+    moe_backend: str | None = None,
+    attention_backend: str | None = None,
+    enable_wideep: bool = False,
+    enable_eplb: bool = False,
+    wideep_num_slots: int | None = None,
 ) -> bytes:
     """Compile a model into bincoded ``EngineSpec`` bytes.
 
@@ -345,6 +350,11 @@ def compile_engine(
         moe_quant_mode=moe_quant_mode,
         comm_quant_mode=comm_quant_mode,
         forward_model=forward_model,
+        moe_backend=moe_backend,
+        attention_backend=attention_backend,
+        enable_wideep=enable_wideep,
+        enable_eplb=enable_eplb,
+        wideep_num_slots=wideep_num_slots,
     )
     # Apply MTP BEFORE get_model so the walked op lists carry the
     # (L+nextn)/L compute scale; accepted-token progress is applied above core.
diff --git a/python/aisimulate/src/aisimulate/__init__.py b/python/aisimulate/src/aisimulate/__init__.py
index 06f94b454..583a1e313 100644
--- a/python/aisimulate/src/aisimulate/__init__.py
+++ b/python/aisimulate/src/aisimulate/__init__.py
@@ -19,6 +19,7 @@
     "CoreRecommendationConfig": ("aisimulate.config", "CoreRecommendationConfig"),
     "EngineReplayRunner": ("aisimulate.runner", "EngineReplayRunner"),
     "EngineReplayRunnerFactory": ("aisimulate.runner", "EngineReplayRunnerFactory"),
+    "EngineRequestSpec": ("aisimulate.sweeper.replay", "EngineRequestSpec"),
     "InvalidRunnerError": ("aisimulate.runner", "InvalidRunnerError"),
     "ReplayOutputRequirements": ("aisimulate.sweeper.replay", "ReplayOutputRequirements"),
     "ReplayReport": ("aisimulate.sweeper.replay", "ReplayReport"),
@@ -52,6 +53,7 @@ def __dir__() -> list[str]:
     "CoreRecommendationConfig",
     "EngineReplayRunner",
     "EngineReplayRunnerFactory",
+    "EngineRequestSpec",
     "InvalidRunnerError",
     "ReplayOutputRequirements",
     "ReplayReport",
diff --git a/python/aisimulate/src/aisimulate/aic.py b/python/aisimulate/src/aisimulate/aic.py
index 9b1704790..e0111d743 100644
--- a/python/aisimulate/src/aisimulate/aic.py
+++ b/python/aisimulate/src/aisimulate/aic.py
@@ -93,6 +93,7 @@ def materialize_aic_num_gpu_blocks(raw: dict[str, Any]) -> dict[str, Any]:
         fmha_dtype=lowered.get("aic_fmha_dtype"),
         kv_cache_dtype=lowered.get("aic_kv_cache_dtype"),
         comm_dtype=lowered.get("aic_comm_dtype"),
+        nextn=int(lowered.get("aic_nextn") or 0),
         systems_path=lowered.get("systems_path"),
     )
     return lowered
@@ -120,14 +121,10 @@ def estimate_num_gpu_blocks(
     fmha_dtype: str | None = None,
     kv_cache_dtype: str | None = None,
     comm_dtype: str | None = None,
+    nextn: int = 0,
     systems_path: str | None = None,
 ) -> int:
-    """Estimate per-rank KV blocks using the replay-wide AIC contract.
-
-    NextN is intentionally absent. AIC currently can return negative KV
-    capacity for Eagle when speculative-decoding state is included. Timing
-    compilation still receives NextN; only capacity estimation omits it.
-    """
+    """Estimate per-rank KV blocks using the replay-wide AIC contract."""
 
     if backend_name not in DEFAULT_BACKEND_VERSIONS:
         supported = ", ".join(sorted(DEFAULT_BACKEND_VERSIONS))
@@ -188,6 +185,7 @@ def estimate_num_gpu_blocks(
             fmha_quant_mode=_quant_mode_name("fmha", fmha_dtype),
             kvcache_quant_mode=_quant_mode_name("kvcache", kv_cache_dtype),
             comm_quant_mode=_quant_mode_name("comm", comm_dtype),
+            nextn=nextn,
             systems_path=systems_path,
         )
     )
diff --git a/python/aisimulate/src/aisimulate/compiler.py b/python/aisimulate/src/aisimulate/compiler.py
index fe61ee512..5d57c528c 100644
--- a/python/aisimulate/src/aisimulate/compiler.py
+++ b/python/aisimulate/src/aisimulate/compiler.py
@@ -108,7 +108,7 @@ def _worker_performance_model_metadata(
             "attention_dp_size": parallel.attention_data,
             "moe_tp_size": parallel.moe_tensor if sharded_moe else None,
             "moe_ep_size": parallel.moe_expert if sharded_moe else None,
-            "nextn": None,
+            "nextn": engine.nextn or None,
         },
     }
 
@@ -120,7 +120,7 @@ def _worker_engine_args(
     parallel = worker.parallelism
     cache = worker.kv_cache
     capacity = cache.capacity
-    memory_fraction = capacity.memory_fraction
+    memory_fraction = engine.free_gpu_memory_fraction or capacity.memory_fraction
     if capacity.type == "default" and memory_fraction is None:
         memory_fraction = 0.88 if backend == "sglang" else 0.9
     block_size = cache.block_size
@@ -153,6 +153,32 @@ def _worker_engine_args(
             if isinstance(engine.context_length, int)
             else resolve_model_context_length(engine.model)
         )
+    chunked_prefill = bool(engine.enable_chunked_prefill and role != "decode")
+    if backend != "sglang" or chunked_prefill:
+        payload["enable_chunked_prefill"] = chunked_prefill
+    if engine.nextn:
+        payload["aic_nextn"] = engine.nextn
+        payload["aic_nextn_accepted"] = engine.nextn_accepted
+    for name in (
+        "enable_wideep",
+        "enable_eplb",
+        "wideep_num_slots",
+        "moe_backend",
+        "attention_backend",
+    ):
+        value = getattr(engine, name)
+        if value not in (None, False):
+            payload[f"aic_{name}"] = value
+    for name, argument in (
+        ("gemm_quant_mode", "aic_gemm_dtype"),
+        ("moe_quant_mode", "aic_moe_dtype"),
+        ("kvcache_quant_mode", "aic_kv_cache_dtype"),
+        ("fmha_quant_mode", "aic_fmha_dtype"),
+        ("comm_quant_mode", "aic_comm_dtype"),
+    ):
+        value = getattr(engine, name)
+        if value is not None:
+            payload[argument] = value
     if capacity.type == "fixed":
         assert capacity.blocks is not None
         payload["num_gpu_blocks"] = capacity.blocks
@@ -231,7 +257,11 @@ def _traffic(
         return workload, None
 
     if isinstance(source, SyntheticSource):
-        workload.update(isl=source.input_tokens, osl=source.output_tokens)
+        workload.update(
+            isl=source.input_tokens,
+            osl=source.output_tokens,
+            cached_prefix_tokens=source.cached_prefix_tokens,
+        )
         stop_count = stop.requests if stop is not None else None
         relative = stop.requests_per_load_unit if stop is not None else None
     else:
diff --git a/python/aisimulate/src/aisimulate/config/engine.py b/python/aisimulate/src/aisimulate/config/engine.py
index f665a18c9..ede946f17 100644
--- a/python/aisimulate/src/aisimulate/config/engine.py
+++ b/python/aisimulate/src/aisimulate/config/engine.py
@@ -12,8 +12,10 @@
 from .common import Choices, IntegerRange, NumericRange, StrictModel
 
 PositiveInt = Annotated[int, Field(strict=True, gt=0)]
+NonNegativeInt = Annotated[int, Field(strict=True, ge=0)]
 PositiveFloat = Annotated[float, Field(strict=True, gt=0, allow_inf_nan=False)]
 Fraction = Annotated[float, Field(strict=True, gt=0, le=1, allow_inf_nan=False)]
+NonNegativeFloat = Annotated[float, Field(strict=True, ge=0, allow_inf_nan=False)]
 EngineMode = Literal["aggregated", "disaggregated"]
 Backend = Literal["vllm", "sglang", "trtllm"]
 
@@ -111,6 +113,20 @@ class EnginePredictionConfig(StrictModel):
     backend: Backend = "vllm"
     backend_version: str | None = None
     context_length: PositiveInt | Literal["max"] = "max"
+    nextn: NonNegativeInt = Field(default=0, le=5)
+    nextn_accepted: NonNegativeFloat | None = None
+    enable_chunked_prefill: bool = False
+    enable_wideep: bool = False
+    enable_eplb: bool = False
+    wideep_num_slots: PositiveInt | None = None
+    moe_backend: Literal["deepep_moe", "megamoe"] | None = None
+    attention_backend: Literal["flashinfer", "fa3"] | None = None
+    gemm_quant_mode: str | None = None
+    moe_quant_mode: str | None = None
+    kvcache_quant_mode: str | None = None
+    fmha_quant_mode: str | None = None
+    comm_quant_mode: str | None = None
+    free_gpu_memory_fraction: Fraction | None = None
     workers: WorkersPredictionConfig
     kv_transfer: KvTransferConfig | None = None
 
@@ -121,6 +137,18 @@ def _validate_nonempty(cls, value: str) -> str:
             raise ValueError("value must be nonempty")
         return value
 
+    @field_validator(
+        "gemm_quant_mode",
+        "moe_quant_mode",
+        "kvcache_quant_mode",
+        "fmha_quant_mode",
+        "comm_quant_mode",
+        mode="before",
+    )
+    @classmethod
+    def _normalize_quant_mode(cls, value: Any) -> Any:
+        return _normalize_engine_control_name(value)
+
     @field_validator("hardware")
     @classmethod
     def _reject_auto_hardware(cls, value: str) -> str:
@@ -138,6 +166,7 @@ def _validate_roles(self) -> EnginePredictionConfig:
         if self.mode == "disaggregated" and self.backend == "trtllm":
             raise ValueError("TensorRT-LLM disaggregated mode is unsupported")
         _validate_backend_block_sizes(backends={self.backend}, modes={self.mode}, workers=self.workers)
+        _validate_engine_controls(self, backends={self.backend})
         return self
 
 
@@ -261,6 +290,20 @@ class EngineRecommendationConfig(StrictModel):
     backend: Backend | Choices[Backend] = Field(default_factory=lambda: Choices[Backend](choices=["vllm", "sglang"]))
     backend_version: str | None = None
     context_length: PositiveInt | Literal["max"] = "max"
+    nextn: NonNegativeInt = Field(default=0, le=5)
+    nextn_accepted: NonNegativeFloat | None = None
+    enable_chunked_prefill: bool = False
+    enable_wideep: bool = False
+    enable_eplb: bool = False
+    wideep_num_slots: PositiveInt | None = None
+    moe_backend: Literal["deepep_moe", "megamoe"] | None = None
+    attention_backend: Literal["flashinfer", "fa3"] | None = None
+    gemm_quant_mode: str | None = None
+    moe_quant_mode: str | None = None
+    kvcache_quant_mode: str | None = None
+    fmha_quant_mode: str | None = None
+    comm_quant_mode: str | None = None
+    free_gpu_memory_fraction: Fraction | None = None
     workers: WorkersRecommendationConfig
     kv_transfer: KvTransferConfig | None = None
 
@@ -271,6 +314,18 @@ def _validate_nonempty(cls, value: str) -> str:
             raise ValueError("value must be nonempty")
         return value
 
+    @field_validator(
+        "gemm_quant_mode",
+        "moe_quant_mode",
+        "kvcache_quant_mode",
+        "fmha_quant_mode",
+        "comm_quant_mode",
+        mode="before",
+    )
+    @classmethod
+    def _normalize_quant_mode(cls, value: Any) -> Any:
+        return _normalize_engine_control_name(value)
+
     @model_validator(mode="after")
     def _validate_roles(self) -> EngineRecommendationConfig:
         modes = set(self.mode.choices) if isinstance(self.mode, Choices) else {self.mode}
@@ -281,9 +336,49 @@ def _validate_roles(self) -> EngineRecommendationConfig:
         )
         backends = set(self.backend.choices) if isinstance(self.backend, Choices) else {self.backend}
         _validate_backend_block_sizes(backends=backends, modes=modes, workers=self.workers)
+        _validate_engine_controls(self, backends=backends)
         return self
 
 
+def _normalize_engine_control_name(value: Any) -> Any:
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError("engine control names must be non-empty strings")
+    return value.strip().lower()
+
+
+def _validate_engine_controls(engine, *, backends: set[str]) -> None:
+    if engine.nextn == 0:
+        if engine.nextn_accepted is not None:
+            raise ValueError("nextn_accepted requires nextn greater than zero")
+    elif engine.nextn_accepted is None:
+        raise ValueError(f"nextn={engine.nextn} requires explicit nextn_accepted")
+    elif engine.nextn_accepted > engine.nextn:
+        raise ValueError("nextn_accepted must be within [0, nextn]")
+    if (engine.moe_backend is not None or engine.attention_backend is not None) and backends != {"sglang"}:
+        raise ValueError("moe_backend and attention_backend require backend='sglang'")
+
+    from aiconfigurator_core.sdk.common import (
+        CommQuantMode,
+        FMHAQuantMode,
+        GEMMQuantMode,
+        KVCacheQuantMode,
+        MoEQuantMode,
+    )
+
+    for name, enum_type in (
+        ("gemm_quant_mode", GEMMQuantMode),
+        ("moe_quant_mode", MoEQuantMode),
+        ("kvcache_quant_mode", KVCacheQuantMode),
+        ("fmha_quant_mode", FMHAQuantMode),
+        ("comm_quant_mode", CommQuantMode),
+    ):
+        value = getattr(engine, name)
+        if value is not None and value not in enum_type.__members__:
+            raise ValueError(f"{name} has unsupported value {value!r}")
+
+
 def _validate_worker_roles(*, modes: set[str], workers, has_transfer: bool) -> None:
     if "aggregated" in modes and workers.aggregated is None:
         raise ValueError("aggregated mode requires workers.aggregated")
diff --git a/python/aisimulate/src/aisimulate/config/traffic.py b/python/aisimulate/src/aisimulate/config/traffic.py
index 5371cb731..9c776d0cb 100644
--- a/python/aisimulate/src/aisimulate/config/traffic.py
+++ b/python/aisimulate/src/aisimulate/config/traffic.py
@@ -22,6 +22,13 @@ class SyntheticSource(StrictModel):
     type: Literal["synthetic"] = "synthetic"
     input_tokens: PositiveInt = 1024
     output_tokens: PositiveInt = 128
+    cached_prefix_tokens: NonNegativeInt = 0
+
+    @model_validator(mode="after")
+    def _validate_cached_prefix(self) -> SyntheticSource:
+        if self.cached_prefix_tokens > self.input_tokens:
+            raise ValueError("cached_prefix_tokens cannot exceed input_tokens")
+        return self
 
 
 class SessionShape(StrictModel):
diff --git a/python/aisimulate/src/aisimulate/recommend.py b/python/aisimulate/src/aisimulate/recommend.py
index 1bcd1e848..fb0e72388 100644
--- a/python/aisimulate/src/aisimulate/recommend.py
+++ b/python/aisimulate/src/aisimulate/recommend.py
@@ -107,6 +107,24 @@ def recommendation_to_sweeper(
         "min_gpu_budget": optimization.constraints.min_candidate_gpus,
         "context_length": (resolve_model_context_length(model) if context == "max" else context),
     }
+    for public_name, internal_name in (
+        ("nextn", "aic_nextn"),
+        ("nextn_accepted", "nextn_accepted"),
+        ("enable_chunked_prefill", "enable_chunked_prefill"),
+        ("enable_wideep", "enable_wideep"),
+        ("enable_eplb", "enable_eplb"),
+        ("wideep_num_slots", "wideep_num_slots"),
+        ("moe_backend", "moe_backend"),
+        ("attention_backend", "attention_backend"),
+        ("gemm_quant_mode", "gemm_quant_mode"),
+        ("moe_quant_mode", "moe_quant_mode"),
+        ("kvcache_quant_mode", "kvcache_quant_mode"),
+        ("fmha_quant_mode", "fmha_quant_mode"),
+        ("comm_quant_mode", "comm_quant_mode"),
+        ("free_gpu_memory_fraction", "free_gpu_memory_fraction"),
+    ):
+        if public_name in engine:
+            search_space[internal_name] = engine[public_name]
     search_space.update(_role_search_space(workers, modes))
     transfer = engine.get("kv_transfer")
     if isinstance(transfer, dict):
@@ -497,7 +515,11 @@ def _recommendation_workload(raw: dict[str, Any] | None) -> dict[str, Any]:
             result["max_sim_time_ms"] = 1_000.0 * float(stop["max_virtual_time_seconds"])
         return result
     if source_type == "synthetic":
-        result.update(isl=source.get("input_tokens", 1024), osl=source.get("output_tokens", 128))
+        result.update(
+            isl=source.get("input_tokens", 1024),
+            osl=source.get("output_tokens", 128),
+            cached_prefix_tokens=source.get("cached_prefix_tokens", 0),
+        )
         count = stop.get("requests") if isinstance(stop, dict) else None
         ratio = stop.get("requests_per_load_unit") if isinstance(stop, dict) else None
     elif source_type == "synthetic-session":
@@ -614,6 +636,25 @@ def _candidate_prediction(
         "workers": {},
     }
     raw_engine = source.engine.model_dump(mode="python", exclude_none=True)
+    for public_name, internal_name in (
+        ("nextn", "aic_nextn"),
+        ("nextn_accepted", "nextn_accepted"),
+        ("enable_chunked_prefill", "enable_chunked_prefill"),
+        ("enable_wideep", "enable_wideep"),
+        ("enable_eplb", "enable_eplb"),
+        ("wideep_num_slots", "wideep_num_slots"),
+        ("moe_backend", "moe_backend"),
+        ("attention_backend", "attention_backend"),
+        ("gemm_quant_mode", "gemm_quant_mode"),
+        ("moe_quant_mode", "moe_quant_mode"),
+        ("kvcache_quant_mode", "kvcache_quant_mode"),
+        ("fmha_quant_mode", "fmha_quant_mode"),
+        ("comm_quant_mode", "comm_quant_mode"),
+        ("free_gpu_memory_fraction", "free_gpu_memory_fraction"),
+    ):
+        value = sample.get(internal_name)
+        if value is not None:
+            engine[public_name] = value
     roles = ("agg",) if deployment.deployment_mode == "agg" else ("prefill", "decode")
     for role in roles:
         prefix = "" if role == "agg" else f"{role}_"
diff --git a/python/aisimulate/src/aisimulate/runner.py b/python/aisimulate/src/aisimulate/runner.py
index d19681f83..91d0ec136 100644
--- a/python/aisimulate/src/aisimulate/runner.py
+++ b/python/aisimulate/src/aisimulate/runner.py
@@ -44,6 +44,7 @@
         "replay_concurrency",
         "isl",
         "osl",
+        "cached_prefix_tokens",
         "request_count",
         "turns_per_session",
         "shared_prefix_ratio",
@@ -70,6 +71,11 @@
     "kv_cache_dtype": ("kv_cache_dtype", "aic_kv_cache_dtype"),
     "comm_dtype": ("comm_dtype", "aic_comm_dtype"),
     "systems_path": ("systems_path",),
+    "enable_wideep": ("aic_enable_wideep",),
+    "enable_eplb": ("aic_enable_eplb",),
+    "wideep_num_slots": ("aic_wideep_num_slots",),
+    "moe_backend": ("aic_moe_backend",),
+    "attention_backend": ("aic_attention_backend",),
 }
 
 
@@ -495,6 +501,26 @@ def _materialize_requests(
         }
         for index in range(request_count)
     ]
+    cached_prefix_tokens = workload.get("cached_prefix_tokens", 0)
+    if (
+        not isinstance(cached_prefix_tokens, int)
+        or isinstance(cached_prefix_tokens, bool)
+        or cached_prefix_tokens < 0
+        or cached_prefix_tokens > min(input_lengths)
+    ):
+        raise ValueError(
+            "cached_prefix_tokens must be a non-negative integer no greater "
+            "than every synthetic input length"
+        )
+    if cached_prefix_tokens:
+        prefix = list(range(1, cached_prefix_tokens + 1))
+        for index, request in enumerate(requests):
+            suffix_length = input_lengths[index] - cached_prefix_tokens
+            suffix_seed = (index + 1) * 1_000_003 + cached_prefix_tokens
+            request["input_token_ids"] = prefix + [
+                (suffix_seed + offset) & 0xFFFF_FFFF
+                for offset in range(suffix_length)
+            ]
     return requests, concurrency
 
 
@@ -639,8 +665,11 @@ def _materialize_engine_role(
         if not configured:
             continue
         value = rank.pop(configured[0])
-        if target in {"pp", "moe_tp_size", "moe_ep_size"}:
+        if target in {"pp", "moe_tp_size", "moe_ep_size", "wideep_num_slots"}:
             value = _positive_int(value, f"engine provider {role} {target}")
+        elif target in {"enable_wideep", "enable_eplb"}:
+            if not isinstance(value, bool):
+                raise ValueError(f"engine provider {role} {target} must be a boolean")
         elif not isinstance(value, str) or not value:
             raise ValueError(f"engine provider {role} {target} must be a string")
         aic_timing_overrides[target] = value
@@ -712,6 +741,25 @@ def _materialize_engine_role(
             )
         rank["aic_nextn_accept_rates"] = accept_rates
 
+    nextn_accepted = _pop_alias(
+        rank,
+        "aic_nextn_accepted",
+        ("aic_nextn_accepted", "nextn_accepted"),
+    )
+    if nextn_accepted is not None:
+        if nextn is None:
+            raise ValueError(
+                f"engine provider {role} aic_nextn_accepted requires aic_nextn"
+            )
+        if accept_rates is not None:
+            raise ValueError(
+                f"engine provider {role} cannot set both aic_nextn_accepted and "
+                "aic_nextn_accept_rates"
+            )
+        rank["aic_nextn_accept_rates"] = _accept_rates_for_expected(
+            nextn, nextn_accepted, role=role
+        )
+
     mtp_seed = _pop_alias(rank, "aic_mtp_seed", ("aic_mtp_seed", "mtp_seed"))
     if mtp_seed is not None:
         if (
@@ -810,6 +858,31 @@ def _positive_int(value: JSONValue, name: str) -> int:
     return value
 
 
+def _accept_rates_for_expected(
+    nextn: int, value: JSONValue, *, role: str
+) -> str:
+    """Lower an explicit expected accepted-token count to conditional rates."""
+
+    if (
+        not isinstance(value, (int, float))
+        or isinstance(value, bool)
+        or not math.isfinite(value)
+        or not 0.0 <= float(value) <= nextn
+    ):
+        raise ValueError(
+            f"engine provider {role} aic_nextn_accepted must be finite and "
+            f"within [0, {nextn}]"
+        )
+    expected = float(value)
+    whole = int(expected)
+    fraction = expected - whole
+    rates = [1.0] * whole
+    if len(rates) < nextn:
+        rates.append(fraction)
+    rates.extend([0.0] * (nextn - len(rates)))
+    return ",".join(format(rate, ".17g") for rate in rates)
+
+
 def _random_range_ratio(value: JSONValue) -> float:
     if (
         not isinstance(value, (int, float))
diff --git a/python/aisimulate/src/aisimulate/sweeper/__init__.py b/python/aisimulate/src/aisimulate/sweeper/__init__.py
index 4d777fd2b..9e9098198 100644
--- a/python/aisimulate/src/aisimulate/sweeper/__init__.py
+++ b/python/aisimulate/src/aisimulate/sweeper/__init__.py
@@ -45,6 +45,7 @@
 from .replay import (
     REPLAY_SPEC_API_VERSION,
     BackendDeploymentSpec,
+    EngineRequestSpec,
     HookCapability,
     ReplayOutputRequirements,
     ReplayReport,
@@ -113,6 +114,7 @@ def __getattr__(name: str) -> Any:
     "CandidateContext",
     "ConditionalSearchSpace",
     "DisaggParallelConfig",
+    "EngineRequestSpec",
     "HookCapability",
     "InfeasibleCandidate",
     "ModelHardware",
diff --git a/python/aisimulate/src/aisimulate/sweeper/config.py b/python/aisimulate/src/aisimulate/sweeper/config.py
index 2431e649e..1ddac93bb 100644
--- a/python/aisimulate/src/aisimulate/sweeper/config.py
+++ b/python/aisimulate/src/aisimulate/sweeper/config.py
@@ -194,6 +194,7 @@ class Workload(BaseModel):
     num_request_ratio: float | None = None  # request count multiplier for concrete concurrency or request_rate
     random_range_ratio: float = 1.0
     random_seed: int = 0
+    cached_prefix_tokens: int = 0  # exact already-cached prefix length per request
     shared_prefix_ratio: float = 0.0  # cache-locality / prefix sharing
     num_prefix_groups: int = 0
     turns_per_session: int = 1  # multi-turn sessions
@@ -234,6 +235,15 @@ def _validate_random_seed_type(cls, value: Any) -> Any:
             raise ValueError(f"random_seed must be an unsigned 64-bit integer, got {value!r}")
         return value
 
+    @field_validator("cached_prefix_tokens", mode="before")
+    @classmethod
+    def _validate_cached_prefix_tokens_type(cls, value: Any) -> Any:
+        if isinstance(value, bool) or not isinstance(value, int):
+            raise ValueError(
+                f"cached_prefix_tokens must be a non-negative integer, got {value!r}"
+            )
+        return value
+
     @property
     def is_trace_based(self) -> bool:
         return self.trace_path is not None
@@ -304,6 +314,8 @@ def _validate_workload(self) -> Workload:
                 set_syn.append("random_range_ratio")
             if self.random_seed != 0:
                 set_syn.append("random_seed")
+            if self.cached_prefix_tokens != 0:
+                set_syn.append("cached_prefix_tokens")
             if set_syn:
                 raise ValueError(f"trace workload (trace_path set) must not set synthetic fields {set_syn}")
             if self.replay_concurrency is not None and self.replay_concurrency <= 0:
@@ -355,6 +367,12 @@ def _validate_workload(self) -> Workload:
                 raise ValueError(
                     f"random_range_ratio={self.random_range_ratio} gives a zero-token lower bound for {name}={length}"
                 )
+        minimum_isl = int((self.isl or 0) * self.random_range_ratio)
+        if not 0 <= self.cached_prefix_tokens <= minimum_isl:
+            raise ValueError(
+                "cached_prefix_tokens must be within the shortest synthetic input "
+                f"length [0, {minimum_isl}], got {self.cached_prefix_tokens}"
+            )
         if isinstance(self.random_seed, bool) or self.random_seed < 0 or self.random_seed > 0xFFFF_FFFF_FFFF_FFFF:
             raise ValueError(f"random_seed must be an unsigned 64-bit integer, got {self.random_seed!r}")
         if self.random_range_ratio != 1.0 and self.turns_per_session != 1:
@@ -401,8 +419,25 @@ class SearchSpace(BaseModel):
     gpu_budget: int = 32  # max GPUs per candidate
     min_gpu_budget: int | None = None
     context_length: int | None = None
+    max_seq_len: int | None = Field(default=None, gt=0)
     startup_time: float | None = None
-    aic_nextn: int | None = None  # speculative-decode (MTP) depth, 1..5
+    aic_nextn: int | None = Field(default=None, ge=0, le=5)
+    nextn_accepted: float | None = None
+
+    # shared engine/request-modeling controls. Role-specific P/D overrides are
+    # intentionally a separate capability; these values apply to every active role.
+    enable_chunked_prefill: bool = False
+    enable_wideep: bool = False
+    enable_eplb: bool = False
+    wideep_num_slots: int | None = Field(default=None, gt=0)
+    moe_backend: str | None = None
+    attention_backend: str | None = None
+    gemm_quant_mode: str | None = None
+    moe_quant_mode: str | None = None
+    kvcache_quant_mode: str | None = None
+    fmha_quant_mode: str | None = None
+    comm_quant_mode: str | None = None
+    free_gpu_memory_fraction: float | None = Field(default=None, gt=0, le=1)
 
     # prefill engine (disagg branch): scheduler batching capacity
     prefill_max_num_batched_tokens: list[int] = [8192, 16384, 32768]
@@ -444,6 +479,24 @@ class SearchSpace(BaseModel):
     engine_log_discrete: list[str] = Field(default_factory=list)
     engine_integer_log_ranges: dict[str, list[int]] = Field(default_factory=dict)
 
+    @field_validator(
+        "moe_backend",
+        "attention_backend",
+        "gemm_quant_mode",
+        "moe_quant_mode",
+        "kvcache_quant_mode",
+        "fmha_quant_mode",
+        "comm_quant_mode",
+        mode="before",
+    )
+    @classmethod
+    def _normalize_engine_control_name(cls, value: Any) -> Any:
+        if value is None:
+            return None
+        if not isinstance(value, str) or not value.strip():
+            raise ValueError("engine control names must be non-empty strings")
+        return value.strip().lower()
+
     @model_validator(mode="after")
     def _validate_search_choices(self) -> SearchSpace:
         """Every backend dimension is a non-empty subset of its allowed choices."""
@@ -479,6 +532,47 @@ def _validate_gpu_budget(self) -> SearchSpace:
             )
         return self
 
+    @model_validator(mode="after")
+    def _validate_engine_controls(self) -> SearchSpace:
+        if (
+            self.context_length is not None
+            and self.max_seq_len is not None
+            and self.context_length != self.max_seq_len
+        ):
+            raise ValueError(
+                "context_length and max_seq_len describe the same sequence-capacity "
+                "limit and must match when both are set"
+            )
+        if not self.aic_nextn:
+            if self.nextn_accepted is not None:
+                raise ValueError(
+                    "nextn_accepted is only valid when aic_nextn is greater than zero"
+                )
+        elif self.nextn_accepted is None:
+            raise ValueError(
+                f"aic_nextn={self.aic_nextn} requires explicit nextn_accepted; "
+                "acceptance is never inferred"
+            )
+        elif (
+            not math.isfinite(self.nextn_accepted)
+            or not 0 <= self.nextn_accepted <= self.aic_nextn
+        ):
+            raise ValueError(
+                "nextn_accepted must be finite and within "
+                f"[0, aic_nextn={self.aic_nextn}], got {self.nextn_accepted!r}"
+            )
+        if self.attention_backend not in (None, "flashinfer", "fa3"):
+            raise ValueError(
+                "attention_backend must be 'flashinfer' or 'fa3', got "
+                f"{self.attention_backend!r}"
+            )
+        if self.moe_backend not in (None, "deepep_moe", "megamoe"):
+            raise ValueError(
+                "moe_backend must be 'deepep_moe' or 'megamoe', got "
+                f"{self.moe_backend!r}"
+            )
+        return self
+
     @model_validator(mode="after")
     def _validate_parallel_configs(self) -> SearchSpace:
         """A pinned ``parallel_configs`` (non-empty) must match a single deployment
diff --git a/python/aisimulate/src/aisimulate/sweeper/deploy.py b/python/aisimulate/src/aisimulate/sweeper/deploy.py
index 6d808de95..b77be973a 100644
--- a/python/aisimulate/src/aisimulate/sweeper/deploy.py
+++ b/python/aisimulate/src/aisimulate/sweeper/deploy.py
@@ -8,7 +8,7 @@
 from typing import Any
 
 from ..aic import estimate_kv_bytes_per_token, materialize_aic_num_gpu_blocks
-from .replay import BackendDeploymentSpec
+from .replay import BackendDeploymentSpec, EngineRequestSpec
 
 
 def _role_prefix(role: str) -> str:
@@ -35,7 +35,13 @@ def _performance_model_metadata(sample: dict[str, Any], role: str, *, backend_ve
     return {"provider": "aic", "config": config}
 
 
-def _engine_args_payload(sample: dict[str, Any], role: str, *, backend_version: str) -> dict[str, Any]:
+def _engine_args_payload(
+    sample: dict[str, Any],
+    role: str,
+    *,
+    backend_version: str,
+    engine_request: EngineRequestSpec | None = None,
+) -> dict[str, Any]:
     """Build the runner-neutral engine argument payload for one role."""
     prefix = _role_prefix(role)
     tp = int(sample[f"{prefix}tp"])
@@ -66,7 +72,11 @@ def _engine_args_payload(sample: dict[str, Any], role: str, *, backend_version:
         "max_num_batched_tokens": int(sample[f"{role}_max_num_batched_tokens"]),
         "max_num_seqs": int(sample[f"{role}_max_num_seqs"]),
         "block_size": int(block_size),
-        memory_fraction_field: float(memory_fraction),
+        memory_fraction_field: float(
+            engine_request.memory_fraction_by_role[role]
+            if engine_request is not None
+            else memory_fraction
+        ),
         "enable_prefix_caching": bool(sample[f"{role}_enable_prefix_caching"]),
     }
     if backend == "vllm" and sample.get("context_length") is not None:
@@ -74,8 +84,36 @@ def _engine_args_payload(sample: dict[str, Any], role: str, *, backend_version:
     if moe_tp * moe_ep > 1:
         payload["aic_moe_tp_size"] = moe_tp
         payload["aic_moe_ep_size"] = moe_ep
-    if sample.get("aic_nextn") is not None:
+    if sample.get("aic_nextn"):
         payload["aic_nextn"] = int(sample["aic_nextn"])
+    if engine_request is not None:
+        if backend == "vllm":
+            payload["max_model_len"] = engine_request.max_seq_len
+        chunked_prefill = bool(engine_request.enable_chunked_prefill and role != "decode")
+        if backend != "sglang" or chunked_prefill:
+            payload["enable_chunked_prefill"] = chunked_prefill
+        if engine_request.nextn_accepted is not None:
+            payload["aic_nextn_accepted"] = engine_request.nextn_accepted
+        for name in (
+            "enable_wideep",
+            "enable_eplb",
+            "wideep_num_slots",
+            "moe_backend",
+            "attention_backend",
+        ):
+            value = getattr(engine_request, name)
+            if value not in (None, False):
+                payload[f"aic_{name}"] = value
+        for name, argument in (
+            ("gemm_quant_mode", "aic_gemm_dtype"),
+            ("moe_quant_mode", "aic_moe_dtype"),
+            ("kvcache_quant_mode", "aic_kv_cache_dtype"),
+            ("fmha_quant_mode", "aic_fmha_dtype"),
+            ("comm_quant_mode", "aic_comm_dtype"),
+        ):
+            value = getattr(engine_request, name)
+            if value is not None:
+                payload[argument] = value
     startup = sample.get(f"{role}_startup_time")
     if startup is None:
         startup = sample.get("startup_time")
@@ -117,13 +155,19 @@ def _engine_args_payload(sample: dict[str, Any], role: str, *, backend_version:
     return payload
 
 
-def build_backend_deployment(sample: dict[str, Any], *, backend_version: str) -> BackendDeploymentSpec:
+def build_backend_deployment(
+    sample: dict[str, Any],
+    *,
+    backend_version: str,
+    engine_request: EngineRequestSpec | None = None,
+) -> BackendDeploymentSpec:
     """Build the Dynamo-independent backend part of a :class:`ReplaySpec`."""
     mode = sample["deployment_mode"]
     common = {
         "deployment_mode": mode,
         "backend": sample["backend"],
         "backend_version": backend_version,
+        "engine_request": engine_request,
         "parallel_config": {
             key: value
             for key, value in sample.items()
@@ -161,12 +205,27 @@ def build_backend_deployment(sample: dict[str, Any], *, backend_version: str) ->
     }
     if mode == "agg":
         return BackendDeploymentSpec(
-            agg_engine_args=_engine_args_payload(sample, "agg", backend_version=backend_version),
+            agg_engine_args=_engine_args_payload(
+                sample,
+                "agg",
+                backend_version=backend_version,
+                engine_request=engine_request,
+            ),
             num_workers=int(sample["replicas"]),
             **common,
         )
-    prefill_args = _engine_args_payload(sample, "prefill", backend_version=backend_version)
-    decode_args = _engine_args_payload(sample, "decode", backend_version=backend_version)
+    prefill_args = _engine_args_payload(
+        sample,
+        "prefill",
+        backend_version=backend_version,
+        engine_request=engine_request,
+    )
+    decode_args = _engine_args_payload(
+        sample,
+        "decode",
+        backend_version=backend_version,
+        engine_request=engine_request,
+    )
     if sample.get("kv_transfer_bytes_per_token") == "auto":
         resolved = max(
             int(prefill_args["kv_bytes_per_token"]),
diff --git a/python/aisimulate/src/aisimulate/sweeper/engine_request.py b/python/aisimulate/src/aisimulate/sweeper/engine_request.py
new file mode 100644
index 000000000..c9ea7a038
--- /dev/null
+++ b/python/aisimulate/src/aisimulate/sweeper/engine_request.py
@@ -0,0 +1,197 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Resolve legacy engine/request controls before the search reaches replay."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from aiconfigurator_core.sdk.common import (
+    CommQuantMode,
+    FMHAQuantMode,
+    GEMMQuantMode,
+    KVCacheQuantMode,
+    MoEQuantMode,
+)
+from aiconfigurator_core.sdk.models import get_model_family
+from aiconfigurator_core.sdk.perf_database import is_blackwell_system
+
+from .config import SmartSearchConfig, Workload
+from .kv_estimate import memory_fraction_kind
+from .model_hw import resolve_model_hardware
+from .replay import EngineRequestSpec
+
+_DEEPSEEK_V4_MEGAMOE_MODELS = frozenset(
+    {
+        "deepseek-ai/DeepSeek-V4-Pro",
+        "sgl-project/DeepSeek-V4-Pro-FP8",
+    }
+)
+
+
+@dataclass(frozen=True)
+class EngineControlTemplate:
+    """Backend-resolved controls which do not depend on a sampled shape."""
+
+    backend: str
+    max_seq_len: int
+    model_family: str
+    is_moe: bool
+    memory_fraction_kind: str
+
+
+def _validate_quant_mode(name: str, value: str | None, enum_type: type) -> None:
+    if value is not None and value not in enum_type.__members__:
+        allowed = ", ".join(enum_type.__members__)
+        raise ValueError(f"{name} has unsupported value {value!r}; allowed: {allowed}")
+
+
+def _required_workload_tokens(workload: Workload) -> int | None:
+    if workload.is_trace_based:
+        return None
+    assert workload.isl is not None and workload.osl is not None
+    return workload.isl + workload.osl
+
+
+def resolve_engine_controls(
+    config: SmartSearchConfig,
+) -> dict[str, EngineControlTemplate]:
+    """Validate selected model/backend combinations and resolve sequence capacity.
+
+    This runs before provider preparation and before any replay work. Controls that
+    cannot affect the selected model/backend are rejected instead of being silently
+    accepted and lost during materialization.
+    """
+
+    ss = config.search_space
+    _validate_quant_mode("gemm_quant_mode", ss.gemm_quant_mode, GEMMQuantMode)
+    _validate_quant_mode("moe_quant_mode", ss.moe_quant_mode, MoEQuantMode)
+    _validate_quant_mode("kvcache_quant_mode", ss.kvcache_quant_mode, KVCacheQuantMode)
+    _validate_quant_mode("fmha_quant_mode", ss.fmha_quant_mode, FMHAQuantMode)
+    _validate_quant_mode("comm_quant_mode", ss.comm_quant_mode, CommQuantMode)
+
+    model_family = get_model_family(ss.model_name)
+    required_tokens = _required_workload_tokens(config.workload)
+    resolved: dict[str, EngineControlTemplate] = {}
+    for backend in dict.fromkeys(ss.backend):
+        model_hw = resolve_model_hardware(
+            ss.model_name, ss.hardware_sku, backend=backend
+        )
+        max_seq_len = ss.max_seq_len or ss.context_length or model_hw.max_context
+        if max_seq_len is None:
+            raise ValueError(
+                f"max_seq_len is required because model {ss.model_name!r} exposes "
+                "no maximum context length"
+            )
+        if required_tokens is not None and max_seq_len < required_tokens:
+            raise ValueError(
+                f"max_seq_len={max_seq_len} cannot hold the synthetic request "
+                f"isl+osl={required_tokens}"
+            )
+        if model_hw.max_context is not None and max_seq_len > model_hw.max_context:
+            raise ValueError(
+                f"max_seq_len={max_seq_len} exceeds model {ss.model_name!r} "
+                f"maximum context {model_hw.max_context}"
+            )
+
+        moe_controls = (
+            ss.enable_wideep,
+            ss.enable_eplb,
+            ss.wideep_num_slots is not None,
+            ss.moe_backend is not None,
+            ss.moe_quant_mode is not None,
+        )
+        if any(moe_controls) and not model_hw.is_moe:
+            raise ValueError(
+                f"MoE controls require an MoE model; {ss.model_name!r} is dense"
+            )
+        if ss.attention_backend is not None:
+            if not model_hw.mla:
+                raise ValueError(
+                    "attention_backend is only supported for MLA models; "
+                    f"{ss.model_name!r} is not MLA"
+                )
+            if backend != "sglang":
+                raise ValueError(
+                    "attention_backend is supported only with backend='sglang'; "
+                    f"got {backend!r}"
+                )
+        if ss.moe_backend is not None and backend != "sglang":
+            raise ValueError(
+                f"moe_backend={ss.moe_backend!r} is supported only with "
+                f"backend='sglang'; got {backend!r}"
+            )
+        if ss.moe_backend == "megamoe":
+            if model_family != "DEEPSEEKV4":
+                raise ValueError(
+                    "moe_backend='megamoe' is supported only for DeepSeek-V4 models"
+                )
+            if ss.model_name not in _DEEPSEEK_V4_MEGAMOE_MODELS:
+                raise ValueError(
+                    "moe_backend='megamoe' has packaged performance data only for "
+                    f"DeepSeek-V4-Pro; got {ss.model_name!r}"
+                )
+            if not is_blackwell_system(ss.hardware_sku):
+                raise ValueError(
+                    "moe_backend='megamoe' requires a Blackwell-class system; "
+                    f"got {ss.hardware_sku!r}"
+                )
+
+        resolved[backend] = EngineControlTemplate(
+            backend=backend,
+            max_seq_len=max_seq_len,
+            model_family=model_family,
+            is_moe=model_hw.is_moe,
+            memory_fraction_kind=memory_fraction_kind(backend),
+        )
+    return resolved
+
+
+def materialize_engine_request(
+    template: EngineControlTemplate,
+    *,
+    config: SmartSearchConfig,
+    sample: dict[str, object],
+) -> EngineRequestSpec:
+    """Bind backend-resolved controls to one candidate's active roles."""
+
+    ss = config.search_space
+    roles = (
+        ("agg",)
+        if sample["deployment_mode"] == "agg"
+        else ("prefill", "decode")
+    )
+    memory_by_role: dict[str, float] = {}
+    for role in roles:
+        value = (
+            ss.free_gpu_memory_fraction
+            if ss.free_gpu_memory_fraction is not None
+            else sample[f"{role}_gpu_memory_utilization"]
+        )
+        if value is None:
+            value = 0.88 if template.backend == "sglang" else 0.9
+        memory_by_role[role] = float(value)
+    return EngineRequestSpec(
+        cached_prefix_tokens=config.workload.cached_prefix_tokens,
+        context_tokens={
+            role: int(sample[f"{role}_max_num_batched_tokens"]) for role in roles
+        },
+        enable_chunked_prefill=ss.enable_chunked_prefill,
+        enable_wideep=ss.enable_wideep,
+        enable_eplb=ss.enable_eplb,
+        wideep_num_slots=ss.wideep_num_slots,
+        moe_backend=ss.moe_backend,
+        attention_backend=ss.attention_backend,
+        gemm_quant_mode=ss.gemm_quant_mode,
+        moe_quant_mode=ss.moe_quant_mode,
+        kvcache_quant_mode=ss.kvcache_quant_mode,
+        fmha_quant_mode=ss.fmha_quant_mode,
+        comm_quant_mode=ss.comm_quant_mode,
+        nextn=ss.aic_nextn or 0,
+        nextn_accepted=ss.nextn_accepted,
+        memory_fraction_kind=template.memory_fraction_kind,
+        memory_fraction_by_role=memory_by_role,
+        max_seq_len=template.max_seq_len,
+        model_family=template.model_family,
+    )
diff --git a/python/aisimulate/src/aisimulate/sweeper/kv_estimate.py b/python/aisimulate/src/aisimulate/sweeper/kv_estimate.py
index 878d6a407..ceba85a75 100644
--- a/python/aisimulate/src/aisimulate/sweeper/kv_estimate.py
+++ b/python/aisimulate/src/aisimulate/sweeper/kv_estimate.py
@@ -68,6 +68,11 @@ def estimate_kv_tokens(
     max_batch_size: int = DEFAULT_MAX_BATCH_SIZE,
     memory_fraction: float = DEFAULT_MEMORY_FRACTION,
     nextn: int = 0,
+    gemm_quant_mode: str | None = None,
+    moe_quant_mode: str | None = None,
+    kvcache_quant_mode: str | None = None,
+    fmha_quant_mode: str | None = None,
+    comm_quant_mode: str | None = None,
 ) -> int | None:
     """Per-rank KV-cache capacity (in tokens) for ``shape``, or ``None`` when the
     shape leaves no KV budget (weights + activations already fill VRAM -> OOM).
@@ -90,6 +95,11 @@ def estimate_kv_tokens(
             moe_tp_size=shape.moe_tp,
             moe_ep_size=shape.moe_ep,
             nextn=nextn,
+            gemm_quant_mode=gemm_quant_mode,
+            moe_quant_mode=moe_quant_mode,
+            kvcache_quant_mode=kvcache_quant_mode,
+            fmha_quant_mode=fmha_quant_mode,
+            comm_quant_mode=comm_quant_mode,
             allow_naive_fallback=False,
         )
     except ValueError as exc:
@@ -117,6 +127,12 @@ def feasible_shape_tokens(
     max_num_tokens: int = DEFAULT_MAX_NUM_TOKENS,
     max_batch_size: int = DEFAULT_MAX_BATCH_SIZE,
     memory_fraction: float = DEFAULT_MEMORY_FRACTION,
+    nextn: int = 0,
+    gemm_quant_mode: str | None = None,
+    moe_quant_mode: str | None = None,
+    kvcache_quant_mode: str | None = None,
+    fmha_quant_mode: str | None = None,
+    comm_quant_mode: str | None = None,
 ) -> dict[ParallelShape, int]:
     """Map each *feasible* shape to its KV-cache token capacity.
 
@@ -137,6 +153,12 @@ def feasible_shape_tokens(
             max_num_tokens=max_num_tokens,
             max_batch_size=max_batch_size,
             memory_fraction=memory_fraction,
+            nextn=nextn,
+            gemm_quant_mode=gemm_quant_mode,
+            moe_quant_mode=moe_quant_mode,
+            kvcache_quant_mode=kvcache_quant_mode,
+            fmha_quant_mode=fmha_quant_mode,
+            comm_quant_mode=comm_quant_mode,
         )
         if tokens is not None and tokens > max_seq_len:
             feasible[shape] = tokens
diff --git a/python/aisimulate/src/aisimulate/sweeper/kv_load.py b/python/aisimulate/src/aisimulate/sweeper/kv_load.py
index 508bffc6e..56742ec4c 100644
--- a/python/aisimulate/src/aisimulate/sweeper/kv_load.py
+++ b/python/aisimulate/src/aisimulate/sweeper/kv_load.py
@@ -40,6 +40,11 @@ def _per_rank_capacity_tokens(
     max_batch_size: int,
     memory_fraction: float,
     nextn: int,
+    gemm_quant_mode: str | None,
+    moe_quant_mode: str | None,
+    kvcache_quant_mode: str | None,
+    fmha_quant_mode: str | None,
+    comm_quant_mode: str | None,
 ) -> int:
     tokens = estimate_kv_tokens(
         shape,
@@ -51,6 +56,11 @@ def _per_rank_capacity_tokens(
         max_batch_size=max_batch_size,
         memory_fraction=memory_fraction,
         nextn=nextn,
+        gemm_quant_mode=gemm_quant_mode,
+        moe_quant_mode=moe_quant_mode,
+        kvcache_quant_mode=kvcache_quant_mode,
+        fmha_quant_mode=fmha_quant_mode,
+        comm_quant_mode=comm_quant_mode,
     )
     if tokens is None:
         raise InfeasibleKVCapacity(
@@ -75,6 +85,13 @@ def _role_capacity_tokens(
     if fixed_blocks is not None:
         per_rank_tokens = int(fixed_blocks) * block_size
     else:
+        memory_fraction = (
+            sample.get("free_gpu_memory_fraction")
+            if sample.get("free_gpu_memory_fraction") is not None
+            else sample[f"{role}_gpu_memory_utilization"]
+        )
+        if memory_fraction is None:
+            memory_fraction = 0.88 if sample["backend"] == "sglang" else 0.9
         per_rank_tokens = _per_rank_capacity_tokens(
             config.shape,
             model_name=str(sample["model_name"]),
@@ -83,8 +100,13 @@ def _role_capacity_tokens(
             backend_version=backend_version,
             max_num_tokens=int(sample[f"{role}_max_num_batched_tokens"]),
             max_batch_size=int(sample[f"{role}_max_num_seqs"]),
-            memory_fraction=float(sample[f"{role}_gpu_memory_utilization"]),
+            memory_fraction=float(memory_fraction),
             nextn=int(sample.get("aic_nextn") or 0),
+            gemm_quant_mode=sample.get("gemm_quant_mode"),
+            moe_quant_mode=sample.get("moe_quant_mode"),
+            kvcache_quant_mode=sample.get("kvcache_quant_mode"),
+            fmha_quant_mode=sample.get("fmha_quant_mode"),
+            comm_quant_mode=sample.get("comm_quant_mode"),
         )
     # Dynamo's AIC estimator returns per-rank blocks. Offline replay models one
     # engine-wide KV pool, so attention-DP ranks contribute independent capacity;
diff --git a/python/aisimulate/src/aisimulate/sweeper/model_hw.py b/python/aisimulate/src/aisimulate/sweeper/model_hw.py
index ee0bdfc65..bb90b96ee 100644
--- a/python/aisimulate/src/aisimulate/sweeper/model_hw.py
+++ b/python/aisimulate/src/aisimulate/sweeper/model_hw.py
@@ -113,6 +113,14 @@ def parallel_configs_for(
     max_num_tokens: int = DEFAULT_MAX_NUM_TOKENS,
     max_batch_size: int = DEFAULT_MAX_BATCH_SIZE,
     memory_fraction: float = DEFAULT_MEMORY_FRACTION,
+    enable_wideep: bool = False,
+    moe_backend: str | None = None,
+    gemm_quant_mode: str | None = None,
+    moe_quant_mode: str | None = None,
+    kvcache_quant_mode: str | None = None,
+    fmha_quant_mode: str | None = None,
+    comm_quant_mode: str | None = None,
+    nextn: int = 0,
     role_runtime: dict[str, tuple[int, int, float] | tuple[int, int, float, int | None]] | None = None,
 ) -> list[ReplicaParallelConfig] | list[DisaggParallelConfig]:
     """Resolve the model/hardware, then enumerate the parallel configs that fit
@@ -147,7 +155,8 @@ def parallel_configs_for(
         backend=backend,
         gpu_budget=gpu_budget,
         min_gpu_budget=min_gpu_budget,
-        enable_wideep=mh.enable_wideep,
+        enable_wideep=mh.enable_wideep or enable_wideep,
+        moe_backend=moe_backend,
         allow_moe_pure_tp=True,
     )
     if deployment_mode == "disagg":
@@ -181,6 +190,12 @@ def feasible_for(role: str, shapes):
             max_num_tokens=role_tokens,
             max_batch_size=role_batch,
             memory_fraction=role_memory,
+            gemm_quant_mode=gemm_quant_mode,
+            moe_quant_mode=moe_quant_mode,
+            kvcache_quant_mode=kvcache_quant_mode,
+            fmha_quant_mode=fmha_quant_mode,
+            comm_quant_mode=comm_quant_mode,
+            nextn=nextn,
         )
 
     if deployment_mode == "agg":
diff --git a/python/aisimulate/src/aisimulate/sweeper/replay.py b/python/aisimulate/src/aisimulate/sweeper/replay.py
index 37970d4cc..e2889b08f 100644
--- a/python/aisimulate/src/aisimulate/sweeper/replay.py
+++ b/python/aisimulate/src/aisimulate/sweeper/replay.py
@@ -19,6 +19,31 @@
 REPLAY_SPEC_API_VERSION = 1
 
 
+@dataclass(frozen=True)
+class EngineRequestSpec:
+    """Resolved legacy engine/request controls evaluated for one candidate."""
+
+    cached_prefix_tokens: int = 0
+    context_tokens: dict[str, int] = field(default_factory=dict)
+    enable_chunked_prefill: bool = False
+    enable_wideep: bool = False
+    enable_eplb: bool = False
+    wideep_num_slots: int | None = None
+    moe_backend: str | None = None
+    attention_backend: str | None = None
+    gemm_quant_mode: str | None = None
+    moe_quant_mode: str | None = None
+    kvcache_quant_mode: str | None = None
+    fmha_quant_mode: str | None = None
+    comm_quant_mode: str | None = None
+    nextn: int = 0
+    nextn_accepted: float | None = None
+    memory_fraction_kind: str = "of_total"
+    memory_fraction_by_role: dict[str, float] = field(default_factory=dict)
+    max_seq_len: int = 0
+    model_family: str = ""
+
+
 @dataclass(frozen=True)
 class BackendDeploymentSpec:
     """Concrete backend engines and fleet shape for one candidate."""
@@ -33,6 +58,7 @@ class BackendDeploymentSpec:
     num_workers: int = 0
     num_prefill_workers: int = 0
     num_decode_workers: int = 0
+    engine_request: EngineRequestSpec | None = None
     performance_model_metadata: dict[str, JSONValue] = field(default_factory=dict)
 
 
diff --git a/python/aisimulate/src/aisimulate/sweeper/sample.py b/python/aisimulate/src/aisimulate/sweeper/sample.py
index d75c9dcfd..93fe6d42a 100644
--- a/python/aisimulate/src/aisimulate/sweeper/sample.py
+++ b/python/aisimulate/src/aisimulate/sweeper/sample.py
@@ -17,8 +17,22 @@
     "gpu_budget",
     "min_gpu_budget",
     "context_length",
+    "max_seq_len",
     "startup_time",
     "aic_nextn",
+    "nextn_accepted",
+    "enable_chunked_prefill",
+    "enable_wideep",
+    "enable_eplb",
+    "wideep_num_slots",
+    "moe_backend",
+    "attention_backend",
+    "gemm_quant_mode",
+    "moe_quant_mode",
+    "kvcache_quant_mode",
+    "fmha_quant_mode",
+    "comm_quant_mode",
+    "free_gpu_memory_fraction",
 )
 
 # engine knobs per branch: searched batching + pinned scalars.
diff --git a/python/aisimulate/src/aisimulate/sweeper/search.py b/python/aisimulate/src/aisimulate/sweeper/search.py
index d054989dc..11bb2897f 100644
--- a/python/aisimulate/src/aisimulate/sweeper/search.py
+++ b/python/aisimulate/src/aisimulate/sweeper/search.py
@@ -44,6 +44,11 @@
 from .config import Candidate, OptimizationGoal, OptimizationTarget, SmartSearchConfig
 from .deploy import build_backend_deployment
 from .discovery import resolve_providers
+from .engine_request import (
+    EngineControlTemplate,
+    materialize_engine_request,
+    resolve_engine_controls,
+)
 from .kv_estimate import resolve_backend_version
 from .kv_load import InfeasibleKVCapacity, resolve_kv_load
 from .provider import (
@@ -506,6 +511,7 @@ def _materialize_one(
     providers: Mapping[str, SweepConfigProvider],
     provider_plans: Mapping[str, AdapterSearchPlan],
     runner_factory: RunnerFactory,
+    engine_controls: Mapping[str, EngineControlTemplate],
     prediction_config_factory: Callable[[dict[str, Any], ReplaySpec], dict[str, Any]] | None = None,
 ) -> tuple[_PreparedCandidate | None, _EvalResult | None]:
     """Build a complete replay specification on the main process."""
@@ -522,6 +528,10 @@ def _materialize_one(
         # on the candidate so downstream artifact generation cannot independently
         # select a different backend version.
         sample["backend_version"] = backend_version
+        engine_request = materialize_engine_request(
+            engine_controls[selection["backend"]], config=config, sample=sample
+        )
+        sample["engine_request"] = asdict(engine_request)
         concurrency = config.workload.concurrency
         workload_payload = config.workload.model_dump(mode="json")
         if "traffic_load" in selection:
@@ -561,7 +571,11 @@ def _materialize_one(
             # Preserve the concrete load on every candidate, including a fixed absolute
             # concurrency and one derived from kv_load_ratio.
             sample["concurrency"] = concurrency
-        backend_deployment = build_backend_deployment(sample, backend_version=backend_version)
+        backend_deployment = build_backend_deployment(
+            sample,
+            backend_version=backend_version,
+            engine_request=engine_request,
+        )
         adapter_specs: dict[str, AdapterReplaySpec] = {}
         for name, provider in providers.items():
             candidate_context = CandidateContext(
@@ -793,12 +807,14 @@ def run(
         goal = config.goal
         capabilities = runner_factory.capabilities()
         capabilities.require_replay_spec_version(REPLAY_SPEC_API_VERSION)
+        engine_controls = resolve_engine_controls(config)
+        max_seq_len = next(iter(engine_controls.values())).max_seq_len
 
         # Preserve the legacy preflight order: reject an impossible backend/topology
         # search before adapters perform any potentially expensive preparation.
         branches = enumerate_branches(
             config,
-            max_seq_len=config.search_space.context_length,
+            max_seq_len=max_seq_len,
             runner_capabilities=capabilities,
         )
         resolved_providers, provider_plans = _prepare_providers(config, injected=providers, show_progress=show_progress)
@@ -851,6 +867,7 @@ def run(
                 "workload": config.workload.model_dump(mode="python"),
                 "goal": goal.model_dump(mode="python"),
                 "provider_plans": provider_plans,
+                "engine_controls": engine_controls,
             }
         )
         replay_cache: dict[Any, tuple[Candidate, dict[str, float]]] = {}
@@ -1111,6 +1128,7 @@ def _run_branch_round(state: _BranchSearchState) -> None:
                             providers=resolved_providers,
                             provider_plans=provider_plans,
                             runner_factory=runner_factory,
+                            engine_controls=engine_controls,
                             prediction_config_factory=prediction_config_factory,
                         )
                         if build_result is not None:
diff --git a/python/aisimulate/src/aisimulate/sweeper/search_space.py b/python/aisimulate/src/aisimulate/sweeper/search_space.py
index 4fd51162e..b5d7ef337 100644
--- a/python/aisimulate/src/aisimulate/sweeper/search_space.py
+++ b/python/aisimulate/src/aisimulate/sweeper/search_space.py
@@ -201,7 +201,11 @@ def role_runtime(backend: str, mode: str) -> dict[str, tuple[int, int, float, in
             memory = getattr(ss, f"{role}_gpu_memory_utilization")
             block_size = getattr(ss, f"{role}_block_size")
             blocks = getattr(ss, f"{role}_num_gpu_blocks")
-            memory_values = memory if isinstance(memory, list) else [memory]
+            memory_values = (
+                [ss.free_gpu_memory_fraction]
+                if ss.free_gpu_memory_fraction is not None
+                else (memory if isinstance(memory, list) else [memory])
+            )
             concrete_memory = [float(value) for value in memory_values if value is not None]
             default_memory = 0.88 if backend == "sglang" else 0.9
             block_values = block_size if isinstance(block_size, list) else [block_size]
@@ -243,6 +247,23 @@ def matches_custom(config: _ParallelConfig) -> bool:
             ):
                 continue
             try:
+                engine_kwargs: dict[str, Any] = {}
+                if ss.enable_wideep:
+                    engine_kwargs["enable_wideep"] = True
+                if ss.moe_backend is not None:
+                    engine_kwargs["moe_backend"] = ss.moe_backend
+                for name in (
+                    "gemm_quant_mode",
+                    "moe_quant_mode",
+                    "kvcache_quant_mode",
+                    "fmha_quant_mode",
+                    "comm_quant_mode",
+                ):
+                    value = getattr(ss, name)
+                    if value is not None:
+                        engine_kwargs[name] = value
+                if ss.aic_nextn is not None:
+                    engine_kwargs["nextn"] = ss.aic_nextn
                 legal = parallel_configs_for(
                     ss.model_name,
                     ss.hardware_sku,
@@ -252,6 +273,7 @@ def matches_custom(config: _ParallelConfig) -> bool:
                     backend_version=ss.backend_version,
                     min_gpu_budget=ss.min_gpu_budget,
                     max_seq_len=max_seq_len,
+                    **engine_kwargs,
                     role_runtime=role_runtime(backend, deployment_mode),
                 )
             except (NoPerfDatabase, NoViableParallelConfig):
diff --git a/tests/sweeper/test_config.py b/tests/sweeper/test_config.py
index 4876b29f2..bb11ba875 100644
--- a/tests/sweeper/test_config.py
+++ b/tests/sweeper/test_config.py
@@ -139,6 +139,71 @@ def test_defaults_are_backend_only():
     )
 
 
+def test_engine_request_controls_round_trip_through_yaml(tmp_path):
+    path = tmp_path / "engine-controls.yaml"
+    path.write_text(
+        """
+search_space:
+  deployment_mode: [agg]
+  backend: [sglang]
+  model_name: example/moe
+  hardware_sku: gb200
+  max_seq_len: 8192
+  enable_chunked_prefill: true
+  enable_wideep: true
+  enable_eplb: true
+  wideep_num_slots: 64
+  moe_backend: deepep_moe
+  attention_backend: fa3
+  gemm_quant_mode: fp8
+  moe_quant_mode: fp8
+  kvcache_quant_mode: fp8
+  fmha_quant_mode: fp8
+  comm_quant_mode: fp8
+  aic_nextn: 3
+  nextn_accepted: 1.5
+  free_gpu_memory_fraction: 0.82
+workload:
+  isl: 4096
+  osl: 1024
+  concurrency: 2
+  num_request_ratio: 1
+  cached_prefix_tokens: 512
+"""
+    )
+
+    config = SmartSearchConfig.from_yaml(path)
+    dumped = config.model_dump(mode="json")
+
+    assert dumped["workload"]["cached_prefix_tokens"] == 512
+    assert dumped["search_space"]["enable_chunked_prefill"] is True
+    assert dumped["search_space"]["wideep_num_slots"] == 64
+    assert dumped["search_space"]["attention_backend"] == "fa3"
+    assert dumped["search_space"]["kvcache_quant_mode"] == "fp8"
+    assert config.search_space.nextn_accepted == 1.5
+    assert config.search_space.free_gpu_memory_fraction == 0.82
+
+
+def test_speculative_acceptance_is_explicit_and_bounded():
+    with pytest.raises(ValidationError, match="requires explicit nextn_accepted"):
+        SearchSpace(**_search_space(aic_nextn=2))
+    with pytest.raises(ValidationError, match=r"within \[0, aic_nextn=2\]"):
+        SearchSpace(**_search_space(aic_nextn=2, nextn_accepted=2.5))
+
+    disabled = SearchSpace(**_search_space(aic_nextn=0))
+    assert disabled.aic_nextn == 0
+    assert disabled.nextn_accepted is None
+    with pytest.raises(ValidationError, match="greater than zero"):
+        SearchSpace(**_search_space(aic_nextn=0, nextn_accepted=0))
+
+
+def test_cached_prefix_is_exact_and_within_shortest_input():
+    workload = Workload(**_workload(cached_prefix_tokens=2000, random_range_ratio=0.5))
+    assert workload.cached_prefix_tokens == 2000
+    with pytest.raises(ValidationError, match="shortest synthetic input"):
+        Workload(**_workload(cached_prefix_tokens=2001, random_range_ratio=0.5))
+
+
 def test_extra_fields_are_forbidden_at_each_boundary():
     with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
         SmartSearchConfig(
diff --git a/tests/sweeper/test_deploy.py b/tests/sweeper/test_deploy.py
index a55c14ff6..24a8a27b7 100644
--- a/tests/sweeper/test_deploy.py
+++ b/tests/sweeper/test_deploy.py
@@ -13,6 +13,7 @@
     ParallelShape,
     ReplicaParallelConfig,
 )
+from aisimulate.sweeper.replay import EngineRequestSpec
 from aisimulate.sweeper.sample import unroll_sample
 
 BACKEND_VERSION = "1.3.0rc10"
@@ -171,13 +172,66 @@ def test_memory_fraction_uses_the_backend_native_field(backend, memory_field):
 
 def test_optional_backend_runtime_values_are_forwarded():
     engine = _agg_deployment(
-        space=_space(startup_time=45.0, aic_nextn=2)
+        space=_space(startup_time=45.0, aic_nextn=2, nextn_accepted=1.25)
     ).agg_engine_args
 
     assert engine["startup_time"] == 45.0
     assert engine["aic_nextn"] == 2
 
 
+def test_zero_speculative_depth_is_not_forwarded_to_the_runner():
+    engine = _agg_deployment(space=_space(aic_nextn=0)).agg_engine_args
+
+    assert "aic_nextn" not in engine
+
+
+def test_resolved_engine_request_controls_are_materialized_for_replay():
+    sample = unroll_sample(
+        search_space=_space(aic_nextn=2, nextn_accepted=1.25),
+        selection=_agg_selection(backend="sglang"),
+        parallel_config=AGG_MOE,
+    )
+    engine_request = EngineRequestSpec(
+        cached_prefix_tokens=512,
+        context_tokens={"agg": 16384},
+        enable_chunked_prefill=True,
+        enable_wideep=True,
+        enable_eplb=True,
+        wideep_num_slots=64,
+        moe_backend="deepep_moe",
+        attention_backend="fa3",
+        gemm_quant_mode="fp8",
+        moe_quant_mode="fp8",
+        kvcache_quant_mode="fp8",
+        fmha_quant_mode="fp8",
+        comm_quant_mode="fp8",
+        nextn=2,
+        nextn_accepted=1.25,
+        memory_fraction_kind="of_total",
+        memory_fraction_by_role={"agg": 0.82},
+        max_seq_len=8192,
+        model_family="DEEPSEEK",
+    )
+
+    deployment = build_backend_deployment(
+        sample,
+        backend_version=BACKEND_VERSION,
+        engine_request=engine_request,
+    )
+    engine = deployment.agg_engine_args
+
+    assert deployment.engine_request == engine_request
+    assert engine["max_num_batched_tokens"] == 16384
+    assert "max_model_len" not in engine
+    assert engine["enable_chunked_prefill"] is True
+    assert engine["mem_fraction_static"] == 0.82
+    assert engine["aic_nextn_accepted"] == 1.25
+    assert engine["aic_enable_eplb"] is True
+    assert engine["aic_wideep_num_slots"] == 64
+    assert engine["aic_attention_backend"] == "fa3"
+    assert engine["aic_kv_cache_dtype"] == "fp8"
+
+
 def test_backend_deployment_contains_no_dynamo_policy_fields():
     deployment = _agg_deployment()
 
diff --git a/tests/sweeper/test_engine_request.py b/tests/sweeper/test_engine_request.py
new file mode 100644
index 000000000..31e885d4f
--- /dev/null
+++ b/tests/sweeper/test_engine_request.py
@@ -0,0 +1,145 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from types import SimpleNamespace
+
+import pytest
+
+import aisimulate.sweeper.engine_request as engine_request_mod
+from aiconfigurator_core.sdk.config_builders import build_model_config
+from aisimulate.sweeper.config import SmartSearchConfig
+from aisimulate.sweeper.engine_request import (
+    EngineControlTemplate,
+    materialize_engine_request,
+    resolve_engine_controls,
+)
+
+
+def _config(**search_overrides) -> SmartSearchConfig:
+    search_space = {
+        "model_name": "example/moe",
+        "hardware_sku": "gb200",
+        "backend": ["sglang"],
+        "deployment_mode": ["disagg"],
+        "max_seq_len": 8192,
+    }
+    search_space.update(search_overrides)
+    return SmartSearchConfig(
+        search_space=search_space,
+        workload={
+            "isl": 4096,
+            "osl": 1024,
+            "concurrency": 2,
+            "num_request_ratio": 1,
+            "cached_prefix_tokens": 512,
+        },
+    )
+
+
+def _stub_model(monkeypatch, *, is_moe=True, mla=True, family="DEEPSEEK") -> None:
+    monkeypatch.setattr(engine_request_mod, "get_model_family", lambda model: family)
+    monkeypatch.setattr(
+        engine_request_mod,
+        "resolve_model_hardware",
+        lambda *args, **kwargs: SimpleNamespace(
+            is_moe=is_moe,
+            mla=mla,
+            max_context=16384,
+        ),
+    )
+
+
+def test_resolves_supported_engine_controls_before_replay(monkeypatch):
+    _stub_model(monkeypatch)
+    config = _config(
+        enable_chunked_prefill=True,
+        enable_wideep=True,
+        enable_eplb=True,
+        wideep_num_slots=64,
+        moe_backend="deepep_moe",
+        attention_backend="fa3",
+        gemm_quant_mode="fp8",
+        moe_quant_mode="fp8",
+        kvcache_quant_mode="fp8",
+        fmha_quant_mode="fp8",
+        comm_quant_mode="fp8",
+        aic_nextn=3,
+        nextn_accepted=1.5,
+        free_gpu_memory_fraction=0.82,
+    )
+
+    assert resolve_engine_controls(config) == {
+        "sglang": EngineControlTemplate(
+            backend="sglang",
+            max_seq_len=8192,
+            model_family="DEEPSEEK",
+            is_moe=True,
+            memory_fraction_kind="of_total",
+        )
+    }
+
+
+def test_rejects_unsupported_model_backend_controls(monkeypatch):
+    _stub_model(monkeypatch, is_moe=False, mla=False, family="GPT")
+    with pytest.raises(ValueError, match="MoE controls require an MoE model"):
+        resolve_engine_controls(_config(enable_eplb=True, enable_wideep=True))
+
+    _stub_model(monkeypatch, is_moe=True, mla=True)
+    with pytest.raises(ValueError, match="only with backend='sglang'"):
+        resolve_engine_controls(_config(backend=["vllm"], attention_backend="fa3"))
+
+
+def test_rejects_invalid_quant_and_sequence_capacity(monkeypatch):
+    _stub_model(monkeypatch)
+    with pytest.raises(ValueError, match="gemm_quant_mode has unsupported value"):
+        resolve_engine_controls(_config(gemm_quant_mode="made_up"))
+    with pytest.raises(ValueError, match="cannot hold the synthetic request"):
+        resolve_engine_controls(_config(max_seq_len=5000))
+
+
+def test_materializes_shared_controls_for_both_disaggregated_roles(monkeypatch):
+    _stub_model(monkeypatch)
+    config = _config(
+        enable_chunked_prefill=True,
+        free_gpu_memory_fraction=0.81,
+        aic_nextn=2,
+        nextn_accepted=1.25,
+    )
+    template = resolve_engine_controls(config)["sglang"]
+    sample = {
+        "deployment_mode": "disagg",
+        "prefill_max_num_batched_tokens": 32768,
+        "decode_max_num_batched_tokens": 8192,
+        "prefill_gpu_memory_utilization": 0.9,
+        "decode_gpu_memory_utilization": 0.88,
+    }
+
+    request = materialize_engine_request(template, config=config, sample=sample)
+
+    assert request.cached_prefix_tokens == 512
+    assert request.context_tokens == {"prefill": 32768, "decode": 8192}
+    assert request.memory_fraction_by_role == {"prefill": 0.81, "decode": 0.81}
+    assert request.max_seq_len == 8192
+    assert request.nextn == 2
+    assert request.nextn_accepted == 1.25
+
+
+def test_aic_model_config_receives_backend_and_eplb_controls():
+    model_config = build_model_config(
+        tp_size=1,
+        pp_size=1,
+        attention_dp_size=8,
+        moe_tp_size=1,
+        moe_ep_size=8,
+        moe_backend="megamoe",
+        attention_backend="fa3",
+        enable_wideep=True,
+        enable_eplb=True,
+        wideep_num_slots=64,
+    )
+
+    assert model_config.moe_backend == "megamoe"
+    assert model_config.attention_backend == "fa3"
+    assert model_config.enable_wideep is True
+    assert model_config.enable_eplb is True
+    assert model_config.wideep_num_slots == 64
diff --git a/tests/sweeper/test_kv_estimate.py b/tests/sweeper/test_kv_estimate.py
index e8915db01..248749ca0 100644
--- a/tests/sweeper/test_kv_estimate.py
+++ b/tests/sweeper/test_kv_estimate.py
@@ -30,13 +30,39 @@ def test_memory_fraction_kind():
 
 
 def test_estimate_kv_tokens_returns_capacity(monkeypatch):
+    calls = []
+
+    def estimate(*args, **kwargs):
+        calls.append(kwargs)
+        return {"total_kv_size_tokens": 123456}
+
     monkeypatch.setattr(
         kv_estimate_mod,
         "estimate_kv_cache",
-        lambda *args, **kwargs: {"total_kv_size_tokens": 123456},
+        estimate,
     )
     sh = ParallelShape(tp=4, dp=1, moe_tp=1, moe_ep=4)
-    assert estimate_kv_tokens(sh, **_COMMON) == 123456
+    assert (
+        estimate_kv_tokens(
+            sh,
+            **_COMMON,
+            memory_fraction=0.82,
+            nextn=3,
+            gemm_quant_mode="fp8",
+            moe_quant_mode="fp8",
+            kvcache_quant_mode="fp8",
+            fmha_quant_mode="fp8",
+            comm_quant_mode="fp8",
+        )
+        == 123456
+    )
+    assert calls[0]["memory_fraction_value"] == 0.82
+    assert calls[0]["nextn"] == 3
+    assert calls[0]["gemm_quant_mode"] == "fp8"
+    assert calls[0]["moe_quant_mode"] == "fp8"
+    assert calls[0]["kvcache_quant_mode"] == "fp8"
+    assert calls[0]["fmha_quant_mode"] == "fp8"
+    assert calls[0]["comm_quant_mode"] == "fp8"
 
 
 def test_estimate_kv_tokens_oom_returns_none(monkeypatch):
diff --git a/tests/sweeper/test_sample.py b/tests/sweeper/test_sample.py
index 7ca424c35..adde4bbd9 100644
--- a/tests/sweeper/test_sample.py
+++ b/tests/sweeper/test_sample.py
@@ -121,6 +121,7 @@ def test_unroll_folds_only_backend_pinned_values():
             context_length=4096,
             startup_time=300.0,
             aic_nextn=2,
+            nextn_accepted=1.25,
             agg_block_size=32,
             agg_gpu_memory_utilization=0.8,
             agg_enable_prefix_caching=False,
@@ -134,6 +135,7 @@ def test_unroll_folds_only_backend_pinned_values():
     assert sample["context_length"] == 4096
     assert sample["startup_time"] == 300.0
     assert sample["aic_nextn"] == 2
+    assert sample["nextn_accepted"] == 1.25
     assert sample["agg_block_size"] == 32
     assert sample["agg_gpu_memory_utilization"] == 0.8
     assert sample["agg_enable_prefix_caching"] is False
diff --git a/tests/sweeper/test_search.py b/tests/sweeper/test_search.py
index 3412bf56e..b5d296188 100644
--- a/tests/sweeper/test_search.py
+++ b/tests/sweeper/test_search.py
@@ -9,6 +9,7 @@
 
 import aisimulate.sweeper.search as search_mod
 from aisimulate.sweeper.config import OptimizationGoal, SLATarget, SmartSearchConfig
+from aisimulate.sweeper.engine_request import EngineControlTemplate
 from aisimulate.sweeper.kv_load import KVLoadResolution
 from aisimulate.sweeper.parallel_enum import ParallelShape, ReplicaParallelConfig
 from aisimulate.sweeper.replay import (
@@ -147,6 +148,19 @@ def _stub(monkeypatch, branch):
         "enumerate_branches",
         lambda config, *, max_seq_len=None, runner_capabilities=None: [branch],
     )
+    monkeypatch.setattr(
+        search_mod,
+        "resolve_engine_controls",
+        lambda config: {
+            "trtllm": EngineControlTemplate(
+                backend="trtllm",
+                max_seq_len=131072,
+                model_family="DEEPSEEK",
+                is_moe=True,
+                memory_fraction_kind="of_free",
+            )
+        },
+    )
     monkeypatch.setattr(
         search_mod, "resolve_backend_version", lambda hw, be: "1.3.0rc10"
     )
@@ -178,6 +192,10 @@ def test_ranks_feasible_best_first_and_passes_replay_specs(monkeypatch):
     assert candidates[0].metrics["gpu_hours"] == 1.0
     assert factory.worker_ids == [0]
     assert all(isinstance(spec, ReplaySpec) for spec in factory.runner.specs)
+    assert candidates[0].config["engine_request"]["max_seq_len"] == 131072
+    assert (
+        factory.runner.specs[0].backend_deployment.engine_request.max_seq_len == 131072
+    )
     assert factory.runner.closed
 
 
@@ -809,6 +827,19 @@ def test_projection_stall_only_stops_current_branch(monkeypatch):
         "enumerate_branches",
         lambda config, *, max_seq_len=None, runner_capabilities=None: [agg, disagg],
     )
+    monkeypatch.setattr(
+        search_mod,
+        "resolve_engine_controls",
+        lambda config: {
+            "trtllm": EngineControlTemplate(
+                backend="trtllm",
+                max_seq_len=131072,
+                model_family="DEEPSEEK",
+                is_moe=True,
+                memory_fraction_kind="of_free",
+            )
+        },
+    )
     monkeypatch.setattr(
         search_mod, "resolve_backend_version", lambda hw, be: "1.3.0rc10"
     )
diff --git a/tests/sweeper/test_search_providers.py b/tests/sweeper/test_search_providers.py
index 44ceac1cf..04eb75bc7 100644
--- a/tests/sweeper/test_search_providers.py
+++ b/tests/sweeper/test_search_providers.py
@@ -9,6 +9,7 @@
 
 import aisimulate.sweeper.search as search_module
 from aisimulate.sweeper.config import SmartSearchConfig
+from aisimulate.sweeper.engine_request import EngineControlTemplate
 from aisimulate.sweeper.parallel_enum import ParallelShape, ReplicaParallelConfig
 from aisimulate.sweeper.provider import (
     AdapterReplaySpec,
@@ -248,6 +249,19 @@ def _stub_branch(monkeypatch) -> None:
         "resolve_backend_version",
         lambda hardware, backend: "0.11.0",
     )
+    monkeypatch.setattr(
+        search_module,
+        "resolve_engine_controls",
+        lambda config: {
+            "vllm": EngineControlTemplate(
+                backend="vllm",
+                max_seq_len=4096,
+                model_family="GPT",
+                is_moe=False,
+                memory_fraction_kind="of_total",
+            )
+        },
+    )
 
 
 def test_conditional_adapter_fragment_is_validated_namespaced_and_merged() -> None:
@@ -390,6 +404,15 @@ def materialize_replay(self, plan, selection, context):
         providers={"test.feature": InfeasibleAdapter()},
         provider_plans={"test.feature": AdapterSearchPlan()},
         runner_factory=_RunnerFactory(),
+        engine_controls={
+            "vllm": EngineControlTemplate(
+                backend="vllm",
+                max_seq_len=4096,
+                model_family="GPT",
+                is_moe=False,
+                memory_fraction_kind="of_total",
+            )
+        },
     )
 
     assert prepared is None
@@ -422,6 +445,19 @@ def reject_branches(*args, **kwargs):
         raise ValueError("no viable backend/topology branch")
 
     monkeypatch.setattr(search_module, "enumerate_branches", reject_branches)
+    monkeypatch.setattr(
+        search_module,
+        "resolve_engine_controls",
+        lambda config: {
+            "vllm": EngineControlTemplate(
+                backend="vllm",
+                max_seq_len=4096,
+                model_family="GPT",
+                is_moe=False,
+                memory_fraction_kind="of_total",
+            )
+        },
+    )
 
     with pytest.raises(ValueError, match="no viable backend/topology branch"):
         _run_sweep(
diff --git a/tests/sweeper/test_unified_optimizer.py b/tests/sweeper/test_unified_optimizer.py
index 2904dffcf..933bb6e28 100644
--- a/tests/sweeper/test_unified_optimizer.py
+++ b/tests/sweeper/test_unified_optimizer.py
@@ -7,7 +7,9 @@
 from typing import ClassVar
 
 import aisimulate.sweeper.search as search_module
+import pytest
 from aisimulate.sweeper.config import SmartSearchConfig
+from aisimulate.sweeper.engine_request import EngineControlTemplate
 from aisimulate.sweeper.parallel_enum import (
     DisaggParallelConfig,
     ParallelShape,
@@ -22,6 +24,24 @@
 from aisimulate.sweeper.search_space import BranchSpace
 
 
+@pytest.fixture(autouse=True)
+def _stub_engine_controls(monkeypatch) -> None:
+    monkeypatch.setattr(
+        search_module,
+        "resolve_engine_controls",
+        lambda config: {
+            backend: EngineControlTemplate(
+                backend=backend,
+                max_seq_len=10,
+                model_family="TEST",
+                is_moe=False,
+                memory_fraction_kind="of_total",
+            )
+            for backend in config.search_space.backend
+        },
+    )
+
+
 class _Runner:
     runs: ClassVar[int] = 0
 
diff --git a/tests/test_aic.py b/tests/test_aic.py
index dbbb5f530..fdf243224 100644
--- a/tests/test_aic.py
+++ b/tests/test_aic.py
@@ -15,7 +15,7 @@
 ]
 
 
-def test_materializer_sets_rank_local_capacity_without_forwarding_nextn(
+def test_materializer_sets_rank_local_capacity_with_nextn(
     monkeypatch,
 ) -> None:
     calls = []
@@ -45,8 +45,8 @@ def estimate(**kwargs):
     assert calls[0]["attention_dp_size"] == 2
     assert calls[0]["pp_size"] == 3
     assert calls[0]["systems_path"] == "/tmp/custom-systems.yaml"
+    assert calls[0]["nextn"] == 3
     assert calls[0]["max_num_sequences"] == 37
-    assert "nextn" not in calls[0]
 
 
 def test_capacity_wrapper_owns_backend_defaults_and_quant_normalization(
@@ -86,7 +86,7 @@ def estimate(*args, **kwargs):
     assert kwargs["systems_path"] == "/tmp/custom-systems.yaml"
     assert kwargs["gemm_quant_mode"] == "int4_wo"
     assert kwargs["fmha_quant_mode"] is None
-    assert "nextn" not in kwargs
+    assert kwargs["nextn"] == 0
 
 
 def test_explicit_capacity_is_preserved_without_estimation(monkeypatch) -> None:
diff --git a/tests/test_cli_config.py b/tests/test_cli_config.py
index 825e4a8bf..942230a2e 100644
--- a/tests/test_cli_config.py
+++ b/tests/test_cli_config.py
@@ -3,7 +3,10 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+
 import pytest
+import yaml
 from pydantic import ValidationError
 
 from aisimulate.config import CorePredictionConfig, CoreRecommendationConfig
@@ -15,6 +18,8 @@
 )
 from aisimulate.recommend import recommendation_to_sweeper
 
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+
 
 def _engine() -> dict:
     return {
@@ -132,6 +137,102 @@ def test_recommendation_accepts_domains_and_parallel_preset() -> None:
     assert isinstance(config.engine, EngineRecommendationConfig)
 
 
+def test_aic_engine_controls_lower_as_pinned_flat_fields() -> None:
+    config = CoreRecommendationConfig.model_validate(
+        {
+            "traffic": {
+                "source": {
+                    "type": "synthetic",
+                    "input_tokens": 4096,
+                    "output_tokens": 1024,
+                    "cached_prefix_tokens": 1024,
+                },
+                "load": {"type": "concurrency", "concurrency": 1},
+                "stop": {"requests": 1},
+            },
+            "engine": {
+                "mode": "aggregated",
+                "model": "example/moe-model",
+                "hardware": "h200_sxm",
+                "backend": "sglang",
+                "context_length": 8192,
+                "nextn": 3,
+                "nextn_accepted": 1.5,
+                "enable_chunked_prefill": True,
+                "enable_wideep": True,
+                "enable_eplb": True,
+                "wideep_num_slots": 64,
+                "moe_backend": "deepep_moe",
+                "attention_backend": "fa3",
+                "gemm_quant_mode": "fp8",
+                "moe_quant_mode": "fp8",
+                "kvcache_quant_mode": "fp8",
+                "fmha_quant_mode": "fp8",
+                "comm_quant_mode": "fp8",
+                "free_gpu_memory_fraction": 0.85,
+                "workers": {"aggregated": {}},
+            },
+            "optimization": {"constraints": {"max_candidate_gpus": 8}},
+        }
+    )
+
+    lowered = recommendation_to_sweeper(config)
+
+    assert lowered.workload.cached_prefix_tokens == 1024
+    assert lowered.search_space.aic_nextn == 3
+    assert lowered.search_space.nextn_accepted == 1.5
+    assert lowered.search_space.enable_chunked_prefill is True
+    assert lowered.search_space.enable_wideep is True
+    assert lowered.search_space.enable_eplb is True
+    assert lowered.search_space.wideep_num_slots == 64
+    assert lowered.search_space.moe_backend == "deepep_moe"
+    assert lowered.search_space.attention_backend == "fa3"
+    assert lowered.search_space.gemm_quant_mode == "fp8"
+    assert lowered.search_space.moe_quant_mode == "fp8"
+    assert lowered.search_space.kvcache_quant_mode == "fp8"
+    assert lowered.search_space.fmha_quant_mode == "fp8"
+    assert lowered.search_space.comm_quant_mode == "fp8"
+    assert lowered.search_space.free_gpu_memory_fraction == 0.85
+
+
+def test_aic_engine_controls_reject_implicit_speculative_acceptance() -> None:
+    with pytest.raises(ValidationError, match="requires explicit nextn_accepted"):
+        CoreRecommendationConfig.model_validate(
+            {
+                "engine": {**_engine(), "mode": "aggregated", "nextn": 2},
+                "optimization": {},
+            }
+        )
+    with pytest.raises(ValidationError, match="cannot exceed input_tokens"):
+        CorePredictionConfig.model_validate(
+            {
+                "traffic": {
+                    "source": {
+                        "type": "synthetic",
+                        "input_tokens": 8,
+                        "cached_prefix_tokens": 9,
+                    },
+                    "load": {"type": "concurrency"},
+                    "stop": {"requests": 1},
+                },
+                "engine": _engine(),
+            }
+        )
+
+
+def test_aic_migration_example_is_a_valid_recommendation() -> None:
+    text = (_REPO_ROOT / "docs/cli/migrate-from-aiconfigurator.md").read_text()
+    section = text.split("### Example with the new controls", 1)[1]
+    yaml_block = section.split("```yaml", 1)[1].split("```", 1)[0]
+
+    config = CoreRecommendationConfig.model_validate(yaml.safe_load(yaml_block))
+    lowered = recommendation_to_sweeper(config)
+
+    assert lowered.search_space.aic_nextn == 3
+    assert lowered.search_space.nextn_accepted == 1.5
+    assert lowered.workload.cached_prefix_tokens == 1024
+
+
 def test_parallel_preset_modes_lower_without_conflating_semantics() -> None:
     independent = {
         "preset": False,
diff --git a/tests/test_runner.py b/tests/test_runner.py
index 1cfcf492a..37d9a2584 100644
--- a/tests/test_runner.py
+++ b/tests/test_runner.py
@@ -94,6 +94,7 @@ def _spec(
 def test_public_namespace_exports_engine_runner_contract():
     assert aisimulate.EngineReplayRunner is EngineReplayRunner
     assert aisimulate.EngineReplayRunnerFactory is EngineReplayRunnerFactory
+    assert aisimulate.EngineRequestSpec.__name__ == "EngineRequestSpec"
 
 
 def test_factory_is_pickleable_and_advertises_engine_only_capabilities():
@@ -133,6 +134,78 @@ def test_runner_lowers_canonical_spec_and_returns_replay_report():
     assert report.metadata == {}
 
 
+def test_runner_materializes_exact_cached_prefix_tokens():
+    runtime = RecordingRuntime()
+    workload = {
+        "isl": 8,
+        "osl": 2,
+        "concurrency": 2,
+        "num_request_ratio": 1,
+        "cached_prefix_tokens": 3,
+    }
+
+    EngineReplayRunnerFactory(runtime=runtime).create(0).run(
+        _spec(workload=workload, concurrency=2)
+    )
+
+    first, second = runtime.execution_spec["requests"]
+    assert first["input_token_ids"][:3] == [1, 2, 3]
+    assert second["input_token_ids"][:3] == [1, 2, 3]
+    assert first["input_token_ids"][3:] != second["input_token_ids"][3:]
+    assert len(first["input_token_ids"]) == first["input_tokens"] == 8
+
+
+def test_runner_lowers_engine_controls_into_scheduler_and_aic_timing():
+    runtime = RecordingRuntime()
+    engine_args = _engine_args()
+    engine_args.pop("timing_model")
+    engine_args.update(
+        {
+            "aic_backend_version": "test",
+            "max_model_len": 8192,
+            "max_num_batched_tokens": 32768,
+            "enable_chunked_prefill": True,
+            "aic_nextn": 3,
+            "aic_nextn_accepted": 1.5,
+            "aic_enable_wideep": True,
+            "aic_enable_eplb": True,
+            "aic_wideep_num_slots": 64,
+            "aic_moe_backend": "deepep_moe",
+            "aic_attention_backend": "fa3",
+            "aic_gemm_dtype": "fp8",
+            "aic_moe_dtype": "fp8",
+            "aic_kv_cache_dtype": "fp8",
+            "aic_fmha_dtype": "fp8",
+            "aic_comm_dtype": "fp8",
+        }
+    )
+    deployment = BackendDeploymentSpec(
+        deployment_mode="agg",
+        backend="vllm",
+        backend_version="test",
+        agg_engine_args=engine_args,
+        num_workers=1,
+    )
+
+    EngineReplayRunnerFactory(runtime=runtime).create(0).run(
+        _spec(deployment=deployment)
+    )
+
+    rank = runtime.execution_spec["engine"]["rank"]
+    timing = rank["timing_model"]["config"]
+    assert rank["max_model_len"] == 8192
+    assert rank["max_num_batched_tokens"] == 32768
+    assert rank["enable_chunked_prefill"] is True
+    assert rank["aic_nextn_accept_rates"] == "1,0.5,0"
+    assert timing["enable_wideep"] is True
+    assert timing["enable_eplb"] is True
+    assert timing["wideep_num_slots"] == 64
+    assert timing["moe_backend"] == "deepep_moe"
+    assert timing["attention_backend"] == "fa3"
+    assert timing["gemm_dtype"] == "fp8"
+    assert timing["kv_cache_dtype"] == "fp8"
+
+
 @pytest.mark.parametrize(("field", "bound"), [("ttft_ms", 800.0), ("itl_ms", 30.0)])
 def test_engine_runner_preserves_independent_sla_bounds(
     field: str, bound: float
@@ -211,7 +284,7 @@ def estimate(**kwargs):
     assert timing_config["systems_path"] == "/tmp/custom-systems.yaml"
     assert calls[0]["pp_size"] == 2
     assert calls[0]["systems_path"] == "/tmp/custom-systems.yaml"
-    assert "nextn" not in calls[0]
+    assert calls[0]["nextn"] == 3
 
 
 def test_runner_keeps_capacity_estimation_independent_from_fixed_timing(monkeypatch):
diff --git a/tests/test_unified_traffic_runtime.py b/tests/test_unified_traffic_runtime.py
index 01d5b5360..1bdafbf18 100644
--- a/tests/test_unified_traffic_runtime.py
+++ b/tests/test_unified_traffic_runtime.py
@@ -65,6 +65,48 @@ def test_prediction_spec_separates_perf_identity_from_fixed_timing() -> None:
     assert deployment.agg_engine_args["timing_model"]["type"] == "fixed"
 
 
+def test_prediction_compiles_aic_engine_and_cached_prefix_controls() -> None:
+    parsed = CorePredictionConfig.model_validate(
+        {
+            "traffic": {
+                "source": {
+                    "type": "synthetic",
+                    "input_tokens": 8,
+                    "output_tokens": 2,
+                    "cached_prefix_tokens": 3,
+                },
+                "load": {"type": "concurrency", "concurrency": 1},
+                "stop": {"requests": 1},
+            },
+            "engine": {
+                "model": "example/model",
+                "hardware": "h200_sxm",
+                "backend": "vllm",
+                "context_length": 1024,
+                "nextn": 2,
+                "nextn_accepted": 1.25,
+                "enable_chunked_prefill": True,
+                "gemm_quant_mode": "fp8",
+                "kvcache_quant_mode": "fp8",
+                "free_gpu_memory_fraction": 0.85,
+                "workers": {"aggregated": {}},
+            },
+        }
+    )
+
+    spec = prediction_to_replay_spec(parsed)
+    args = spec.backend_deployment.agg_engine_args
+
+    assert spec.workload["cached_prefix_tokens"] == 3
+    assert args["max_model_len"] == 1024
+    assert args["enable_chunked_prefill"] is True
+    assert args["aic_nextn"] == 2
+    assert args["aic_nextn_accepted"] == 1.25
+    assert args["aic_gemm_dtype"] == "fp8"
+    assert args["aic_kv_cache_dtype"] == "fp8"
+    assert args["gpu_memory_utilization"] == 0.85
+
+
 def test_engine_stack_runs_ordered_synthetic_sessions() -> None:
     report = _run(
         {
@@ -95,6 +137,29 @@ def test_engine_stack_runs_ordered_synthetic_sessions() -> None:
     assert [row["input_length"] for row in records[:3]] == [8, 18, 28]
 
 
+def test_engine_stack_reuses_exact_cached_prefix_from_public_traffic() -> None:
+    engine = _engine()
+    engine["workers"]["aggregated"]["kv_cache"]["block_size"] = 4
+    report = _run(
+        {
+            "traffic": {
+                "source": {
+                    "type": "synthetic",
+                    "input_tokens": 8,
+                    "output_tokens": 2,
+                    "cached_prefix_tokens": 4,
+                },
+                "load": {"type": "concurrency", "concurrency": 1},
+                "stop": {"requests": 2},
+            },
+            "engine": engine,
+        }
+    )
+
+    records = report.metadata["native_report"]["per_request"]
+    assert [row["reused_input_tokens"] for row in records] == [0, 4]
+
+
 def test_engine_stack_runs_mooncake_delta(tmp_path) -> None:
     trace = tmp_path / "delta.jsonl"
     rows = [