From e5a3c9018f4bb960970c6bdb57020c29460a608c Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Fri, 7 Aug 2026 10:27:34 -0700
Subject: [PATCH 1/8] First draft

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 ...ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml |   7 +
 .../defs/disaggregated/test_disaggregated.py  | 186 ++++++++++++++++--
 tests/integration/test_lists/waives.txt       |   1 -
 3 files changed, 175 insertions(+), 19 deletions(-)

diff --git a/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml b/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml
index 189c85a55ec0..33584a4e45dc 100644
--- a/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml
+++ b/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml
@@ -18,6 +18,13 @@ context_servers:
     dtype: fp8
   moe_config:
     backend: TRTLLM
+    # Chunk the MoE forward at half of max_num_tokens: the TRTLLM-Gen FP4
+    # MoE workspace for a full 16640-token chunk is a 7-8 GiB transient,
+    # which exceeds the headroom left after weights (~107.5 GiB/GPU) and the
+    # KV pool on 192GB B200 and intermittently OOMs the ctx worker under
+    # 512-concurrency 8k prefill load (memory estimation only observes
+    # ~10.8 GiB dynamic peak, so the KV pool leaves no slack for it).
+    max_num_tokens: 8320
   cuda_graph_config: null
   print_iter_log: true
   cache_transceiver_config:
diff --git a/tests/integration/defs/disaggregated/test_disaggregated.py b/tests/integration/defs/disaggregated/test_disaggregated.py
index c0147c8423b4..7aa9b70e8472 100644
--- a/tests/integration/defs/disaggregated/test_disaggregated.py
+++ b/tests/integration/defs/disaggregated/test_disaggregated.py
@@ -23,7 +23,7 @@
 import subprocess
 import tempfile
 import time
-from collections import namedtuple
+from collections import deque, namedtuple
 from dataclasses import dataclass
 from typing import Any, Optional
 
@@ -129,6 +129,49 @@ def scan_logs_for_fatal_errors(processes):
     return findings
 
 
+def print_first_fatal_log_context(processes, context_lines=20):
+    """Print the lines around the FIRST fatal-pattern match in each log.
+
+    The last-N-lines tail printed on failure usually shows only post-crash
+    shutdown spam ("LLM is shutting down" storms); the root cause — e.g. the
+    OOM traceback with the allocation site — is at the first match, often
+    thousands of lines earlier. Streams each log to avoid loading multi-GB
+    worker logs into memory.
+    """
+    for proc in processes:
+        log_path = getattr(proc, "log_path", None)
+        if not log_path or not os.path.exists(log_path):
+            continue
+        before = deque(maxlen=context_lines)
+        after = []
+        matched = None
+        try:
+            with open(log_path, "r", errors="replace") as f:
+                for lineno, line in enumerate(f, 1):
+                    if matched is None:
+                        pat = next(
+                            (p for p in _FATAL_LOG_PATTERNS if p in line), None)
+                        if pat is None:
+                            before.append(line)
+                            continue
+                        matched = (lineno, pat)
+                        after.append(line)
+                    else:
+                        after.append(line)
+                        if len(after) > context_lines:
+                            break
+        except OSError:
+            continue
+        if matched is None:
+            continue
+        lineno, pat = matched
+        logger.error(f"-------- {log_path}: first fatal pattern '{pat}' at "
+                     f"line {lineno} (+/-{context_lines} lines) --------")
+        for line in [*before, *after]:
+            if line.strip():
+                logger.error(line.rstrip())
+
+
 def _crashed_workers(workers):
     return [
         w for w in workers
@@ -165,6 +208,20 @@ def get_default_disagg_cluster_config():
     }
 
 
+# Production service-discovery timings, matching the DisaggClusterConfig
+# defaults in tensorrt_llm/llmapi/disagg_utils.py. The tight 1s/2s defaults
+# above keep short functional tests snappy, but at stress-level concurrency
+# they leave <1s of heartbeat slack while the worker heartbeat task, the
+# cluster-storage /expire handler, and the expiry sweep all share event loops
+# saturated by request traffic — so workers get spuriously expired and the
+# router flaps "Cluster is not ready" (nvbugs/6472256). Stress runners must
+# use these production values instead.
+PRODUCTION_CLUSTER_TIMINGS = {
+    "heartbeat_interval_sec": 5,
+    "inactive_timeout_sec": 10,
+}
+
+
 def build_worker_config(base_config: dict[str, Any],
                         server_type_config: dict[str, Any],
                         disagg_cluster: dict[str, Any]) -> dict[str, Any]:
@@ -657,6 +714,7 @@ def setup_disagg_cluster(
     startup_callback=None,
     startup_tick: int = 30,
     perf_metrics_output_dir: str | None = None,
+    disagg_cluster_overrides: dict[str, Any] | None = None,
 ) -> tuple[dict[str, Any], list[ProcessWrapper], list[ProcessWrapper],
            ProcessWrapper, int, str]:
     """Load config, launch workers + disagg server, wait for ready.
@@ -667,6 +725,9 @@ def setup_disagg_cluster(
         env: Environment variables to pass to subprocess (workers and disagg server)
         server_start_timeout: Timeout in seconds for server to become ready
         schedule_style: Disagg schedule style ('context_first' or 'generation_first')
+        disagg_cluster_overrides: Entries merged over
+            get_default_disagg_cluster_config(), e.g. PRODUCTION_CLUSTER_TIMINGS
+            for stress runs (cluster_uri/minimal_instances are still derived below)
 
     Returns:
         tuple: (config, ctx_workers, gen_workers, disagg_server, server_port, work_dir)
@@ -689,6 +750,8 @@ def setup_disagg_cluster(
                 speculative_model)
 
     disagg_cluster = get_default_disagg_cluster_config()
+    if disagg_cluster_overrides:
+        disagg_cluster.update(disagg_cluster_overrides)
     server_host = config.get("hostname", "localhost")
     server_port = get_free_port()
     if save_log:
@@ -2308,6 +2371,69 @@ def get_config_for_benchmark(model_root, backend):
     return serve_config
 
 
+def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
+    """Fail if the fraction of non-cancellation request errors exceeds max_error_rate.
+
+    aiperf exits 0 and counts HTTP 500s as completed requests, so without this
+    gate a mid-run server error storm (e.g. "Cluster is not ready" readiness
+    flapping, nvbugs/6472256) passes silently. Reads aiperf's per-record export
+    (profile_export.jsonl in artifact_dir), where each line carries an optional
+    "error" object with code/type/message. Intentional client-side cancellations
+    (HTTP 499 / RequestCancellationError) are excluded from both the numerator
+    and the denominator — stress tests cancel a fraction of requests on purpose.
+    """
+    export_path = os.path.join(artifact_dir, "profile_export.jsonl")
+    assert os.path.exists(export_path), (
+        f"aiperf per-record export not found at {export_path}; cannot enforce "
+        "the request error-rate gate. If this aiperf version/export level does "
+        "not produce it, pass max_error_rate=None explicitly.")
+    total = 0
+    cancelled = 0
+    # (code, type) -> [count, example message]
+    error_counts: dict[tuple, list] = {}
+    with open(export_path, "r", errors="replace") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                record = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            total += 1
+            error = record.get("error")
+            if not error:
+                continue
+            code = error.get("code")
+            err_type = error.get("type")
+            if code == 499 or err_type == "RequestCancellationError":
+                cancelled += 1
+                continue
+            entry = error_counts.setdefault((code, err_type),
+                                            [0, error.get("message", "")])
+            entry[0] += 1
+    considered = total - cancelled
+    if considered <= 0:
+        return
+    errors = sum(count for count, _ in error_counts.values())
+    error_rate = errors / considered
+    print(
+        f"[aiperf-gate] non-cancellation errors: {errors}/{considered} "
+        f"({error_rate:.2%}), cancelled: {cancelled}, "
+        f"threshold: {max_error_rate:.2%}",
+        flush=True)
+    if error_rate > max_error_rate:
+        breakdown = "\n".join(
+            f"  code={code} type={err_type}: {count} (e.g. {message[:200]})"
+            for (code, err_type), (count, message) in sorted(
+                error_counts.items(), key=lambda kv: -kv[1][0]))
+        raise AssertionError(
+            f"aiperf request error rate {error_rate:.2%} exceeds threshold "
+            f"{max_error_rate:.2%} ({errors} non-cancellation errors out of "
+            f"{considered} requests; {cancelled} intentional cancellations "
+            f"excluded). Error breakdown:\n{breakdown}")
+
+
 def run_disaggregated_aiperf(config_file,
                              model_path,
                              server_start_timeout=1200,
@@ -2325,6 +2451,7 @@ def run_disaggregated_aiperf(config_file,
                              threshold=0.8,
                              cancellation_rate=None,
                              cancellation_delay=None,
+                             max_error_rate=0.05,
                              env=None,
                              cwd=None):
     """Run disaggregated test with genai-perf for performance/stress testing.
@@ -2343,6 +2470,8 @@ def run_disaggregated_aiperf(config_file,
         random_seed: Random seed for reproducibility
         accuracy_test: Whether to run accuracy test
         threshold: Threshold for accuracy test
+        max_error_rate: Fail if the fraction of non-cancellation request
+            errors recorded by aiperf exceeds this (None disables the gate)
         env: Environment variables dict
         cwd: Working directory
     """
@@ -2354,7 +2483,8 @@ def run_disaggregated_aiperf(config_file,
     config, ctx_workers, gen_workers, disagg_server, server_port, work_dir = \
         setup_disagg_cluster(config_file, model_name=model_path, env=run_env, cwd=cwd,
                              server_start_timeout=server_start_timeout,
-                             save_log=True)
+                             save_log=True,
+                             disagg_cluster_overrides=PRODUCTION_CLUSTER_TIMINGS)
 
     server_host = config.get("hostname", "localhost")
     artifact_dir = os.path.join(cwd or ".", "benchmark-results")
@@ -2453,6 +2583,13 @@ def run_disaggregated_aiperf(config_file,
                 "Fatal error patterns detected in disaggregated worker/server "
                 f"logs:\n{summary}")
 
+        # Gate on the per-request error rate from aiperf's record export:
+        # aiperf exits 0 even when the server returns 500s for a large share
+        # of requests, so this is the only check that catches a mid-run error
+        # storm on this path (the fatal-pattern scan above is hang/OOM only).
+        if max_error_rate is not None:
+            enforce_aiperf_error_rate(artifact_dir, max_error_rate)
+
         if accuracy_test:
             accuracy_test_result, accuracy_value = run_accuracy_test(
                 model_path=model_path,
@@ -2489,14 +2626,17 @@ def run_disaggregated_aiperf(config_file,
                     f"worker/server logs after accuracy run:\n{summary}")
 
     except Exception:
-        # Print tail of each captured worker/server log to aid triage.
+        # Print the context around the first fatal-pattern match (the root
+        # cause, e.g. an OOM traceback) and the tail of each captured
+        # worker/server log to aid triage.
+        print_first_fatal_log_context(
+            [*ctx_workers, *gen_workers, disagg_server])
         for proc in [*ctx_workers, *gen_workers, disagg_server]:
             log_path = getattr(proc, "log_path", None)
             if not log_path or not os.path.exists(log_path):
                 continue
             logger.error(f"-------- {log_path} (last 30 lines) --------")
             try:
-                from collections import deque
                 with open(log_path, "r", errors="replace") as f:
                     for line in deque(f, maxlen=30):
                         if line.strip():
@@ -2676,13 +2816,20 @@ def test_llama4_long_context_kv_cache_overflow(disaggregated_test_root,
                                   disaggregated_example_root,
                                   os.path.dirname(__file__))
 
-    run_disaggregated_aiperf(config_file=config_file,
-                             model_path=llama4_model_root,
-                             server_start_timeout=1200,
-                             input_tokens=128000,
-                             output_tokens=100,
-                             env=llm_venv._new_env,
-                             cwd=llm_venv.get_working_directory())
+    run_disaggregated_aiperf(
+        config_file=config_file,
+        model_path=llama4_model_root,
+        server_start_timeout=1200,
+        input_tokens=128000,
+        output_tokens=100,
+        # This repro intentionally degrades the KV
+        # transfer path (tiny max_tokens_in_buffer vs
+        # 128k inputs), so sporadic request errors are
+        # by-design; keep the test scoped to its
+        # original crash/fatal-log checks.
+        max_error_rate=None,
+        env=llm_venv._new_env,
+        cwd=llm_venv.get_working_directory())
 
 
 @skip_pre_blackwell
@@ -3446,9 +3593,10 @@ async def _warmup_requests(server_url: str, profiles: list, count: int,
     The first request of each shape pays a one-time autotuner/compile cost
     (~20s host-steps observed on B200). Running those here, before the measured
     run, keeps them out of the accuracy/incomplete accounting and out of the
-    heartbeat-eviction path (a worker stuck in a 20s step misses the 2s cluster
-    heartbeat and gets evicted under a high-concurrency flood). Failures are
-    ignored — the only goal is to trigger the autotuner across the profile mix.
+    heartbeat-eviction path (a worker stuck in a 20s step exceeds even the 10s
+    PRODUCTION_CLUSTER_TIMINGS inactive timeout and gets evicted under a
+    high-concurrency flood). Failures are ignored — the only goal is to
+    trigger the autotuner across the profile mix.
     """
     import random
 
@@ -3523,7 +3671,8 @@ def run_disaggregated_mixed_stress(example_dir: str,
     config, ctx_workers, gen_workers, disagg_server, server_port, work_dir = \
         setup_disagg_cluster(config_file, model_name=model_path, env=run_env,
                              cwd=cwd, server_start_timeout=server_start_timeout,
-                             save_log=True, startup_callback=startup_callback)
+                             save_log=True, startup_callback=startup_callback,
+                             disagg_cluster_overrides=PRODUCTION_CLUSTER_TIMINGS)
     print(f"[startup] cluster ready in {time.monotonic() - setup_start:.1f}s",
           flush=True)
 
@@ -3540,9 +3689,10 @@ def run_disaggregated_mixed_stress(example_dir: str,
         # Pay the one-time autotuner cost before the measured run. The first
         # request of each shape triggers a ~20s autotuner host-step; left in
         # the measured run at high concurrency, a worker stuck in that step
-        # misses the 2s cluster heartbeat and is evicted mid-run, causing
-        # "Cluster is not ready" 500s. Default count = ~20s at the ~6 req/s
-        # observed in the 5k B200 run; results are discarded.
+        # exceeds even the 10s PRODUCTION_CLUSTER_TIMINGS inactive timeout
+        # and is evicted mid-run, causing "Cluster is not ready" 500s.
+        # Default count = ~20s at the ~6 req/s observed in the 5k B200 run;
+        # results are discarded.
         if warmup_request_count is None:
             warmup_request_count = 120
         print(
diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt
index 21f7011ea80e..60a8f730392b 100644
--- a/tests/integration/test_lists/waives.txt
+++ b/tests/integration/test_lists/waives.txt
@@ -163,7 +163,6 @@ full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpu
 full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpus_block_reuse[ADP4_MTP] SKIP (https://nvbugs/6525008)
 full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpus_block_reuse[TEP4] SKIP (https://nvbugs/6474894)
 full:B200/accuracy/test_llm_api_pytorch.py::TestQwen3_30B_A3B::test_dummy_load_format SKIP (https://nvbugs/6525059)
-full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-deepseek_r1_v2_fp4_stress] SKIP (https://nvbugs/6472256)
 full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-glm5_nvfp4_tp4_ep4_dp_stress] SKIP (https://nvbugs/6544407)
 full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-qwen3_32b_fp8_stress] SKIP (https://nvbugs/6472256)
 full:B200/llmapi/test_llm_api_pytorch_moe_lora.py::test_qwen_moe_routed_expert_multi_lora_varying_ranks[cudagraph] SKIP (https://nvbugs/6475623)

From 04df3512a4b1ca8180828ad1126a941bd81f7cd9 Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 15:44:45 -0700
Subject: [PATCH 2/8] [https://nvbugs/6472256][fix] Raise FileNotFoundError for
 missing aiperf export instead of assert

Addresses CodeRabbit review on #17427: assert statements are stripped
under python -O, which would defer a missing-export failure to open()
without the intended diagnostic.

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 .../defs/disaggregated/test_disaggregated.py         | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/tests/integration/defs/disaggregated/test_disaggregated.py b/tests/integration/defs/disaggregated/test_disaggregated.py
index 7aa9b70e8472..1cf46fdde0b5 100644
--- a/tests/integration/defs/disaggregated/test_disaggregated.py
+++ b/tests/integration/defs/disaggregated/test_disaggregated.py
@@ -2383,10 +2383,14 @@ def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
     and the denominator — stress tests cancel a fraction of requests on purpose.
     """
     export_path = os.path.join(artifact_dir, "profile_export.jsonl")
-    assert os.path.exists(export_path), (
-        f"aiperf per-record export not found at {export_path}; cannot enforce "
-        "the request error-rate gate. If this aiperf version/export level does "
-        "not produce it, pass max_error_rate=None explicitly.")
+    # Explicit raise (not assert): asserts are stripped under python -O, which
+    # would defer the failure to open() without this diagnostic.
+    if not os.path.exists(export_path):
+        raise FileNotFoundError(
+            f"aiperf per-record export not found at {export_path}; cannot "
+            "enforce the request error-rate gate. If this aiperf "
+            "version/export level does not produce it, pass "
+            "max_error_rate=None explicitly.")
     total = 0
     cancelled = 0
     # (code, type) -> [count, example message]

From 94bedd3ba9aaa4b66fae0a4b03382e1c6be81954 Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 15:53:49 -0700
Subject: [PATCH 3/8] [https://nvbugs/6472256][fix] Harden aiperf error-rate
 gate; chunk MoE in MTP config too

Addresses brnguyen2's review on #17427:
- The gate no longer degrades into a silent pass on a broken export:
  empty exports, wholesale JSON parse failures, all-cancelled record
  sets, and record counts far below the requested load now raise, with
  decode failures counted and reported in the [aiperf-gate] line.
- Docstring pins the validated record schema to aiperf==0.8.0 (the
  requirements-dev.txt pin) and records the reference gate output from
  the 35000-request DeepSeek R1 FP4 validation run
  (0/31561 non-cancellation errors, cancelled: 3439 vs ~3500 expected).
- GPU-free unit tests (test_aiperf_gate.py) feed the gate synthetic
  profile_export.jsonl cases: the nvbugs/6472256 500-storm replay
  (fires), healthy run with cancellations (passes), and the
  empty/corrupt/all-cancelled/incomplete failure modes.
- disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml gets the
  same ctx moe_config.max_num_tokens chunking as the non-MTP config: its
  ctx sizing is identical, so it carries the same 7-8 GiB TRTLLM-Gen FP4
  MoE workspace transient and is a live OOM candidate on 192GB B200.

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 .../defs/disaggregated/test_aiperf_gate.py    | 122 ++++++++++++++++++
 ...p4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml |   6 +
 .../defs/disaggregated/test_disaggregated.py  |  48 ++++++-
 3 files changed, 173 insertions(+), 3 deletions(-)
 create mode 100644 tests/integration/defs/disaggregated/test_aiperf_gate.py

diff --git a/tests/integration/defs/disaggregated/test_aiperf_gate.py b/tests/integration/defs/disaggregated/test_aiperf_gate.py
new file mode 100644
index 000000000000..4a38ab34abce
--- /dev/null
+++ b/tests/integration/defs/disaggregated/test_aiperf_gate.py
@@ -0,0 +1,122 @@
+# 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.
+"""GPU-free unit tests for the aiperf error-rate gate.
+
+The gate (enforce_aiperf_error_rate) is applied by default to every
+disaggregated stress config, so these synthetic profile_export.jsonl cases
+prove it (a) fires on the server-error storm it exists to catch, (b) passes a
+healthy run with intentional cancellations, and (c) refuses to treat a broken
+or implausible export as a clean pass. Run with:
+
+    pytest -sv disaggregated/test_aiperf_gate.py
+"""
+
+import json
+
+import pytest
+from test_disaggregated import enforce_aiperf_error_rate
+
+
+def _record(error=None):
+    rec = {
+        "metadata": {"x_request_id": "id"},
+        "metrics": {"request_latency": 1.0},
+    }
+    if error is not None:
+        rec["error"] = error
+    return rec
+
+
+_CANCEL = {
+    "code": 499,
+    "type": "RequestCancellationError",
+    "message": "Request cancelled 0.500s after being sent",
+}
+_SERVER_500 = {
+    "code": 500,
+    "type": "InternalServerError",
+    "message": '{"detail":"Internal server error Cluster is not ready"}',
+}
+
+
+def _write_export(tmp_path, records, raw_lines=()):
+    export = tmp_path / "profile_export.jsonl"
+    with open(export, "w") as f:
+        for rec in records:
+            f.write(json.dumps(rec) + "\n")
+        for line in raw_lines:
+            f.write(line + "\n")
+    return str(tmp_path)
+
+
+def test_fires_on_error_storm(tmp_path):
+    """Replay of the nvbugs/6472256 CI failure distribution: must fire."""
+    records = (
+        [_record(_CANCEL)] * 3038
+        + [_record(_SERVER_500)] * 4359
+        + [_record()] * (35000 - 3038 - 4359)
+    )
+    artifact_dir = _write_export(tmp_path, records)
+    with pytest.raises(AssertionError, match="exceeds threshold"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=35000)
+
+
+def test_passes_healthy_run_with_cancellations(tmp_path):
+    records = [_record()] * 898 + [_record(_CANCEL)] * 100 + [_record(_SERVER_500)] * 2
+    artifact_dir = _write_export(tmp_path, records)
+    enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
+
+
+def test_missing_export_raises(tmp_path):
+    with pytest.raises(FileNotFoundError, match="profile_export.jsonl"):
+        enforce_aiperf_error_rate(str(tmp_path), 0.05)
+
+
+def test_empty_export_fails(tmp_path):
+    artifact_dir = _write_export(tmp_path, [])
+    with pytest.raises(AssertionError, match="no parseable request records"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05)
+
+
+def test_all_cancelled_fails(tmp_path):
+    artifact_dir = _write_export(tmp_path, [_record(_CANCEL)] * 50)
+    with pytest.raises(AssertionError, match="classified as \\W*cancelled"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05)
+
+
+def test_corrupt_export_fails(tmp_path):
+    """Wholesale parse failure (format change) must not read as a clean run."""
+    artifact_dir = _write_export(tmp_path, [_record()] * 10, raw_lines=["{not json"] * 10)
+    with pytest.raises(AssertionError, match="failed to parse"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05)
+
+
+def test_single_truncated_line_tolerated(tmp_path):
+    """One partial trailing line (killed writer) does not fail the gate."""
+    artifact_dir = _write_export(tmp_path, [_record()] * 200, raw_lines=['{"metadata": {"x_req'])
+    enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=200)
+
+
+def test_incomplete_accounting_fails(tmp_path):
+    """Far fewer records than requests => refuse to compute a rate."""
+    artifact_dir = _write_export(tmp_path, [_record()] * 100)
+    with pytest.raises(AssertionError, match="accounting is \\W*incomplete"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
+
+
+def test_gate_disabled_paths_not_affected(tmp_path):
+    """expected_records=None skips the plausibility check (dataset-entry runs)."""
+    artifact_dir = _write_export(tmp_path, [_record()] * 5)
+    enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=None)
diff --git a/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml b/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml
index 82902fa21f6b..9f1c0f151be2 100644
--- a/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml
+++ b/tests/integration/defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml
@@ -22,6 +22,12 @@ context_servers:
     dtype: fp8
   moe_config:
     backend: TRTLLM
+    # Chunk the MoE forward at half of max_num_tokens, mirroring the non-MTP
+    # config: the ctx sizing here is identical (16640 tokens, 0.8 KV fraction,
+    # TRTLLM-Gen FP4 MoE), so a full-size chunk carries the same 7-8 GiB
+    # workspace transient that OOMs the ctx worker on 192GB B200 under
+    # 512-concurrency 8k prefill load.
+    max_num_tokens: 8320
   cuda_graph_config: null
   print_iter_log: true
   cache_transceiver_config:
diff --git a/tests/integration/defs/disaggregated/test_disaggregated.py b/tests/integration/defs/disaggregated/test_disaggregated.py
index 1cf46fdde0b5..cf50b1ae8f3c 100644
--- a/tests/integration/defs/disaggregated/test_disaggregated.py
+++ b/tests/integration/defs/disaggregated/test_disaggregated.py
@@ -2371,7 +2371,9 @@ def get_config_for_benchmark(model_root, backend):
     return serve_config
 
 
-def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
+def enforce_aiperf_error_rate(artifact_dir,
+                              max_error_rate,
+                              expected_records=None):
     """Fail if the fraction of non-cancellation request errors exceeds max_error_rate.
 
     aiperf exits 0 and counts HTTP 500s as completed requests, so without this
@@ -2381,6 +2383,19 @@ def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
     "error" object with code/type/message. Intentional client-side cancellations
     (HTTP 499 / RequestCancellationError) are excluded from both the numerator
     and the denominator — stress tests cancel a fraction of requests on purpose.
+
+    Record schema validated against aiperf==0.8.0 (the pin in
+    requirements-dev.txt): each JSONL line is a MetricRecordInfo object
+    ({"metadata", "metrics", "error"}), and both cancellation construction
+    sites in aiperf's aiohttp client emit exactly error.code == 499 with
+    error.type == "RequestCancellationError". Reference gate output from the
+    35000-request DeepSeek R1 FP4 validation run (cancellation_rate=10):
+    [aiperf-gate] non-cancellation errors: 0/31561 (0.00%), cancelled: 3439.
+
+    The export must be substantive for the gate to pass: an empty or
+    unparsable export, an all-cancelled record set, or a record count far
+    below expected_records fails loudly instead of degrading into a silent
+    pass (a broken export is itself evidence of a broken run).
     """
     export_path = os.path.join(artifact_dir, "profile_export.jsonl")
     # Explicit raise (not assert): asserts are stripped under python -O, which
@@ -2393,6 +2408,7 @@ def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
             "max_error_rate=None explicitly.")
     total = 0
     cancelled = 0
+    decode_failures = 0
     # (code, type) -> [count, example message]
     error_counts: dict[tuple, list] = {}
     with open(export_path, "r", errors="replace") as f:
@@ -2403,6 +2419,7 @@ def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
             try:
                 record = json.loads(line)
             except json.JSONDecodeError:
+                decode_failures += 1
                 continue
             total += 1
             error = record.get("error")
@@ -2416,14 +2433,37 @@ def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
             entry = error_counts.setdefault((code, err_type),
                                             [0, error.get("message", "")])
             entry[0] += 1
+    # Substantiveness checks: never let a broken export read as a clean run.
+    # A single truncated trailing line is tolerated; wholesale parse failure
+    # means the record format changed and the gate can no longer be trusted.
+    if decode_failures > max(1, (total + decode_failures) // 100):
+        raise AssertionError(
+            f"{export_path}: {decode_failures} of {total + decode_failures} "
+            "lines failed to parse as JSON — the export is corrupt or the "
+            "aiperf record format changed; refusing to compute an error rate "
+            "from it.")
+    if total == 0:
+        raise AssertionError(
+            f"{export_path} contained no parseable request records — the "
+            "benchmark produced no per-request accounting, which itself "
+            "indicates a broken run.")
+    if expected_records is not None and total < expected_records * 0.9:
+        raise AssertionError(
+            f"{export_path} contained only {total} records but "
+            f"~{expected_records} were expected — per-request accounting is "
+            "incomplete; refusing to compute an error rate from it.")
     considered = total - cancelled
     if considered <= 0:
-        return
+        raise AssertionError(
+            f"all {total} records in {export_path} were classified as "
+            "cancelled — implausible at the configured cancellation rate and "
+            "indicates a broken run or cancellation-classification drift.")
     errors = sum(count for count, _ in error_counts.values())
     error_rate = errors / considered
     print(
         f"[aiperf-gate] non-cancellation errors: {errors}/{considered} "
         f"({error_rate:.2%}), cancelled: {cancelled}, "
+        f"decode failures: {decode_failures}, "
         f"threshold: {max_error_rate:.2%}",
         flush=True)
     if error_rate > max_error_rate:
@@ -2592,7 +2632,9 @@ def run_disaggregated_aiperf(config_file,
         # of requests, so this is the only check that catches a mid-run error
         # storm on this path (the fatal-pattern scan above is hang/OOM only).
         if max_error_rate is not None:
-            enforce_aiperf_error_rate(artifact_dir, max_error_rate)
+            enforce_aiperf_error_rate(artifact_dir,
+                                      max_error_rate,
+                                      expected_records=request_count)
 
         if accuracy_test:
             accuracy_test_result, accuracy_value = run_accuracy_test(

From dcbf25cdb66d36e5c5143f44f297eea28c204734 Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 15:55:32 -0700
Subject: [PATCH 4/8] [https://nvbugs/6472256][fix] Widen aiperf cancellation
 detection; exclude non-request records

Addresses fredricz-20070104's review on #17427 (the blast-radius concern
for the default-on gate across all stress configs at cancellation_rate=10):
- Cancellation classification no longer hinges solely on
  error.code == 499 / error.type == "RequestCancellationError": the
  per-record metadata.was_cancelled field is honored as a fallback, so a
  future aiperf that records cancellations with a different error shape
  (or no error object) cannot push ~10% intentional cancellations into
  the error numerator and trip every stress test at once.
- Records carrying neither metrics nor an error object (future
  non-request metadata lines) are excluded from the denominator instead
  of diluting the reported rate; the [aiperf-gate] line now reports the
  skipped count.
- Unit tests cover both: drifted-schema cancellations at 10% must pass
  the 5% threshold, and 10% real errors must still fire even when
  padded with 9x non-request records.

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 .../defs/disaggregated/test_aiperf_gate.py    | 30 +++++++++++++++++++
 .../defs/disaggregated/test_disaggregated.py  | 20 ++++++++++++-
 2 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/tests/integration/defs/disaggregated/test_aiperf_gate.py b/tests/integration/defs/disaggregated/test_aiperf_gate.py
index 4a38ab34abce..af5beaa9696b 100644
--- a/tests/integration/defs/disaggregated/test_aiperf_gate.py
+++ b/tests/integration/defs/disaggregated/test_aiperf_gate.py
@@ -120,3 +120,33 @@ def test_gate_disabled_paths_not_affected(tmp_path):
     """expected_records=None skips the plausibility check (dataset-entry runs)."""
     artifact_dir = _write_export(tmp_path, [_record()] * 5)
     enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=None)
+
+
+def test_was_cancelled_metadata_fallback(tmp_path):
+    """Cancellations are excluded even if the error shape drifts.
+
+    A future aiperf may record cancellations with a different error type, a
+    null code, or no error object at all; metadata.was_cancelled still
+    classifies them as intentional cancellations rather than server errors.
+    """
+    drifted_error = {"code": None, "type": "ClientDisconnected", "message": "x"}
+    records = [_record()] * 900
+    for rec_error in ([drifted_error] * 50, [None] * 50):
+        for err in rec_error:
+            rec = _record(err)
+            rec["metadata"]["was_cancelled"] = True
+            records.append(rec)
+    artifact_dir = _write_export(tmp_path, records)
+    # 100 drifted cancellations at 10% must not trip the 5% threshold.
+    enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
+
+
+def test_non_request_records_excluded_from_denominator(tmp_path):
+    """Records without metrics/error (future metadata lines) do not dilute the rate."""
+    records = [_record()] * 90 + [_record(_SERVER_500)] * 10
+    non_request = [{"summary": {"total": 100}}] * 900
+    artifact_dir = _write_export(tmp_path, records + non_request)
+    # 10 errors over 100 requests = 10% — must fire even though 900 metadata
+    # lines would dilute it to ~1% if they were counted as requests.
+    with pytest.raises(AssertionError, match="exceeds threshold"):
+        enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=100)
diff --git a/tests/integration/defs/disaggregated/test_disaggregated.py b/tests/integration/defs/disaggregated/test_disaggregated.py
index cf50b1ae8f3c..a0dbb8388624 100644
--- a/tests/integration/defs/disaggregated/test_disaggregated.py
+++ b/tests/integration/defs/disaggregated/test_disaggregated.py
@@ -2409,6 +2409,7 @@ def enforce_aiperf_error_rate(artifact_dir,
     total = 0
     cancelled = 0
     decode_failures = 0
+    non_request = 0
     # (code, type) -> [count, example message]
     error_counts: dict[tuple, list] = {}
     with open(export_path, "r", errors="replace") as f:
@@ -2421,13 +2422,29 @@ def enforce_aiperf_error_rate(artifact_dir,
             except json.JSONDecodeError:
                 decode_failures += 1
                 continue
+            if not isinstance(record, dict) or ("metrics" not in record
+                                                and "error" not in record):
+                # Non-request records would inflate the denominator and
+                # under-report the rate. The aiperf==0.8.0 record export is
+                # request-only, but stay robust to future metadata additions.
+                non_request += 1
+                continue
             total += 1
             error = record.get("error")
+            # Schema-drift fallback: per-record metadata carries was_cancelled
+            # independently of the error object's code/type, so a cancelled
+            # request is excluded even if a future aiperf changes the error
+            # shape (or omits the error object for cancellations entirely).
+            metadata = record.get("metadata") or {}
+            was_cancelled = bool(metadata.get("was_cancelled"))
             if not error:
+                if was_cancelled:
+                    cancelled += 1
                 continue
             code = error.get("code")
             err_type = error.get("type")
-            if code == 499 or err_type == "RequestCancellationError":
+            if (code == 499 or err_type == "RequestCancellationError"
+                    or was_cancelled):
                 cancelled += 1
                 continue
             entry = error_counts.setdefault((code, err_type),
@@ -2464,6 +2481,7 @@ def enforce_aiperf_error_rate(artifact_dir,
         f"[aiperf-gate] non-cancellation errors: {errors}/{considered} "
         f"({error_rate:.2%}), cancelled: {cancelled}, "
         f"decode failures: {decode_failures}, "
+        f"non-request records: {non_request}, "
         f"threshold: {max_error_rate:.2%}",
         flush=True)
     if error_rate > max_error_rate:

From fbf59713fed18d4fb6ee8214d56e5e74449f8d1d Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 16:05:30 -0700
Subject: [PATCH 5/8] [https://nvbugs/6472256][fix] Unwaive
 qwen3_32b_fp8_stress on B200

Follow-up to brnguyen2's review on #17427: the B200 qwen3_32b_fp8_stress
waive references the same NVBug this PR fixes, and the cluster-timing fix
applies to it (same run_disaggregated_aiperf path), so it is unwaived
alongside the DeepSeek param. A dedicated validation run on 8xB200 is
queued; the H100 waive for the same test stays (separate bug, 6312828).

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 tests/integration/test_lists/waives.txt | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt
index 60a8f730392b..44f51ed36cb0 100644
--- a/tests/integration/test_lists/waives.txt
+++ b/tests/integration/test_lists/waives.txt
@@ -164,7 +164,6 @@ full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpu
 full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpus_block_reuse[TEP4] SKIP (https://nvbugs/6474894)
 full:B200/accuracy/test_llm_api_pytorch.py::TestQwen3_30B_A3B::test_dummy_load_format SKIP (https://nvbugs/6525059)
 full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-glm5_nvfp4_tp4_ep4_dp_stress] SKIP (https://nvbugs/6544407)
-full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-qwen3_32b_fp8_stress] SKIP (https://nvbugs/6472256)
 full:B200/llmapi/test_llm_api_pytorch_moe_lora.py::test_qwen_moe_routed_expert_multi_lora_varying_ranks[cudagraph] SKIP (https://nvbugs/6475623)
 full:B200/llmapi/test_llm_api_pytorch_moe_lora.py::test_qwen_moe_routed_expert_multi_lora_varying_ranks[eager] SKIP (https://nvbugs/6475621)
 full:B200/test_e2e.py::test_multi_nodes_eval[Qwen3/Qwen3-235B-A22B-tp16-mmlu] SKIP (https://nvbugs/6424188)

From f4dd1fd959c98d33d5e254e6e617ccb96ffe65ca Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 22:51:13 -0700
Subject: [PATCH 6/8] [https://nvbugs/6472256][fix] Add type annotations and
 docstrings to aiperf gate tests

Addresses CodeRabbit's review of the gate unit tests on #17427: every
helper and test function is now annotated (tests return None), the
previously bare functions have Google-style docstrings, and the regex
match= patterns containing metacharacters are raw strings (Ruff RUF043).

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 .../defs/disaggregated/test_aiperf_gate.py    | 63 ++++++++++++++-----
 1 file changed, 46 insertions(+), 17 deletions(-)

diff --git a/tests/integration/defs/disaggregated/test_aiperf_gate.py b/tests/integration/defs/disaggregated/test_aiperf_gate.py
index af5beaa9696b..d33824b1a9a1 100644
--- a/tests/integration/defs/disaggregated/test_aiperf_gate.py
+++ b/tests/integration/defs/disaggregated/test_aiperf_gate.py
@@ -24,13 +24,24 @@
 """
 
 import json
+from pathlib import Path
+from typing import Any, Optional, Sequence
 
 import pytest
 from test_disaggregated import enforce_aiperf_error_rate
 
 
-def _record(error=None):
-    rec = {
+def _record(error: Optional[dict[str, Any]] = None) -> dict[str, Any]:
+    """Build a minimal aiperf MetricRecordInfo-shaped request record.
+
+    Args:
+        error: Optional ErrorDetails-shaped object ({"code", "type",
+            "message"}) attached to the record.
+
+    Returns:
+        A dict with "metadata" and "metrics" keys, plus "error" when given.
+    """
+    rec: dict[str, Any] = {
         "metadata": {"x_request_id": "id"},
         "metrics": {"request_latency": 1.0},
     }
@@ -51,7 +62,21 @@ def _record(error=None):
 }
 
 
-def _write_export(tmp_path, records, raw_lines=()):
+def _write_export(
+    tmp_path: Path,
+    records: Sequence[dict[str, Any]],
+    raw_lines: Sequence[str] = (),
+) -> str:
+    """Write a synthetic profile_export.jsonl into tmp_path.
+
+    Args:
+        tmp_path: Directory to write the export into (pytest tmp_path).
+        records: Records serialized one-per-line as JSON.
+        raw_lines: Extra lines appended verbatim (e.g. corrupt/truncated).
+
+    Returns:
+        The artifact directory path to pass to enforce_aiperf_error_rate.
+    """
     export = tmp_path / "profile_export.jsonl"
     with open(export, "w") as f:
         for rec in records:
@@ -61,7 +86,7 @@ def _write_export(tmp_path, records, raw_lines=()):
     return str(tmp_path)
 
 
-def test_fires_on_error_storm(tmp_path):
+def test_fires_on_error_storm(tmp_path: Path) -> None:
     """Replay of the nvbugs/6472256 CI failure distribution: must fire."""
     records = (
         [_record(_CANCEL)] * 3038
@@ -73,56 +98,60 @@ def test_fires_on_error_storm(tmp_path):
         enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=35000)
 
 
-def test_passes_healthy_run_with_cancellations(tmp_path):
+def test_passes_healthy_run_with_cancellations(tmp_path: Path) -> None:
+    """A clean run with 10% intentional cancellations passes the 5% gate."""
     records = [_record()] * 898 + [_record(_CANCEL)] * 100 + [_record(_SERVER_500)] * 2
     artifact_dir = _write_export(tmp_path, records)
     enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
 
 
-def test_missing_export_raises(tmp_path):
-    with pytest.raises(FileNotFoundError, match="profile_export.jsonl"):
+def test_missing_export_raises(tmp_path: Path) -> None:
+    """A missing export file raises FileNotFoundError with the path."""
+    with pytest.raises(FileNotFoundError, match=r"profile_export\.jsonl"):
         enforce_aiperf_error_rate(str(tmp_path), 0.05)
 
 
-def test_empty_export_fails(tmp_path):
+def test_empty_export_fails(tmp_path: Path) -> None:
+    """An export with zero request records must not read as a clean pass."""
     artifact_dir = _write_export(tmp_path, [])
     with pytest.raises(AssertionError, match="no parseable request records"):
         enforce_aiperf_error_rate(artifact_dir, 0.05)
 
 
-def test_all_cancelled_fails(tmp_path):
+def test_all_cancelled_fails(tmp_path: Path) -> None:
+    """An all-cancelled record set indicates a broken run and must fail."""
     artifact_dir = _write_export(tmp_path, [_record(_CANCEL)] * 50)
-    with pytest.raises(AssertionError, match="classified as \\W*cancelled"):
+    with pytest.raises(AssertionError, match=r"classified as \W*cancelled"):
         enforce_aiperf_error_rate(artifact_dir, 0.05)
 
 
-def test_corrupt_export_fails(tmp_path):
+def test_corrupt_export_fails(tmp_path: Path) -> None:
     """Wholesale parse failure (format change) must not read as a clean run."""
     artifact_dir = _write_export(tmp_path, [_record()] * 10, raw_lines=["{not json"] * 10)
     with pytest.raises(AssertionError, match="failed to parse"):
         enforce_aiperf_error_rate(artifact_dir, 0.05)
 
 
-def test_single_truncated_line_tolerated(tmp_path):
+def test_single_truncated_line_tolerated(tmp_path: Path) -> None:
     """One partial trailing line (killed writer) does not fail the gate."""
     artifact_dir = _write_export(tmp_path, [_record()] * 200, raw_lines=['{"metadata": {"x_req'])
     enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=200)
 
 
-def test_incomplete_accounting_fails(tmp_path):
+def test_incomplete_accounting_fails(tmp_path: Path) -> None:
     """Far fewer records than requests => refuse to compute a rate."""
     artifact_dir = _write_export(tmp_path, [_record()] * 100)
-    with pytest.raises(AssertionError, match="accounting is \\W*incomplete"):
+    with pytest.raises(AssertionError, match=r"accounting is \W*incomplete"):
         enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
 
 
-def test_gate_disabled_paths_not_affected(tmp_path):
+def test_gate_disabled_paths_not_affected(tmp_path: Path) -> None:
     """expected_records=None skips the plausibility check (dataset-entry runs)."""
     artifact_dir = _write_export(tmp_path, [_record()] * 5)
     enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=None)
 
 
-def test_was_cancelled_metadata_fallback(tmp_path):
+def test_was_cancelled_metadata_fallback(tmp_path: Path) -> None:
     """Cancellations are excluded even if the error shape drifts.
 
     A future aiperf may record cancellations with a different error type, a
@@ -141,7 +170,7 @@ def test_was_cancelled_metadata_fallback(tmp_path):
     enforce_aiperf_error_rate(artifact_dir, 0.05, expected_records=1000)
 
 
-def test_non_request_records_excluded_from_denominator(tmp_path):
+def test_non_request_records_excluded_from_denominator(tmp_path: Path) -> None:
     """Records without metrics/error (future metadata lines) do not dilute the rate."""
     records = [_record()] * 90 + [_record(_SERVER_500)] * 10
     non_request = [{"summary": {"total": 100}}] * 900

From f0604294dc08e059447f827f57e2a4900ac65101 Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Mon, 10 Aug 2026 22:58:45 -0700
Subject: [PATCH 7/8] [https://nvbugs/6472256][fix] Derive expected records in
 default request mode; register gate tests

Addresses CodeRabbit's follow-up review on #17427:
- When request_count is None, aiperf 0.8.0 derives the request count as
  max(10, concurrency * 2) (--num-dataset-entries only sizes the prompt
  pool), so the gate now mirrors that derivation instead of silently
  disabling its completeness check in the default request mode.
- The 11 aiperf gate tests are registered in test-db/l0_cpu.yml
  (GPU-free, pre-merge) and qa/llm_function_core.txt so they actually
  run in CI/QA.

Skipped with reason: the raw-regex RUF043 comment was already fixed in
f4dd1fd959 (stale review anchor), and registering the pre-existing
test_llama4_long_context_kv_cache_overflow RCCA repro is out of scope
for this PR.

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 .../defs/disaggregated/test_disaggregated.py          |  9 ++++++++-
 tests/integration/test_lists/qa/llm_function_core.txt | 11 +++++++++++
 tests/integration/test_lists/test-db/l0_cpu.yml       |  1 +
 3 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/tests/integration/defs/disaggregated/test_disaggregated.py b/tests/integration/defs/disaggregated/test_disaggregated.py
index f521a165f84e..4c7deaf7e37b 100644
--- a/tests/integration/defs/disaggregated/test_disaggregated.py
+++ b/tests/integration/defs/disaggregated/test_disaggregated.py
@@ -2733,9 +2733,16 @@ def run_disaggregated_aiperf(config_file,
         # of requests, so this is the only check that catches a mid-run error
         # storm on this path (the fatal-pattern scan above is hang/OOM only).
         if max_error_rate is not None:
+            # aiperf 0.8.0 derives the request count from concurrency when
+            # --request-count is not passed (max(10, concurrency * 2) for
+            # synthetic datasets; --num-dataset-entries only sizes the prompt
+            # pool). Mirror that so the completeness check also covers the
+            # default request mode instead of silently disabling.
+            expected_records = (request_count if request_count is not None else
+                                max(10, concurrency * 2))
             enforce_aiperf_error_rate(artifact_dir,
                                       max_error_rate,
-                                      expected_records=request_count)
+                                      expected_records=expected_records)
 
         if accuracy_test:
             accuracy_test_result, accuracy_value = run_accuracy_test(
diff --git a/tests/integration/test_lists/qa/llm_function_core.txt b/tests/integration/test_lists/qa/llm_function_core.txt
index 6f222f7bab1f..656efedd1518 100644
--- a/tests/integration/test_lists/qa/llm_function_core.txt
+++ b/tests/integration/test_lists/qa/llm_function_core.txt
@@ -821,6 +821,17 @@ disaggregated/test_ad_disagg.py::test_async_generation_matches_aggregate
 disaggregated/test_ad_disagg.py::test_async_generation_no_overlap_matches_aggregate
 disaggregated/test_ad_disagg.py::test_async_sharded_generation_handoff
 disaggregated/test_ad_disagg_trtllm_serve.py::test_openai_completion
+disaggregated/test_aiperf_gate.py::test_fires_on_error_storm
+disaggregated/test_aiperf_gate.py::test_passes_healthy_run_with_cancellations
+disaggregated/test_aiperf_gate.py::test_missing_export_raises
+disaggregated/test_aiperf_gate.py::test_empty_export_fails
+disaggregated/test_aiperf_gate.py::test_all_cancelled_fails
+disaggregated/test_aiperf_gate.py::test_corrupt_export_fails
+disaggregated/test_aiperf_gate.py::test_single_truncated_line_tolerated
+disaggregated/test_aiperf_gate.py::test_incomplete_accounting_fails
+disaggregated/test_aiperf_gate.py::test_gate_disabled_paths_not_affected
+disaggregated/test_aiperf_gate.py::test_was_cancelled_metadata_fallback
+disaggregated/test_aiperf_gate.py::test_non_request_records_excluded_from_denominator
 disaggregated/test_auto_scaling.py::test_disagg_server_restart[etcd-round_robin]
 disaggregated/test_auto_scaling.py::test_disagg_server_restart[http-round_robin]
 disaggregated/test_auto_scaling.py::test_minimal_instances[etcd-round_robin]
diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml
index 7c4347f18efc..cd1028dd1c52 100644
--- a/tests/integration/test_lists/test-db/l0_cpu.yml
+++ b/tests/integration/test_lists/test-db/l0_cpu.yml
@@ -12,6 +12,7 @@ l0_cpu:
       backend: generic
       orchestrator: mpi
   tests:
+  - disaggregated/test_aiperf_gate.py
   - unittest/_torch/auto_deploy
   - unittest/_torch/cute_dsl/test_kimi_k3_kda_ptx_patch.py
   - unittest/_torch/distributed

From cf26ed1344a6922318e541368b312c5eb91fc806 Mon Sep 17 00:00:00 2001
From: Michal Guzek <mguzek@nvidia.com>
Date: Tue, 11 Aug 2026 13:43:00 -0700
Subject: [PATCH 8/8] [https://nvbugs/6472256][fix] Use explicit test IDs for
 aiperf gate tests in l0_cpu.yml

The Check Test List CI stage validates test-db entries against collected
test IDs, so the bare module path was rejected as invalid; expand it into
the 11 ::-qualified test names (matching the qa list entries, which
validated fine).

Signed-off-by: Michal Guzek <mguzek@nvidia.com>
---
 tests/integration/test_lists/test-db/l0_cpu.yml | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml
index cd1028dd1c52..a09de11af546 100644
--- a/tests/integration/test_lists/test-db/l0_cpu.yml
+++ b/tests/integration/test_lists/test-db/l0_cpu.yml
@@ -12,7 +12,17 @@ l0_cpu:
       backend: generic
       orchestrator: mpi
   tests:
-  - disaggregated/test_aiperf_gate.py
+  - disaggregated/test_aiperf_gate.py::test_fires_on_error_storm
+  - disaggregated/test_aiperf_gate.py::test_passes_healthy_run_with_cancellations
+  - disaggregated/test_aiperf_gate.py::test_missing_export_raises
+  - disaggregated/test_aiperf_gate.py::test_empty_export_fails
+  - disaggregated/test_aiperf_gate.py::test_all_cancelled_fails
+  - disaggregated/test_aiperf_gate.py::test_corrupt_export_fails
+  - disaggregated/test_aiperf_gate.py::test_single_truncated_line_tolerated
+  - disaggregated/test_aiperf_gate.py::test_incomplete_accounting_fails
+  - disaggregated/test_aiperf_gate.py::test_gate_disabled_paths_not_affected
+  - disaggregated/test_aiperf_gate.py::test_was_cancelled_metadata_fallback
+  - disaggregated/test_aiperf_gate.py::test_non_request_records_excluded_from_denominator
   - unittest/_torch/auto_deploy
   - unittest/_torch/cute_dsl/test_kimi_k3_kda_ptx_patch.py
   - unittest/_torch/distributed