Harrilee · Oct 22, 2026
@@ -75,6 +75,7 @@ areas:
       - tests/test_agentx_g2.py
       - tests/test_agentx_snapshot_export.py
       - tests/test_agentx_live_warmup.py
+      - tests/test_conversation_routing_cli.py
       - tests/test_runner.py
       - tests/test_traffic.py
       - tests/e2e/
@@ -289,6 +290,8 @@ shared:
     owners: [replay, maintainers]
   - glob: tests/test_agentx_live_warmup.py
     owners: [replay, maintainers]
+  - glob: tests/test_conversation_routing_cli.py
+    owners: [replay, maintainers]
   - glob: tests/test_runner.py
     owners: [replay, maintainers]
   - glob: tests/test_traffic.py
@@ -313,6 +316,8 @@ shared:
     owners: [forward-pass-engine, maintainers]
   - glob: crates/core/src/python.rs
     owners: [mocker, replay, forward-pass-engine, maintainers]
+  - glob: crates/core/src/python/**
+    owners: [mocker, replay, forward-pass-engine, maintainers]
   - glob: Cargo.lock
     owners: [infra, maintainers]
   - glob: Cargo.toml

@@ -84,6 +84,7 @@
 /tests/test_output_adapter_context.py                    @ai-dynamo/aisimulate-replay-codeowners
 /tests/fixtures/power-contract-v1.json                   @ai-dynamo/aisimulate-replay-codeowners
 /tests/test_unified_traffic_runtime.py                   @ai-dynamo/aisimulate-replay-codeowners
+/tests/test_conversation_routing_cli.py                  @ai-dynamo/aisimulate-replay-codeowners
 /tests/test_output_adapter_discovery.py                  @ai-dynamo/aisimulate-replay-codeowners
 
 # === mocker  (@ai-dynamo/aisimulate-mocker-codeowners) ===
@@ -181,6 +182,7 @@
 /tests/test_cli_config.py                                @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_pages_site.py                                @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /crates/core/src/python.rs                               @ai-dynamo/aisimulate-mocker-codeowners @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
+/crates/core/src/python/**                               @ai-dynamo/aisimulate-mocker-codeowners @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/docker/                               @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/uv.lock                               @ai-dynamo/aisimulate-infra-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_unified_cli.py                               @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
@@ -231,6 +233,7 @@
 /tests/fixtures/power-contract-v1.json                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_spline_estimator_config.py                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_unified_traffic_runtime.py                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
+/tests/test_conversation_routing_cli.py                  @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_output_adapter_discovery.py                  @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/src/aisimulate/replay/                @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
 /.claude/skills/aisimulate-fpe-performance               @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain

@@ -1432,3 +1432,38 @@ https://github.com/alibaba/tair-kvcache/blob/85402f1a012fb4279be5bb4b57613bb0a06
 
 The AISimulate files are modified re-implementations, not verbatim copies.
 Both derived files carry the source URLs and revisions in their headers.
+
+## NVIDIA Dynamo offline routing adapter and acceptance
+
+The repository-root file `python/aisimulate/src/aisimulate/router.py` is adapted
+and modified from Dynamo's
+`components/src/dynamo/router/simulation/config.py` and `provider.py` at commit
+`6d6d36d7bfa63d80a0302df637c08f85523aa5b6`:
+https://github.com/ai-dynamo/dynamo/tree/6d6d36d7bfa63d80a0302df637c08f85523aa5b6/components/src/dynamo/router/simulation
+The adaptation retains the configuration vocabulary and field mapping, and
+transports policy calls to the separately installed Dynamo runtime. It does not
+copy the Router, KV index, or session-affinity algorithms into AISimulate.
+The AISim lifecycle and conversation-metadata translation was also adapted and
+modified from the same Apache-2.0 NVIDIA project at
+`2c4be58bdd6a2dc78c409bdfd67ad54d1b6fedf8`, original files
+`lib/bindings/python/rust/llm/offline_router/{affinity,mod,wire}.rs`:
+https://github.com/ai-dynamo/dynamo/tree/2c4be58bdd6a2dc78c409bdfd67ad54d1b6fedf8/lib/bindings/python/rust/llm/offline_router
+
+The repository-root file `tests/test_conversation_routing_cli.py` is adapted and
+modified from Dynamo's
+`components/src/dynamo/replay/tests/e2e/test_unified_cli_dynamo.py` and
+`components/src/dynamo/replay/tests/e2e/configs/unified_cli/predict/dynamo/13-trace-dynamo-agentic.yaml`
+at commit `35ea1e7f689198333b1af0a2db8f94f545817792`:
+https://github.com/ai-dynamo/dynamo/tree/35ea1e7f689198333b1af0a2db8f94f545817792/components/src/dynamo/replay/tests/e2e
+The adaptation runs the conversation fixtures through AISimulate's engine CLI,
+uses the standalone policy API, and rejects dependence on Dynamo's whole-run
+plugin. The synthetic request tree contains no external dataset material.
+
+Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+These source revisions are licensed under Apache-2.0. The full license is
+reproduced in `LICENSE`; the applicable upstream licenses are:
+https://github.com/ai-dynamo/dynamo/blob/6d6d36d7bfa63d80a0302df637c08f85523aa5b6/LICENSE
+https://github.com/ai-dynamo/dynamo/blob/35ea1e7f689198333b1af0a2db8f94f545817792/LICENSE
+https://github.com/ai-dynamo/dynamo/blob/2c4be58bdd6a2dc78c409bdfd67ad54d1b6fedf8/LICENSE
+The source files' copyright/license headers are preserved. These source
+revisions have no root NOTICE file applicable to these files.
@@ -69,6 +69,13 @@ pub struct ForwardPassSpeculationConfig {
 }
 
 impl ForwardPassSpeculationConfig {
+    /// Resolve executor depth through the SDK scheme registry, using the cost
+    /// identity returned by `ForwardPassPerfModel::best_available` provenance.
+    #[cfg(feature = "python")]
+    pub fn replay_depth(&self) -> Result<u32, AicError> {
+        crate::py::speculation_replay_depth(self)
+    }
+
     /// Preserve the existing prompt-lookup contract for canonical and replay callers.
     pub(crate) fn ngram_depth(&self) -> Result<u32, AicError> {
         if self.kind != "ngram" {

@@ -16,7 +16,8 @@ use crate::replay::{
     EncoderSpec, POWER_DATA_COVERAGE_THRESHOLD, ReplayArtifactKvEventVisibility, ReplayArtifacts,
     ReplayCaptureOptions, ReplayDeterminism, ReplayEngineConfig, ReplayEngineFactory,
     ReplayOperationPowerDiagnostics, ReplayPhasePowerDiagnostics, ReplayPowerDiagnostics,
-    ReplayRoleConfig, ReplayRuntimeInput, ReplaySpec, ReplayTopology, Replayer, TracePowerStats,
+    ReplayReport, ReplayRoleConfig, ReplayRuntimeInput, ReplaySpec, ReplayTopology, Replayer,
+    TracePowerStats,
     loadgen::{
         AgenticProfileOptions, AgenticSnapshotOptions, AgenticWarmupPolicy, ArrivalSpec, DelaySpec,
         DynamoRequestTrace, LengthSpec, SyntheticImages, SyntheticTraceSpec, Trace,
@@ -35,6 +36,8 @@ use pyo3::prelude::*;
 use pyo3::types::{PyAny, PyDict, PyModule};
 use serde::Deserialize;
 
+mod external_policy;
+
 #[derive(Debug, Deserialize)]
 #[serde(untagged)]
 enum ExecutionPayload {
@@ -455,7 +458,7 @@ impl AicTimingConfig {
                     speculation.kind
                 ),
             }
-            return Ok(crate::py::speculation_replay_depth(speculation)?);
+            return Ok(speculation.replay_depth()?);
         }
         Ok(self.nextn)
     }
@@ -2616,6 +2619,68 @@ fn replay_fpm_coverage(
 }
 
 fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
+    execute_replay_with_dispatch(payload, capture_artifacts, run_with_input)?.into_json()
+}
+
+/// Shared report serialization for built-in and external placement policies.
+#[derive(Debug)]
+struct ReplayExecutionResult {
+    /// Includes per-request records and execution-owned runtime evidence.
+    report: ReplayReport,
+    /// Single-worker parity artifacts, when requested through the built-in runtime.
+    artifacts: Option<ReplayArtifacts>,
+    /// Additional execution diagnostics and input provenance that are not
+    /// part of the topology runtime's typed report.
+    report_fields: serde_json::Map<String, serde_json::Value>,
+    include_per_request: bool,
+}
+
+impl ReplayExecutionResult {
+    fn into_json(self) -> Result<String> {
+        let mut report_json = serde_json::to_value(&self.report)
+            .context("serializing AISimulate replay report summary")?;
+        let object = report_json
+            .as_object_mut()
+            .context("AISimulate replay report did not serialize as an object")?;
+        object.extend(self.report_fields);
+        if self.include_per_request || !self.report.per_request.is_empty() {
+            // Keep records typed until serialization, avoiding a second full
+            // in-memory JSON tree for large replay reports.
+            object.insert("per_request".to_string(), serde_json::Value::Null);
+        }
+        let report_json = ReportWithRecords {
+            report: &report_json,
+            records: &self.report.per_request,
+        };
+        if let Some(artifacts) = self.artifacts {
+            let output = serde_json::json!({"report": report_json, "artifacts": artifacts});
+            return serde_json::to_string(&output).context("serializing AISimulate replay output");
+        }
+        serde_json::to_string(&report_json).context("serializing AISimulate replay output")
+    }
+}
+
+/// Resolve and finalize a replay while the host selects its execution composition.
+/// The callback receives the same prepared inputs as the built-in executor,
+/// after canonical parsing, timing construction, and workload preparation.
+fn execute_replay_with_dispatch<F>(
+    payload: &str,
+    capture_artifacts: bool,
+    run: F,
+) -> Result<ReplayExecutionResult>
+where
+    F: FnOnce(
+        ReplaySpec,
+        ReplayEngineFactory,
+        Option<ReplayRuntimeInput>,
+        bool,
+        ReplayDeterminism,
+        Option<TokenizationPayload>,
+    ) -> crate::replay::ReplayResult<(
+        crate::replay::ReplayReport,
+        Option<ReplayArtifacts>,
+    )>,
+{
     let (mut spec, mut traffic, capture_performance_diagnostics, determinism, tokenization) =
         match serde_json::from_str(payload).context("invalid AISimulate execution ReplaySpec")? {
             ExecutionPayload::Configured {
@@ -2764,7 +2829,7 @@ fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
             if let Some(timing) = &encoder_timing {
                 factory = factory.with_encoder_timing(Arc::clone(timing));
             }
-            run_with_input(spec, factory, input, capture_artifacts, determinism, tokenization)
+            run(spec, factory, input, capture_artifacts, determinism, tokenization)
                 .map(|(report, artifacts)| (report, artifacts, resolved_basis))
         }
         ReplayTopology::Disaggregated { .. } => {
@@ -2884,7 +2949,7 @@ fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
             if let Some(timing) = &encoder_timing {
                 factory = factory.with_encoder_timing(Arc::clone(timing));
             }
-            run_with_input(spec, factory, input, capture_artifacts, determinism, tokenization)
+            run(spec, factory, input, capture_artifacts, determinism, tokenization)
             .map(|(report, artifacts)| (report, artifacts, resolved_basis))
         }
         }
@@ -2921,19 +2986,23 @@ fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
         timing_evidence.as_ref(),
         unavailable_reason,
     )?));
-    let mut report_json =
-        serde_json::to_value(&report).context("serializing AISimulate replay report summary")?;
-    report_json["prediction_provenance"] = prediction_provenance.into();
+    let mut report_fields = serde_json::Map::new();
+    report_fields.insert(
+        "prediction_provenance".to_string(),
+        prediction_provenance.into(),
+    );
     if let Some(coverage) = replay_fpm_coverage(&coverage_sources, Some(&report))? {
-        report_json["fpm_query_coverage"] = coverage;
+        report_fields.insert("fpm_query_coverage".to_string(), coverage);
     }
     if capture_performance_diagnostics {
         let performance_evidence = replay_timing_evidence_with_fpm(&power_sources, true)?;
-        report_json["performance_diagnostics"] =
-            replay_performance_diagnostics(performance_evidence.as_ref());
+        report_fields.insert(
+            "performance_diagnostics".to_string(),
+            replay_performance_diagnostics(performance_evidence.as_ref()),
+        );
     }
     if let Some(evidence) = replay_fpm_query_evidence(&power_sources)? {
-        report_json["fpm_query_evidence"] = evidence;
+        report_fields.insert("fpm_query_evidence".to_string(), evidence);
     }
     if report.agentic_graph.is_some()
         && let Some((input_format, agentic_lanes, execution_model)) = agentic_input
@@ -2949,29 +3018,26 @@ fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
             .map(str::trim)
             .filter(|model| !model.is_empty())
             .context("agentic execution did not declare its configured target model")?;
-        let object = report_json
-            .as_object_mut()
-            .context("AISimulate replay report did not serialize as an object")?;
-        object.insert(
+        report_fields.insert(
             "agentic_qualification".to_string(),
             serde_json::Value::String("functional_only".to_string()),
         );
-        object.insert(
+        report_fields.insert(
             "agentic_input_format".to_string(),
             serde_json::Value::String(input_format),
         );
-        object.insert(
+        report_fields.insert(
             "agentic_lanes".to_string(),
             serde_json::to_value(agentic_lanes)
                 .context("serializing configured agentic lane count")?,
         );
         if let Some(resolved_basis) = resolved_weka_timestamp_basis {
-            object.insert(
+            report_fields.insert(
                 "weka_nested_timestamp_basis".to_string(),
                 serde_json::Value::String(resolved_basis.as_str().to_string()),
             );
         }
-        object.insert(
+        report_fields.insert(
             "agentic_model_projection".to_string(),
             serde_json::json!({
                 "policy": AGENTIC_MODEL_PROJECTION_POLICY,
@@ -2980,26 +3046,12 @@ fn execute_json(payload: &str, capture_artifacts: bool) -> Result<String> {
             }),
         );
     }
-    if record_per_request || !report.per_request.is_empty() {
-        // Holds the key's place in the map; the records are written there during
-        // serialization, one `Value` at a time.
-        report_json
-            .as_object_mut()
-            .context("AISimulate replay report did not serialize as an object")?
-            .insert("per_request".to_string(), serde_json::Value::Null);
-    }
-    let report_json = ReportWithRecords {
-        report: &report_json,
-        records: &report.per_request,
-    };
-    if let Some(artifacts) = artifacts {
-        let output = serde_json::json!({
-            "report": report_json,
-            "artifacts": artifacts,
-        });
-        return serde_json::to_string(&output).context("serializing AISimulate replay output");
-    }
-    serde_json::to_string(&report_json).context("serializing AISimulate replay output")
+    Ok(ReplayExecutionResult {
+        report,
+        artifacts,
+        report_fields,
+        include_per_request: record_per_request,
+    })
 }
 
 /// The report object with its `per_request` placeholder serialized as the
@@ -3078,6 +3130,18 @@ fn run_replay_json(py: Python<'_>, payload: &str) -> PyResult<String> {
         .map_err(replay_python_error)
 }
 
+/// Execute with an independently installed placement policy while retaining
+/// AISimulate's workload, engine, virtual clock and report ownership.
+#[pyfunction]
+fn run_replay_with_policy_json(
+    py: Python<'_>,
+    payload: &str,
+    factory: Py<PyAny>,
+) -> PyResult<String> {
+    py.allow_threads(|| external_policy::execute(payload, factory))
+        .map_err(replay_python_error)
+}
+
 /// Execute one fixed aggregated ReplaySpec and return report plus parity artifacts.
 #[pyfunction]
 fn run_replay_with_artifacts_json(py: Python<'_>, payload: &str) -> PyResult<String> {
@@ -3144,6 +3208,7 @@ fn _runtime(module: &Bound<'_, PyModule>) -> PyResult<()> {
     )?)?;
     module.add_function(wrap_pyfunction!(canonical_timing_config_json, module)?)?;
     module.add_function(wrap_pyfunction!(run_replay_json, module)?)?;
+    module.add_function(wrap_pyfunction!(run_replay_with_policy_json, module)?)?;
     module.add_function(wrap_pyfunction!(run_replay_with_artifacts_json, module)?)?;
     crate::perfmodel::register_python(module)?;
     Ok(())