fix(mocker): adopt AISimulate nightly 82

Cargo.toml

@@ -52,7 +52,7 @@ keywords = ["llm", "genai", "inference", "nvidia", "distributed"]
 [workspace.dependencies]
 dynamo-custom-policy-builtin = { path = "lib/router-plugins/builtin", version = "1.7.0" }
 # Published separately by https://github.com/ai-dynamo/aisimulate.
-aisimulate-core = { version = "=0.13.0-dev.202610160000000073" }
+aisimulate-core = { version = "=0.13.0-dev.202610240000000082" }
 
 # Local crates
 dynamo-runtime = { path = "lib/runtime", version = "1.7.0" }

benchmarks/pyproject.toml

@@ -39,7 +39,7 @@ classifiers = [
 ]
 
 dependencies = [
-    "aisimulate==0.13.0.dev202610160000000073 ; python_version >= '3.11' and python_version < '3.14'",
+    "aisimulate==0.13.0.dev202610240000000082 ; python_version >= '3.11' and python_version < '3.14'",
     "aiperf==0.13.0",
     "matplotlib",
     "networkx",

container/deps/requirements.aisimulate.txt

@@ -3,4 +3,4 @@
 
 # Published AISimulate wheel staged into planner and framework runtime wheelhouses.
 # Keep in sync with the aisimulate-core pins in Cargo.toml and lib/bindings/python/Cargo.toml.
-aisimulate==0.13.0.dev202610160000000073
+aisimulate==0.13.0.dev202610240000000082

lib/bench/multiturn_bench.rs

@@ -609,6 +609,7 @@ async fn main() -> Result<()> {
         },
         seed: args.seed,
         arrival_seed: args.seed,
+        images: None,
     })?;
     let sessions = workload.sessions;
 

lib/bindings/python/Cargo.toml

@@ -60,7 +60,7 @@ request-trace-s3 = ["dynamo-llm/request-trace-s3"]
 # former AIC crate surface. The `python` feature is required for
 # AicEngineBuilder; this extension supplies the interpreter, so the dependency
 # must not enable `extension-module`.
-aisimulate-core = { version = "=0.13.0-dev.202610160000000073", optional = true, features = ["python"] }
+aisimulate-core = { version = "=0.13.0-dev.202610240000000082", optional = true, features = ["python"] }
 dynamo-runtime = { path = "../../runtime" }
 dynamo-mocker = { path = "../../mocker" }
 dynamo-parsers = { version = "=10.0.7" }

lib/bindings/python/rust/llm/replay.rs

@@ -1824,6 +1824,7 @@ fn build_synthetic_workload(
         },
         seed: 42,
         arrival_seed,
+        images: None,
     })
 }
 

lib/mocker/src/generalized_live.rs

@@ -804,6 +804,8 @@ mod tests {
             max_output_tokens: output_len,
             output_token_ids: Some((0..output_len as u32).map(|token| token + 10_000).collect()),
             session_id: None,
+            images: Vec::new(),
+            kv_hint: None,
         }
     }
 
@@ -1944,6 +1946,8 @@ mod tests {
             max_output_tokens: 3,
             output_token_ids: Some(vec![10_000, 10_001, 10_002]),
             session_id: None,
+            images: Vec::new(),
+            kv_hint: None,
         };
 
         // One pass stores both prompts; the FIFO D2H lane finishes the copies

lib/mocker/src/grouped_scheduler.rs

@@ -297,6 +297,10 @@ fn create_grouped_scheduler_from_components(
             .is_none_or(|host| host.scope != G2Scope::ClusterShared),
         "cluster_shared native_host_offload is supported only in offline replay"
     );
+    ensure!(
+        !components.rank.sglang.host_loop,
+        "sglang.host_loop is supported only in offline replay"
+    );
     let factory = engine_factory(components.rank, components.timing)?;
     // Existing live schedulers seed every process-local worker from DP rank.
     // A logical worker therefore retains worker_id=0 at this compatibility
@@ -690,7 +694,7 @@ fn translate_command(
 ) -> Result<TranslatedCommand> {
     let (command, on_success, on_suppressed_output, on_error) = match command {
         SchedulerCommand::Submit(request) => {
-            let request = compatibility.native_request(request);
+            let request = compatibility.native_request(request)?;
             let request_id = request.request_id;
             (
                 Command::Submit(request),
@@ -712,8 +716,8 @@ fn translate_command(
             handoff_id,
             request,
         } => {
+            let request = compatibility.native_request(request)?;
             let engine_handoff = compatibility.engine_handoff(handoff_id)?;
-            let request = compatibility.native_request(request);
             let request_id = request.request_id;
             compatibility.mark_source(handoff_id, request_id);
             (
@@ -749,8 +753,8 @@ fn translate_command(
             handoff_id,
             request,
         } => {
+            let request = compatibility.native_request(request)?;
             let engine_handoff = compatibility.engine_handoff(handoff_id)?;
-            let request = compatibility.native_request(request);
             let request_id = request.request_id;
             compatibility.mark_destination(handoff_id, request_id);
             (

lib/mocker/src/grouped_scheduler/compatibility.rs

@@ -40,20 +40,30 @@ impl CompatibilityState {
         }
     }
 
-    pub(super) fn native_request(&self, request: DirectRequest) -> Request {
+    pub(super) fn native_request(&self, request: DirectRequest) -> Result<Request> {
+        anyhow::ensure!(
+            request.kv_hints.is_none(),
+            "Dynamo grouped live engines do not support authored KV fetch hints"
+        );
+        anyhow::ensure!(
+            request.images.is_empty(),
+            "Dynamo grouped live engines do not support image requests"
+        );
         let request_id = request.uuid.unwrap_or_else(Uuid::new_v4);
         self.request_prompt_lengths
             .lock()
             .insert(request_id, request.tokens.len());
-        Request {
+        Ok(Request {
             request_id,
             tokens: request.tokens,
             max_output_tokens: request.max_output_tokens,
             output_token_ids: request.output_token_ids,
             session_id: request
                 .replay_context
                 .and_then(|context| context.session_id().map(str::to_owned)),
-        }
+            images: Vec::new(),
+            kv_hint: None,
+        })
     }
 
     pub(super) fn engine_handoff(&self, handoff_id: DynamoHandoffId) -> Result<EngineHandoffId> {
@@ -146,6 +156,11 @@ impl CompatibilityState {
                 request_id,
                 transferable_prompt_tokens,
             },
+            LifecycleEvent::TtftMilestone { .. } => {
+                return Err(anyhow!(
+                    "host-aware TTFT milestones are not supported by Dynamo grouped live engines"
+                ));
+            }
         })
     }
 }
@@ -316,14 +331,72 @@ mod tests {
             (Some(" "), None),
             (None, None),
         ] {
-            let request = state.native_request(DirectRequest {
-                replay_context: session_id.map(|id| ReplayRequestContext {
-                    session_id: Some(id.to_owned()),
+            let request = state
+                .native_request(DirectRequest {
+                    replay_context: session_id.map(|id| ReplayRequestContext {
+                        session_id: Some(id.to_owned()),
+                        ..Default::default()
+                    }),
                     ..Default::default()
-                }),
-                ..Default::default()
-            });
+                })
+                .unwrap();
             assert_eq!(request.session_id.as_deref(), expected);
+            assert!(request.images.is_empty());
+            assert!(request.kv_hint.is_none());
         }
     }
+
+    #[test]
+    fn host_aware_ttft_milestones_are_rejected_explicitly() {
+        let state = CompatibilityState::new(MockerConfig::default());
+        let error = state
+            .lifecycle_event(LifecycleEvent::TtftMilestone {
+                request_id: Uuid::from_u128(12),
+                stage: aisimulate_core::engine::TtftMilestone::Received,
+                at_ms: 0.0,
+            })
+            .unwrap_err();
+        assert!(error.to_string().contains("host-aware TTFT milestones"));
+    }
+
+    #[test]
+    fn unhandled_request_fields_do_not_mutate_compatibility_state() {
+        let state = CompatibilityState::new(MockerConfig::default());
+        let mut request = DirectRequest {
+            kv_hints: Some(aisimulate_core::replay::AuthoredKvHints {
+                source_worker: 0,
+                source_dp_rank: 0,
+            }),
+            ..Default::default()
+        };
+        assert!(
+            state
+                .native_request(request.clone())
+                .unwrap_err()
+                .to_string()
+                .contains("authored KV fetch hints")
+        );
+        assert!(state.request_prompt_lengths.lock().is_empty());
+        request.kv_hints = None;
+        request.images.push(aisimulate_core::engine::ImageSpec {
+            identity: 1,
+            token_start: 0,
+            token_end: 1,
+            encoder: aisimulate_core::engine::EncoderShape {
+                sequences: 1,
+                patch_tokens: 1,
+                transformer_tokens: 1,
+                output_tokens: 1,
+            },
+            embedding_bytes: 1,
+        });
+        assert!(
+            state
+                .native_request(request)
+                .unwrap_err()
+                .to_string()
+                .contains("image requests")
+        );
+        assert!(state.request_prompt_lengths.lock().is_empty());
+    }
 }

lib/mocker/src/grouped_scheduler/tests.rs

@@ -17,6 +17,43 @@ use crate::live::{LiveEngine, LiveEngineOptions};
 
 use super::*;
 
+#[test]
+fn host_loop_is_rejected_before_live_tasks_start() {
+    let args = MockerConfig::from_value(serde_json::json!({
+        "engine": {"backend": "sglang", "sglang": {"host_loop": true}}
+    }))
+    .unwrap();
+    let grouped = create_grouped_scheduler(
+        args.clone(),
+        vec![GroupedSchedulerRankSinks::default()],
+        None,
+    );
+    assert!(
+        grouped
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("sglang.host_loop")
+    );
+    let single = create_single_rank_scheduler_with_rank_sink(
+        args,
+        0,
+        RankSinks {
+            output: RankOutputSink::None,
+            kv_event_publishers: KvEventPublishers::default(),
+            fpm_publisher: FpmPublisher::default(),
+        },
+        None,
+    );
+    assert!(
+        single
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("sglang.host_loop")
+    );
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum CapturedEffect {
     KvStored,
@@ -128,7 +165,9 @@ async fn noop_cancellation_only_cleans_metadata_when_output_is_discarded() {
         engine_args.kv_transfer_bytes_per_token = Some(1_000_000);
         let compatibility = CompatibilityState::new(engine_args);
         let request_id = Uuid::from_u128(10 + u128::from(suppressed_pending_output));
-        compatibility.native_request(request(request_id.as_u128(), 0));
+        compatibility
+            .native_request(request(request_id.as_u128(), 0))
+            .unwrap();
 
         let (lifecycle_tx, _lifecycle_rx) = mpsc::channel(1);
         let (metrics_tx, _metrics_rx) = watch::channel(MockerMetrics::new(0, 0, 128));

lib/mocker/src/replay/agentic.rs

@@ -150,7 +150,7 @@ mod tests {
     fn snapshot_sampling_keeps_source_clock_and_requested_lane_count() {
         let build = |speedup| {
             AgenticReplayOptions {
-                snapshot: Some(AgenticSnapshotOptions { seed: 7 }),
+                snapshot: Some(serde_json::from_value(json!({"seed": 7})).unwrap()),
                 arrival_speedup_ratio: speedup,
                 ..Default::default()
             }
@@ -205,7 +205,7 @@ mod tests {
     #[test]
     fn warmup_defers_profile_clock_until_runtime_barrier() {
         let options = AgenticReplayOptions {
-            snapshot: Some(AgenticSnapshotOptions { seed: 7 }),
+            snapshot: Some(serde_json::from_value(json!({"seed": 7})).unwrap()),
             warmup: true,
             profile: Some(AgenticProfileOptions {
                 duration_seconds: 0.5,
@@ -261,7 +261,7 @@ mod tests {
             ),
             (
                 AgenticReplayOptions {
-                    snapshot: Some(AgenticSnapshotOptions { seed: 7 }),
+                    snapshot: Some(serde_json::from_value(json!({"seed": 7})).unwrap()),
                     ..Default::default()
                 },
                 None,
@@ -270,7 +270,7 @@ mod tests {
             ),
             (
                 AgenticReplayOptions {
-                    snapshot: Some(AgenticSnapshotOptions { seed: 7 }),
+                    snapshot: Some(serde_json::from_value(json!({"seed": 7})).unwrap()),
                     profile: Some(AgenticProfileOptions::default()),
                     ..Default::default()
                 },

lib/mocker/src/replay/offline/entrypoints.rs

@@ -148,6 +148,7 @@ fn replay_spec(
         max_in_flight,
         record_per_request,
         sla,
+        encoder: None,
         requests: Vec::new(),
     })
 }

lib/mocker/src/replay/offline/extensions/kv_events/mod.rs

@@ -201,6 +201,7 @@ pub(in crate::replay) fn generate_trace_worker_artifacts_with_visibility(
         max_in_flight: None,
         record_per_request: false,
         sla: Default::default(),
+        encoder: None,
         requests: Vec::new(),
     };
     let visibility = match router_event_visibility_override {