From e9ab6afc24fd01cb1f12b31b991d4ef9230c2ea8 Mon Sep 17 00:00:00 2001 From: hongkuanz Date: Fri, 30 Oct 2026 15:48:31 -0700 Subject: [PATCH 1/2] feat(replay): replay DGDR trace Runs with common random numbers DynamoReplayRunner now replays every candidate of a trace sweep with _deterministic=True: ordinal request ids and the canonical router seed. Candidates of one trace then differ only by their configuration, so a candidate's score does not depend on the other candidates, the Run or the parallelism. Live planner retune compares candidates that different DGDR Runs of one window scored on that basis. The synthetic-workload path is unchanged. First layer of the live retune stack split from #17562. Co-Authored-By: Claude Opus 5.5 Signed-off-by: hongkuanz --- components/src/dynamo/replay/simulation.py | 9 ++ .../replay/tests/test_simulation_crn.py | 140 ++++++++++++++++++ 2 files changed, 149 insertions(+) create mode 100644 components/src/dynamo/replay/tests/test_simulation_crn.py diff --git a/components/src/dynamo/replay/simulation.py b/components/src/dynamo/replay/simulation.py index 5b1a5bd49936..ca04ea4b9874 100644 --- a/components/src/dynamo/replay/simulation.py +++ b/components/src/dynamo/replay/simulation.py @@ -409,6 +409,13 @@ def _run_trace( # embedded value instead of imposing the synthetic-workload default. if trace_format not in {"weka", "dynamo"} and trace_block_size is None: trace_block_size = self.trace_block_size + # Common random numbers: every candidate of a trace sweep replays the + # same trace with ordinal request ids and the canonical router seed, + # so candidates differ only by their configuration and a score does + # not depend on the other candidates, the Run or the parallelism. + # Live planner retune compares candidates scored by different DGDR + # Runs of one window on that basis. The synthetic path has no such + # option. if deployment.deployment_mode == "agg": return run_trace_replay( trace_files=trace_files, @@ -423,6 +430,7 @@ def _run_trace( execution_model=execution_model, extra_engine_args=self._engine_args(deployment, "aggregated"), num_workers=deployment.num_workers, + _deterministic=True, **common, ) return run_trace_replay( @@ -440,6 +448,7 @@ def _run_trace( decode_engine_args=self._engine_args(deployment, "decode"), num_prefill_workers=deployment.num_prefill_workers, num_decode_workers=deployment.num_decode_workers, + _deterministic=True, **common, ) diff --git a/components/src/dynamo/replay/tests/test_simulation_crn.py b/components/src/dynamo/replay/tests/test_simulation_crn.py new file mode 100644 index 000000000000..ea45aeabdf05 --- /dev/null +++ b/components/src/dynamo/replay/tests/test_simulation_crn.py @@ -0,0 +1,140 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# ruff: noqa: E402 +# Optional-dependency preflight must run before the simulation imports. + +"""Common random numbers in the Sweeper replay runner. + +Trace replays (what a DGDR trace Run evaluates) are deterministic, so live +planner retune can compare candidates scored by different Runs of one +window; synthetic replays have no such option and must not receive it. +""" + +from __future__ import annotations + +import inspect + +import pytest + +pytest.importorskip( + "aisimulate.sweeper", + reason="AI Simulate is an optional Dynamo simulation dependency", +) + +from aisimulate.sweeper.replay import BackendDeploymentSpec, ReplaySpec + +from dynamo.replay import ReplayReport +from dynamo.replay import api as replay_api +from dynamo.replay import simulation + +pytestmark = [ + pytest.mark.pre_merge, + pytest.mark.unit, + pytest.mark.gpu_0, + pytest.mark.planner, +] + +_REPORT = ReplayReport( + summary={"completed_requests": 1, "output_throughput_tok_s": 1.0}, + per_request=None, + coverage={}, + planner=None, +) + + +def _deployment(mode: str) -> BackendDeploymentSpec: + if mode == "agg": + return BackendDeploymentSpec( + deployment_mode="agg", + backend="vllm", + backend_version="0.11.0", + agg_engine_args={"timing_model": {"type": "polynomial"}}, + num_workers=2, + ) + return BackendDeploymentSpec( + deployment_mode="disagg", + backend="vllm", + backend_version="0.11.0", + prefill_engine_args={ + "worker_type": "prefill", + "timing_model": {"type": "polynomial"}, + }, + decode_engine_args={ + "worker_type": "decode", + "timing_model": {"type": "polynomial"}, + }, + num_prefill_workers=1, + num_decode_workers=2, + ) + + +def _strict_fake(real, seen: dict): + """A replay entry point that binds its kwargs to ``real``'s signature.""" + + signature = inspect.signature(real) + + def fake(*args, **kwargs): + signature.bind(*args, **kwargs) + seen.update(kwargs) + return _REPORT + + return fake + + +@pytest.mark.parametrize("mode", ["agg", "disagg"]) +@pytest.mark.parametrize( + "workload", + [ + {"trace_path": "window.jsonl.gz", "trace_format": "dynamo"}, + {"trace_paths": ["a.jsonl", "b.jsonl"], "trace_format": "mooncake"}, + ], +) +def test_trace_replays_are_deterministic(monkeypatch, mode, workload): + seen: dict = {} + monkeypatch.setattr( + simulation, + "run_trace_replay", + _strict_fake(replay_api.run_trace_replay, seen), + ) + spec = ReplaySpec( + backend_deployment=_deployment(mode), + workload=workload, + goal={"target": "throughput"}, + ) + + simulation.DynamoReplayRunnerFactory().create(0).run(spec) + + assert seen["_deterministic"] is True + + +@pytest.mark.parametrize("mode", ["agg", "disagg"]) +def test_synthetic_replays_do_not_receive_the_flag(monkeypatch, mode): + # The synthetic entry point has no such parameter; the strict fake fails + # if the flag leaks into the shared kwargs. + assert ( + "_deterministic" + not in inspect.signature(replay_api.run_synthetic_trace_replay).parameters + ) + seen: dict = {} + monkeypatch.setattr( + simulation, + "run_synthetic_trace_replay", + _strict_fake(replay_api.run_synthetic_trace_replay, seen), + ) + spec = ReplaySpec( + backend_deployment=_deployment(mode), + workload={ + "trace_path": None, + "isl": 512, + "osl": 128, + "num_request_ratio": 10.0, + "concurrency": None, + "request_rate": 20.0, + }, + goal={"target": "throughput"}, + ) + + simulation.DynamoReplayRunnerFactory().create(0).run(spec) + + assert seen # the synthetic path ran + assert "_deterministic" not in seen From fd45a8592be81a07589051e8c93b914fb1418bb5 Mon Sep 17 00:00:00 2001 From: hongkuanz Date: Mon, 2 Nov 2026 08:26:41 -0800 Subject: [PATCH 2/2] fix(replay): address review on trace replay common random numbers Fold the trace replay determinism check into test_trace_runner_preserves_current_replay_arguments with an agg and a disagg row, and drop the separate test file. Cut the runner comment to the invariant, and update replay/api.py: _deterministic now applies to every trace replay through the Sweeper's Dynamo runner, not only to retune sweeps. Co-Authored-By: Claude Opus 5.5 Signed-off-by: hongkuanz --- components/src/dynamo/replay/api.py | 7 +- components/src/dynamo/replay/simulation.py | 10 +- .../dynamo/replay/tests/test_simulation.py | 35 ++++- .../replay/tests/test_simulation_crn.py | 140 ------------------ 4 files changed, 39 insertions(+), 153 deletions(-) delete mode 100644 components/src/dynamo/replay/tests/test_simulation_crn.py diff --git a/components/src/dynamo/replay/api.py b/components/src/dynamo/replay/api.py index 0ac237c70b99..baf1a285538a 100644 --- a/components/src/dynamo/replay/api.py +++ b/components/src/dynamo/replay/api.py @@ -97,7 +97,7 @@ class _TraceReplayOptions(_CommonReplayOptions, total=False): trace_num_prefix_groups: int report_jsonl_path: str | os.PathLike[str] | None max_sim_time_ms: float | None - # Internal to retune sweeps; not part of the public API. + # Internal; not part of the public API. _planner_warmup_observations: "Sequence[TrafficObservation] | None" _deterministic: bool @@ -420,8 +420,9 @@ def run_trace_replay( # Internal: a retune sweep's inner planner is warmed with the history it # already bucketed instead of its load_predictor_warmup_trace. _planner_warmup_observations=None, - # Internal: a retune sweep replays the baseline and every candidate with - # the canonical router seed and ordinal request ids (common random numbers). + # Internal: replay with the canonical router seed and ordinal request ids + # (common random numbers). Retune sweeps and every trace replay through + # the Sweeper's Dynamo runner set it. _deterministic=False, ) -> ReplayReport | dict[str, Any]: """Run trace replay. diff --git a/components/src/dynamo/replay/simulation.py b/components/src/dynamo/replay/simulation.py index ca04ea4b9874..251e033199b4 100644 --- a/components/src/dynamo/replay/simulation.py +++ b/components/src/dynamo/replay/simulation.py @@ -409,13 +409,9 @@ def _run_trace( # embedded value instead of imposing the synthetic-workload default. if trace_format not in {"weka", "dynamo"} and trace_block_size is None: trace_block_size = self.trace_block_size - # Common random numbers: every candidate of a trace sweep replays the - # same trace with ordinal request ids and the canonical router seed, - # so candidates differ only by their configuration and a score does - # not depend on the other candidates, the Run or the parallelism. - # Live planner retune compares candidates scored by different DGDR - # Runs of one window on that basis. The synthetic path has no such - # option. + # Trace replays use ordinal request ids and the canonical router seed, + # so a candidate's score depends only on its configuration and the + # trace. if deployment.deployment_mode == "agg": return run_trace_replay( trace_files=trace_files, diff --git a/components/src/dynamo/replay/tests/test_simulation.py b/components/src/dynamo/replay/tests/test_simulation.py index 4c7c7b3b7cea..23725992c6d5 100644 --- a/components/src/dynamo/replay/tests/test_simulation.py +++ b/components/src/dynamo/replay/tests/test_simulation.py @@ -97,7 +97,35 @@ def _agg_deployment() -> BackendDeploymentSpec: ) -def test_trace_runner_preserves_current_replay_arguments(monkeypatch) -> None: +def _disagg_deployment() -> BackendDeploymentSpec: + return BackendDeploymentSpec( + deployment_mode="disagg", + backend="vllm", + backend_version="0.11.0", + prefill_engine_args={ + "worker_type": "prefill", + "timing_model": {"type": "polynomial"}, + }, + decode_engine_args={ + "worker_type": "decode", + "timing_model": {"type": "polynomial"}, + }, + num_prefill_workers=1, + num_decode_workers=2, + ) + + +@pytest.mark.parametrize( + ("deployment", "workers"), + [ + (_agg_deployment(), {"num_workers": 3}), + (_disagg_deployment(), {"num_prefill_workers": 1, "num_decode_workers": 2}), + ], + ids=["agg", "disagg"], +) +def test_trace_runner_preserves_current_replay_arguments( + monkeypatch, deployment, workers +) -> None: seen = {} def fake_run_trace_replay(**kwargs): @@ -113,7 +141,7 @@ def fake_run_trace_replay(**kwargs): monkeypatch.setattr(simulation, "KvRouterConfig", _FakeRouterConfig) monkeypatch.setattr(simulation, "run_trace_replay", fake_run_trace_replay) spec = ReplaySpec( - backend_deployment=_agg_deployment(), + backend_deployment=deployment, workload={ "trace_path": "tiny.jsonl", "trace_format": "dynamo", @@ -159,7 +187,8 @@ def fake_run_trace_replay(**kwargs): assert seen["trace_files"] == "tiny.jsonl" assert seen["trace_format"] == "dynamo" - assert seen["num_workers"] == 3 + assert {name: seen[name] for name in workers} == workers + assert seen["_deterministic"] is True assert seen["router_mode"] == "kv_router" assert seen["planner_config"] == {"mode": "agg"} assert seen["arrival_speedup_ratio"] == 2.0 diff --git a/components/src/dynamo/replay/tests/test_simulation_crn.py b/components/src/dynamo/replay/tests/test_simulation_crn.py deleted file mode 100644 index ea45aeabdf05..000000000000 --- a/components/src/dynamo/replay/tests/test_simulation_crn.py +++ /dev/null @@ -1,140 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# ruff: noqa: E402 -# Optional-dependency preflight must run before the simulation imports. - -"""Common random numbers in the Sweeper replay runner. - -Trace replays (what a DGDR trace Run evaluates) are deterministic, so live -planner retune can compare candidates scored by different Runs of one -window; synthetic replays have no such option and must not receive it. -""" - -from __future__ import annotations - -import inspect - -import pytest - -pytest.importorskip( - "aisimulate.sweeper", - reason="AI Simulate is an optional Dynamo simulation dependency", -) - -from aisimulate.sweeper.replay import BackendDeploymentSpec, ReplaySpec - -from dynamo.replay import ReplayReport -from dynamo.replay import api as replay_api -from dynamo.replay import simulation - -pytestmark = [ - pytest.mark.pre_merge, - pytest.mark.unit, - pytest.mark.gpu_0, - pytest.mark.planner, -] - -_REPORT = ReplayReport( - summary={"completed_requests": 1, "output_throughput_tok_s": 1.0}, - per_request=None, - coverage={}, - planner=None, -) - - -def _deployment(mode: str) -> BackendDeploymentSpec: - if mode == "agg": - return BackendDeploymentSpec( - deployment_mode="agg", - backend="vllm", - backend_version="0.11.0", - agg_engine_args={"timing_model": {"type": "polynomial"}}, - num_workers=2, - ) - return BackendDeploymentSpec( - deployment_mode="disagg", - backend="vllm", - backend_version="0.11.0", - prefill_engine_args={ - "worker_type": "prefill", - "timing_model": {"type": "polynomial"}, - }, - decode_engine_args={ - "worker_type": "decode", - "timing_model": {"type": "polynomial"}, - }, - num_prefill_workers=1, - num_decode_workers=2, - ) - - -def _strict_fake(real, seen: dict): - """A replay entry point that binds its kwargs to ``real``'s signature.""" - - signature = inspect.signature(real) - - def fake(*args, **kwargs): - signature.bind(*args, **kwargs) - seen.update(kwargs) - return _REPORT - - return fake - - -@pytest.mark.parametrize("mode", ["agg", "disagg"]) -@pytest.mark.parametrize( - "workload", - [ - {"trace_path": "window.jsonl.gz", "trace_format": "dynamo"}, - {"trace_paths": ["a.jsonl", "b.jsonl"], "trace_format": "mooncake"}, - ], -) -def test_trace_replays_are_deterministic(monkeypatch, mode, workload): - seen: dict = {} - monkeypatch.setattr( - simulation, - "run_trace_replay", - _strict_fake(replay_api.run_trace_replay, seen), - ) - spec = ReplaySpec( - backend_deployment=_deployment(mode), - workload=workload, - goal={"target": "throughput"}, - ) - - simulation.DynamoReplayRunnerFactory().create(0).run(spec) - - assert seen["_deterministic"] is True - - -@pytest.mark.parametrize("mode", ["agg", "disagg"]) -def test_synthetic_replays_do_not_receive_the_flag(monkeypatch, mode): - # The synthetic entry point has no such parameter; the strict fake fails - # if the flag leaks into the shared kwargs. - assert ( - "_deterministic" - not in inspect.signature(replay_api.run_synthetic_trace_replay).parameters - ) - seen: dict = {} - monkeypatch.setattr( - simulation, - "run_synthetic_trace_replay", - _strict_fake(replay_api.run_synthetic_trace_replay, seen), - ) - spec = ReplaySpec( - backend_deployment=_deployment(mode), - workload={ - "trace_path": None, - "isl": 512, - "osl": 128, - "num_request_ratio": 10.0, - "concurrency": None, - "request_rate": 20.0, - }, - goal={"target": "throughput"}, - ) - - simulation.DynamoReplayRunnerFactory().create(0).run(spec) - - assert seen # the synthetic path ran - assert "_deterministic" not in seen