diff --git a/jenkins/scripts/perf/submit.py b/jenkins/scripts/perf/submit.py
index 561f7425e469..72b176df3a95 100755
--- a/jenkins/scripts/perf/submit.py
+++ b/jenkins/scripts/perf/submit.py
@@ -33,9 +33,12 @@
 """
 
 import argparse
+import heapq
+import json
 import math
 import os
 import re
+import shlex
 import sys
 
 import yaml
@@ -63,26 +66,161 @@ def _import_precheck_config(llm_src):
 # --------------------------------------------------------------------------- #
 # Test list parsing
 # --------------------------------------------------------------------------- #
-def parse_test_case_name(test_list_path, llm_src, split_group=0):
-    """Parse the selected line of the test list.
-
-    Returns (config_yaml_path, server_name, benchmark_mode, runtime_mode).
-    See the module docstring for the supported test name shapes.
-    """
+def _read_test_list_lines(test_list_path):
     with open(test_list_path, "r") as f:
-        lines = [line.strip() for line in f if line.strip()]
-
+        lines = []
+        for line in f:
+            stripped_line = line.strip()
+            if stripped_line and not stripped_line.startswith("#"):
+                lines.append(stripped_line)
     if not lines:
         raise ValueError(f"Test list is empty: {test_list_path}")
+    return lines
+
+
+def _pytest_command_tokens(script_prefix_lines):
+    pytest_command_line = next(
+        (line for line in script_prefix_lines if "export pytestCommand=" in line), ""
+    )
+    if not pytest_command_line:
+        return []
+    command = pytest_command_line.split("=", 1)[1].strip()
+    if len(command) >= 2 and command[0] == command[-1] and command[0] in ('"', "'"):
+        command = command[1:-1]
+    return shlex.split(command)
+
+
+def _pytest_option(tokens, option):
+    for index, token in enumerate(tokens):
+        if token == option:
+            return tokens[index + 1] if index + 1 < len(tokens) else None
+        if token.startswith(f"{option}="):
+            return token.split("=", 1)[1]
+    return None
+
+
+def _test_nodeid(test_line):
+    """Strip test-list markers from a line, matching pytest's selected nodeid."""
+    return re.split(
+        r"\s+(?:XFAIL|SKIP|UNSTABLE|TIMEOUT)(?:\s|$)",
+        test_line,
+        maxsplit=1,
+    )[0]
+
+
+def _load_pytest_split_durations(tokens, llm_src):
+    durations_option = _pytest_option(tokens, "--durations-path")
+    if durations_option:
+        durations_path = durations_option
+        if not os.path.exists(durations_path):
+            durations_path = os.path.join(
+                llm_src,
+                "tests",
+                "integration",
+                "defs",
+                os.path.basename(durations_option),
+            )
+    else:
+        durations_path = os.path.join(llm_src, "tests", "integration", "defs", ".test_durations")
+
+    try:
+        with open(durations_path, "r") as durations_file:
+            durations = json.load(durations_file)
+    except FileNotFoundError as error:
+        raise FileNotFoundError(
+            f"pytest-split durations file not found: {durations_path}"
+        ) from error
+    if isinstance(durations, list):
+        durations = dict(durations)
+    if not isinstance(durations, dict):
+        raise ValueError(f"Invalid pytest-split durations file: {durations_path}")
+    return durations, durations_path
+
+
+def _select_least_duration_group(lines, durations, splits, group):
+    """Mirror pytest-split's LeastDurationAlgorithm exactly."""
+    if splits < 1:
+        raise ValueError(f"pytest --splits must be >= 1, got {splits}")
+    if group < 1 or group > splits:
+        raise ValueError(f"pytest --group must be in [1, {splits}], got {group}")
+
+    nodeids = [_test_nodeid(line) for line in lines]
+    relevant_durations = {
+        nodeid: float(durations[nodeid]) for nodeid in nodeids if nodeid in durations
+    }
+    average_duration = (
+        sum(relevant_durations.values()) / len(relevant_durations) if relevant_durations else 1.0
+    )
+    items = [
+        (line, nodeid, relevant_durations.get(nodeid, average_duration), original_index)
+        for original_index, (line, nodeid) in enumerate(zip(lines, nodeids))
+    ]
 
-    if split_group > 0:
+    # pytest-split first sorts by item name, then performs a stable descending
+    # duration sort. It greedily places each item in the least-loaded group.
+    items.sort(key=lambda item: item[1])
+    items.sort(key=lambda item: item[2], reverse=True)
+    selected = [[] for _ in range(splits)]
+    group_heap = [(0.0, group_index) for group_index in range(splits)]
+    heapq.heapify(group_heap)
+    for line, _nodeid, duration, original_index in items:
+        group_duration, group_index = heapq.heappop(group_heap)
+        selected[group_index].append((original_index, line))
+        heapq.heappush(group_heap, (group_duration + duration, group_index))
+
+    return [line for _original_index, line in sorted(selected[group - 1], key=lambda item: item[0])]
+
+
+def select_test_case_line(test_list_path, llm_src, script_prefix_lines, split_group=0):
+    """Select the same test as the pytest-split shard in ``pytestCommand``."""
+    lines = _read_test_list_lines(test_list_path)
+    if split_group <= 0:
+        return lines[0]
+
+    tokens = _pytest_command_tokens(script_prefix_lines)
+    splits_option = _pytest_option(tokens, "--splits")
+    group_option = _pytest_option(tokens, "--group")
+    algorithm = _pytest_option(tokens, "--splitting-algorithm")
+    if splits_option is None or group_option is None:
         if split_group > len(lines):
             raise ValueError(
                 f"split_group {split_group} exceeds number of tests in test list ({len(lines)})"
             )
-        line = lines[split_group - 1]
-    else:
-        line = lines[0]
+        return lines[split_group - 1]
+    if algorithm != "least_duration":
+        raise ValueError(
+            "Multi-node perf launcher only supports pytest-split's least_duration "
+            f"algorithm, got {algorithm!r}"
+        )
+
+    splits = int(splits_option)
+    pytest_group = int(group_option)
+    if pytest_group != split_group:
+        raise ValueError(
+            f"submit.py split_group={split_group} disagrees with pytest --group={pytest_group}"
+        )
+    durations, durations_path = _load_pytest_split_durations(tokens, llm_src)
+    selected = _select_least_duration_group(lines, durations, splits, pytest_group)
+    if len(selected) != 1:
+        raise ValueError(
+            "Multi-node perf launch requires exactly one test in each pytest-split "
+            f"group, but group {pytest_group}/{splits} selected {len(selected)} tests "
+            f"using {durations_path}: {selected}"
+        )
+    print(
+        f"Selected pytest-split group {pytest_group}/{splits} test using "
+        f"{durations_path}: {_test_nodeid(selected[0])}"
+    )
+    return selected[0]
+
+
+def parse_test_case_name(llm_src, selected_line):
+    """Parse the selected test-list line.
+
+    Returns (config_yaml_path, server_name, benchmark_mode, runtime_mode).
+    See the module docstring for the supported test name shapes.
+    """
+    line = selected_line
 
     if "[" not in line or "]" not in line:
         raise ValueError(f"Invalid test list format. Expected name with brackets: {line}")
@@ -484,7 +622,9 @@ def main():
         "--split-group",
         type=int,
         default=0,
-        help="1-indexed split group id. Selects the N-th test from the test list.",
+        help=(
+            "1-indexed pytest-split group id. Selects the same duration-balanced test as pytest."
+        ),
     )
     parser.add_argument("--stage-name", default="", help="Stage name (for logging / GPU detect)")
     parser.add_argument(
@@ -497,19 +637,27 @@ def main():
 
     args = parser.parse_args()
 
+    with open(args.script_prefix, "r") as f:
+        script_prefix_content = f.read()
+    script_prefix_lines = script_prefix_content.split("\n")
+
+    selected_test_line = select_test_case_line(
+        args.test_list,
+        args.llm_src,
+        script_prefix_lines,
+        args.split_group,
+    )
     config_yaml, server_name, benchmark_mode, runtime_mode = parse_test_case_name(
-        args.test_list, args.llm_src, args.split_group
+        args.llm_src,
+        selected_test_line,
     )
 
     with open(config_yaml, "r") as f:
         config = yaml.safe_load(f)
 
-    # Recover test_case_name (the bracketed pytest test id) for the per-test
-    # output dir — same line/split logic as parse_test_case_name.
-    with open(args.test_list, "r") as f:
-        lines = [ln.strip() for ln in f if ln.strip()]
-    sel = lines[args.split_group - 1] if args.split_group > 0 else lines[0]
-    test_case_name = sel.split("[")[-1].split("]")[0] if "[" in sel else ""
+    test_case_name = (
+        selected_test_line.split("[")[-1].split("]")[0] if "[" in selected_test_line else ""
+    )
 
     hardware_config = get_hardware_config(config, runtime_mode, benchmark_mode, server_name)
     env_config = get_env_config(config, runtime_mode, benchmark_mode, server_name)
@@ -522,10 +670,6 @@ def main():
     print(f"Environment configuration: {env_config}")
     print(f"Benchmark configuration: {benchmark_config}")
 
-    with open(args.script_prefix, "r") as f:
-        script_prefix_content = f.read()
-    script_prefix_lines = script_prefix_content.split("\n")
-
     with open(args.srun_args, "r") as f:
         srun_args_content = f.read()
     srun_args_lines = srun_args_content.split()
diff --git a/tests/integration/defs/perf/test_perf_sanity.py b/tests/integration/defs/perf/test_perf_sanity.py
index 4e86d5a34eba..35d1fa872022 100644
--- a/tests/integration/defs/perf/test_perf_sanity.py
+++ b/tests/integration/defs/perf/test_perf_sanity.py
@@ -104,6 +104,10 @@ def ensure_bench_serving_repo() -> str:
 # once EVERY ctx/gen worker has finished model load + autotune + warmup.
 AGG_SERVER_READY_TIMEOUT = 1800
 DISAGG_SERVER_READY_TIMEOUT = 3600
+# GEN workers normally reap within seconds after benchmark_status is written.
+# Keep this well below the whole-test timeout so a stuck multi-node srun cannot
+# turn the optional log-flush synchronization into a pytest/Slurm cancellation.
+GEN_LOG_SENTINEL_TIMEOUT = 120
 
 
 def server_ready_timeout(default: int, mode: str) -> int:
@@ -338,13 +342,14 @@ def parse_gen_worker_device_step_time(
     end-of-file are considered for gen_server_{i}.log — used to slice out a
     single client's iteration segment.
 
-    The log is read exactly once. The caller (DisaggTestCmds.run_cmd) blocks
-    on the gen_server_{i}.done sentinels before calling this, so every gen
-    srun has already exited and its &> aggregate log is fully flushed — there
-    is no partially-written tail to poll for. This replaces the earlier
-    settle-poll heuristic, which could return a mean over a truncated prefix
-    when it accepted the first repeated row count while the log was still
-    flushing across NFS (nvbugs 6487036 / 6487040).
+    The log is read exactly once. The caller (DisaggTestCmds.run_cmd) normally
+    waits for the gen_server_{i}.done sentinels first, so every gen srun has
+    exited and its &> aggregate log is fully flushed. If the dedicated
+    sentinel wait expires, the caller parses the current contents instead of
+    consuming the whole-test timeout; a missing metric then hard-fails before
+    upload. This replaces the earlier settle-poll heuristic, which could
+    accept a truncated prefix while the log was still flushing across NFS
+    (nvbugs 6487036 / 6487040 / 6487038).
     """
     per_file_scans, _total_count = _scan_gen_worker_device_step_time(
         output_dir, num_gen_servers, start_offsets
@@ -1348,7 +1353,11 @@ def wait_for_benchmark_ready(
                 )
             time.sleep(10)
 
-    def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool:
+    def wait_for_gen_log_sentinels(
+        self,
+        timeout: float = GEN_LOG_SENTINEL_TIMEOUT,
+        poll_interval: float = 2.0,
+    ) -> bool:
         """Block until every gen worker signals that its log is fully written.
 
         Each gen worker's srun in slurm_launch_draft.sh redirects all of its
@@ -1357,25 +1366,26 @@ def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool:
         flushed). The benchmark writes benchmark_status *before* calling this,
         which is what lets the gen srun exit — so this is not circular.
 
-        Returns True once all sentinels exist, or False if self.timeout is
-        reached first. On False the caller still parses whatever is on disk:
-        the sentinel is a correctness optimization against reading a
-        mid-flush log (nvbugs 6487036 / 6487040), never a hang risk for CI.
+        Returns True once all sentinels exist, or False if the dedicated
+        sentinel timeout is reached first. On False the caller falls back to
+        parsing the current log contents. The bounded wait prevents a stuck
+        multi-node srun from consuming the whole-test timeout and triggering
+        Slurm's kill-on-bad-exit cascade (nvbugs 6487036 / 6487040 / 6487038).
         """
         sentinels = [
             os.path.join(self.test_output_dir, f"gen_server_{i}.done")
             for i in range(self.num_gen_servers)
         ]
-        start_time = time.time()
+        start_time = time.monotonic()
         while True:
             missing = [p for p in sentinels if not os.path.exists(p)]
             if not missing:
                 print_info("All gen worker log sentinels present; log flush complete.")
                 return True
-            elapsed_time = time.time() - start_time
-            if elapsed_time > self.timeout:
+            elapsed_time = time.monotonic() - start_time
+            if elapsed_time > timeout:
                 print_info(
-                    f"Timeout ({self.timeout}s) waiting for gen worker log "
+                    f"Timeout ({timeout}s) waiting for gen worker log "
                     f"sentinels {missing}; parsing current log contents."
                 )
                 return False
@@ -1384,6 +1394,36 @@ def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool:
             )
             time.sleep(poll_interval)
 
+    def _append_gen_worker_device_step_time(
+        self,
+        pending_device_step_time: List[dict],
+        outputs: List[str],
+    ) -> None:
+        """Wait for GEN log flush, then append each pending client metric.
+
+        A sentinel timeout is a bounded teardown fallback, not a reason to
+        discard metrics that are already present in the GEN logs. If the
+        fallback parse finds no usable metric, check_test_failure still fails
+        the gen_only run before results are uploaded.
+        """
+        if not pending_device_step_time:
+            return
+
+        self.wait_for_gen_log_sentinels()
+        for record in pending_device_step_time:
+            device_step_time_mean = parse_gen_worker_device_step_time(
+                self.test_output_dir,
+                self.num_gen_servers,
+                start_offsets=record["start_offsets"],
+            )
+            if device_step_time_mean is None:
+                continue
+            summary_line = f"Average Per Iter Device Step Time (ms): {device_step_time_mean:.2f}"
+            with open(record["benchmark_file_path"], "a") as benchmark_ctx:
+                benchmark_ctx.write(f"\n{summary_line}\n")
+            idx = record["output_index"]
+            outputs[idx] = f"{outputs[idx]}\n{summary_line}\n"
+
     def get_server_logs(self, server_idx: int) -> List[str]:
         server_logs = []
         for i in range(self.num_ctx_servers):
@@ -1497,13 +1537,16 @@ def run_cmd(self, server_idx: int) -> List[str]:
 
         elif self.disagg_serving_type == "BENCHMARK":
             # Perf-benchmark clients whose gen-worker device step time must be
-            # parsed once the gen logs are flushed. The parse is deferred out of
+            # parsed after the gen-log flush wait. The parse is deferred out of
             # the client loop because gen_server_*.log keeps being written until
             # the gen srun exits, and the gen srun only exits after
             # benchmark_status is written in the finally below. Parsing inside
             # the loop (as before) could read a truncated / not-yet-flushed log
             # and report a wrong mean (nvbugs 6487036 / 6487040).
             pending_device_step_time: List[dict] = []
+            collect_device_step_time = (
+                configs_for_idx is not None and configs_for_idx[2].benchmark_mode == "gen_only"
+            )
             try:
                 disagg_server_hostname, disagg_server_port = (
                     self._get_disagg_server_hostname_and_port(server_idx)
@@ -1535,11 +1578,15 @@ def run_cmd(self, server_idx: int) -> List[str]:
                         )
                         print_info(f"Starting benchmark. cmd is {client_cmd_with_port}")
 
-                        # Snapshot gen_server log sizes so the per-client
-                        # average covers only iterations driven by this client.
-                        gen_log_start_offsets = gen_worker_log_sizes(
-                            self.test_output_dir, self.num_gen_servers
-                        )
+                        # Snapshot gen_server log sizes so the gen_only
+                        # per-client average covers only iterations driven by
+                        # this client. Other modes do not emit this metric and
+                        # must not wait for the GEN teardown sentinel.
+                        gen_log_start_offsets = None
+                        if collect_device_step_time:
+                            gen_log_start_offsets = gen_worker_log_sizes(
+                                self.test_output_dir, self.num_gen_servers
+                            )
 
                         bench_env = copy.deepcopy(os.environ)
                         if client_config:
@@ -1554,16 +1601,17 @@ def run_cmd(self, server_idx: int) -> List[str]:
                             benchmark_ctx.write(output)
 
                         outputs.append(output)
-                        # Defer the gen-worker device-step-time parse until the
-                        # gen logs are flushed (see below); remember where to
-                        # write the summary back.
-                        pending_device_step_time.append(
-                            {
-                                "output_index": len(outputs) - 1,
-                                "benchmark_file_path": benchmark_file_path,
-                                "start_offsets": gen_log_start_offsets,
-                            }
-                        )
+                        if collect_device_step_time:
+                            # Defer the gen-worker device-step-time parse until
+                            # the gen logs are flushed (see below); remember
+                            # where to write the summary back.
+                            pending_device_step_time.append(
+                                {
+                                    "output_index": len(outputs) - 1,
+                                    "benchmark_file_path": benchmark_file_path,
+                                    "start_offsets": gen_log_start_offsets,
+                                }
+                            )
                     else:
                         print_info(
                             f"Skipping perf benchmark for client {client_idx}: "
@@ -1599,27 +1647,12 @@ def run_cmd(self, server_idx: int) -> List[str]:
 
             # benchmark_status is written, so the gen workers can now stop and
             # their srun will exit and drop gen_server_{i}.done. Wait once for
-            # those sentinels (bounded by self.timeout), then parse each
-            # benchmark client's gen-worker device step time a single time: the
-            # flushed log is complete, so no settle polling is needed. Only
-            # gen_only runs emit prev_device_step_time; other modes parse to
-            # None and skip the summary line.
-            if pending_device_step_time:
-                self.wait_for_gen_log_sentinels()
-                for record in pending_device_step_time:
-                    device_step_time_mean = parse_gen_worker_device_step_time(
-                        self.test_output_dir,
-                        self.num_gen_servers,
-                        start_offsets=record["start_offsets"],
-                    )
-                    if device_step_time_mean is not None:
-                        summary_line = (
-                            f"Average Per Iter Device Step Time (ms): {device_step_time_mean:.2f}"
-                        )
-                        with open(record["benchmark_file_path"], "a") as benchmark_ctx:
-                            benchmark_ctx.write(f"\n{summary_line}\n")
-                        idx = record["output_index"]
-                        outputs[idx] = f"{outputs[idx]}\n{summary_line}\n"
+            # those sentinels (bounded independently of the whole-test timeout),
+            # then parse each benchmark client's gen-worker device step time a
+            # single time. A timeout falls back to the current log contents.
+            # Only gen_only runs populate this queue; other modes skip both the
+            # sentinel wait and device-step-time parsing.
+            self._append_gen_worker_device_step_time(pending_device_step_time, outputs)
 
         return outputs
 
diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt
index 647729a6c0da..91815d8f8938 100644
--- a/tests/integration/test_lists/waives.txt
+++ b/tests/integration/test_lists/waives.txt
@@ -380,7 +380,6 @@ perf/test_perf_sanity.py::test_e2e[aggr_upload-super_ad_blackwell-super_ad_ws1_1
 perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb200_deepseek-r1-fp4_128k8k_con128_ctx1_pp8_gen1_dep16_eplb0_mtp1_ccb-NIXL] SKIP (https://nvbugs/6426890)
 perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb200_deepseek-v32-fp4_32k4k_con2048_ctx1_dep4_gen1_dep32_eplb288_mtp1_ccb-NIXL] SKIP (https://nvbugs/6550133)
 perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb200_deepseek-v32-fp4_8k1k_con4096_ctx1_dep4_gen1_dep32_eplb256_mtp0_ccb-NIXL] SKIP (https://nvbugs/6550133)
-perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb300_kimi-k25-thinking-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp0_ccb-NIXL] SKIP (https://nvbugs/6490049)
 perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-r1-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp1_ccb-NIXL] SKIP (https://nvbugs/6550133)
 perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_32k4k_con2048_ctx1_dep4_gen1_dep32_eplb288_mtp1_ccb-NIXL] SKIP (https://nvbugs/6374872)
 perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_8k1k_con1024_ctx1_dep4_gen1_dep32_eplb256_mtp3_ccb-NIXL] SKIP (https://nvbugs/6550133)
diff --git a/tests/unittest/scripts/test_perf_sanity_helpers.py b/tests/unittest/scripts/test_perf_sanity_helpers.py
new file mode 100644
index 000000000000..f344529c2f35
--- /dev/null
+++ b/tests/unittest/scripts/test_perf_sanity_helpers.py
@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("torch._inductor")
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+sys.path.insert(0, str(REPO_ROOT / "tests" / "integration"))
+
+from defs.perf import test_perf_sanity as perf_sanity  # noqa: E402
+
+
+def test_sentinel_timeout_falls_back_to_current_gen_logs(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    benchmark_log = tmp_path / "trtllm-benchmark.0.0.log"
+    benchmark_log.write_text("benchmark output", encoding="utf-8")
+    outputs = ["benchmark output"]
+    pending = [
+        {
+            "output_index": 0,
+            "benchmark_file_path": str(benchmark_log),
+            "start_offsets": [10, 20],
+        }
+    ]
+    commands = perf_sanity.DisaggTestCmds(
+        server_cmds=[],
+        client_cmds={},
+        timeout=1,
+        hostname="localhost",
+        disagg_serving_type="BENCHMARK",
+        num_ctx_servers=1,
+        num_gen_servers=2,
+        output_dir=str(tmp_path),
+        test_output_dir=str(tmp_path),
+    )
+
+    monkeypatch.setattr(
+        perf_sanity.DisaggTestCmds,
+        "wait_for_gen_log_sentinels",
+        lambda self: False,
+    )
+    parse_calls: list[tuple[str, int, list[int]]] = []
+
+    def parse_device_step_time(
+        output_dir: str,
+        num_gen_servers: int,
+        start_offsets: list[int],
+    ) -> float:
+        parse_calls.append((output_dir, num_gen_servers, start_offsets))
+        return 7.25
+
+    monkeypatch.setattr(
+        perf_sanity,
+        "parse_gen_worker_device_step_time",
+        parse_device_step_time,
+    )
+
+    commands._append_gen_worker_device_step_time(pending, outputs)
+
+    assert parse_calls == [(str(tmp_path), 2, [10, 20])]
+    assert outputs == ["benchmark output\nAverage Per Iter Device Step Time (ms): 7.25\n"]
+    assert benchmark_log.read_text(encoding="utf-8").endswith(
+        "\nAverage Per Iter Device Step Time (ms): 7.25\n"
+    )
diff --git a/tests/unittest/scripts/test_perf_submit.py b/tests/unittest/scripts/test_perf_submit.py
index 059b5d792527..ed4fb8dedea7 100644
--- a/tests/unittest/scripts/test_perf_submit.py
+++ b/tests/unittest/scripts/test_perf_submit.py
@@ -15,10 +15,12 @@
 # limitations under the License.
 
 import importlib.util
+import json
 from pathlib import Path
 from types import ModuleType
 
 import pytest
+from pytest_split.algorithms import LeastDurationAlgorithm
 
 REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 SUBMIT_PATHS = (
@@ -28,6 +30,14 @@
 EXAMPLE_SUBMIT_PATH = REPO_ROOT / "examples" / "disaggregated" / "slurm" / "benchmark" / "submit.py"
 
 
+class _FakePytestItem:
+    def __init__(self, nodeid: str) -> None:
+        self.nodeid = nodeid
+
+    def __str__(self) -> str:
+        return self.nodeid
+
+
 def _load_module(path: Path, monkeypatch: pytest.MonkeyPatch) -> ModuleType:
     monkeypatch.syspath_prepend(str(path.parent))
     spec = importlib.util.spec_from_file_location(f"perf_submit_{path.parent.name}", path)
@@ -62,6 +72,11 @@ def test_get_benchmark_config_accepts_positive_integer(submit_module: ModuleType
     assert benchmark_config["concurrency"] == int(concurrency)
 
 
+@pytest.fixture
+def ci_submit_module(monkeypatch: pytest.MonkeyPatch) -> ModuleType:
+    return _load_module(SUBMIT_PATHS[0], monkeypatch)
+
+
 @pytest.mark.parametrize(
     "concurrency",
     (True, 1.5, [], {}, "0", 0, "-1", -1, "1.5", "not-an-integer", None),
@@ -102,3 +117,156 @@ def test_example_worker_environment_exports_positive_concurrency(example_submit_
     )
 
     assert worker_environment["TLLM_BENCHMARK_REQ_QUEUES_SIZE"] == "4301"
+
+
+def test_ci_submit_selects_same_least_duration_shard_as_pytest_split(
+    ci_submit_module: ModuleType,
+    tmp_path: Path,
+) -> None:
+    test_lines = [
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300_deepseek-r1] TIMEOUT (90)",
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300_kimi-k25] TIMEOUT (90)",
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb300_deepseek-r1] TIMEOUT (90)",
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb300_kimi-k25] TIMEOUT (90)",
+    ]
+    test_list_path = tmp_path / "test_list.txt"
+    test_list_path.write_text("\n".join(test_lines), encoding="utf-8")
+
+    durations_dir = tmp_path / "tests" / "integration" / "defs"
+    durations_dir.mkdir(parents=True)
+    durations = {
+        ci_submit_module._test_nodeid(test_lines[0]): 836.268,
+        ci_submit_module._test_nodeid(test_lines[1]): 1462.754,
+        ci_submit_module._test_nodeid(test_lines[2]): 2211.1548,
+        ci_submit_module._test_nodeid(test_lines[3]): 2548.912,
+    }
+    (durations_dir / ".test_durations").write_text(json.dumps(durations), encoding="utf-8")
+    script_prefix_lines = [
+        'export pytestCommand="pytest --splitting-algorithm least_duration '
+        "--splits 4 --group 3 "
+        '--durations-path /remote/tests/integration/defs/.test_durations"'
+    ]
+
+    selected = ci_submit_module.select_test_case_line(
+        test_list_path,
+        tmp_path,
+        script_prefix_lines,
+        split_group=3,
+    )
+
+    assert selected == test_lines[1]
+
+
+def test_ci_submit_selector_matches_installed_pytest_split(
+    ci_submit_module: ModuleType,
+) -> None:
+    lines = [
+        f"perf/test_perf_sanity.py::test_e2e[case-{case_name}] TIMEOUT (90)"
+        for case_name in ("zeta", "alpha", "gamma", "beta", "epsilon", "delta")
+    ]
+    nodeids = [ci_submit_module._test_nodeid(line) for line in lines]
+    items = [_FakePytestItem(nodeid) for nodeid in nodeids]
+    duration_sets = (
+        {nodeid: float(index + 1) for index, nodeid in enumerate(nodeids)},
+        dict.fromkeys(nodeids, 4.0),
+        {nodeids[1]: 8.0, nodeids[4]: 2.0, "irrelevant::test": 1000.0},
+        {},
+    )
+
+    for durations in duration_sets:
+        for splits in (2, 3, 4):
+            expected_groups = LeastDurationAlgorithm()(splits, items, durations)
+            for group, expected_group in enumerate(expected_groups, start=1):
+                selected = ci_submit_module._select_least_duration_group(
+                    lines,
+                    durations,
+                    splits,
+                    group,
+                )
+                assert [ci_submit_module._test_nodeid(line) for line in selected] == [
+                    item.nodeid for item in expected_group.selected
+                ]
+
+
+@pytest.mark.parametrize(
+    ("splits", "group", "match"),
+    (
+        (0, 1, "--splits"),
+        (2, 0, "--group"),
+        (2, 3, "--group"),
+    ),
+)
+def test_ci_submit_rejects_invalid_least_duration_groups(
+    ci_submit_module: ModuleType,
+    splits: int,
+    group: int,
+    match: str,
+) -> None:
+    with pytest.raises(ValueError, match=match):
+        ci_submit_module._select_least_duration_group(
+            ["perf/test_perf_sanity.py::test_e2e[case]"],
+            {},
+            splits,
+            group,
+        )
+
+
+def test_ci_submit_ignores_test_list_comments(
+    ci_submit_module: ModuleType,
+    tmp_path: Path,
+) -> None:
+    test_list_path = tmp_path / "test_list.txt"
+    test_list_path.write_text(
+        "# section comment\n\nperf/test_perf_sanity.py::test_e2e[case]\n",
+        encoding="utf-8",
+    )
+
+    assert ci_submit_module._read_test_list_lines(test_list_path) == [
+        "perf/test_perf_sanity.py::test_e2e[case]"
+    ]
+
+
+def test_ci_submit_rejects_split_group_disagreement(
+    ci_submit_module: ModuleType,
+    tmp_path: Path,
+) -> None:
+    test_list_path = tmp_path / "test_list.txt"
+    test_list_path.write_text(
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300-kimi]\n",
+        encoding="utf-8",
+    )
+    script_prefix_lines = [
+        'export pytestCommand="pytest --splitting-algorithm least_duration --splits 1 --group 1"'
+    ]
+
+    with pytest.raises(ValueError, match="disagrees with pytest --group"):
+        ci_submit_module.select_test_case_line(
+            test_list_path,
+            tmp_path,
+            script_prefix_lines,
+            split_group=2,
+        )
+
+
+def test_ci_submit_rejects_missing_pytest_split_durations(
+    ci_submit_module: ModuleType,
+    tmp_path: Path,
+) -> None:
+    test_list_path = tmp_path / "test_list.txt"
+    test_list_path.write_text(
+        "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300-kimi]\n",
+        encoding="utf-8",
+    )
+    script_prefix_lines = [
+        'export pytestCommand="pytest --splitting-algorithm least_duration '
+        '--splits 1 --group 1 --durations-path /remote/.test_durations"'
+    ]
+
+    expected_path = tmp_path / "tests" / "integration" / "defs" / ".test_durations"
+    with pytest.raises(FileNotFoundError, match=f"durations file not found: {expected_path}"):
+        ci_submit_module.select_test_case_line(
+            test_list_path,
+            tmp_path,
+            script_prefix_lines,
+            split_group=1,
+        )