fix(scenario): forbid idle-gap caps in agentperf-v2

docs/tutorials/agentperf-v2.md

@@ -37,6 +37,9 @@ uv run aiperf profile \
 `--concurrency` is the number of users. `--benchmark-duration` is a time limit, not the length of the run, and defaults
 to 12 hours.
 
+The scenario replays recorded idle time as is, so it rejects `--trace-idle-gap-cap-seconds` and
+`--system-idle-gap-cap-seconds` (with `--unsafe-override`, the run is marked `submission_valid=false`).
+
 At high concurrency, cap aiperf's session cache so finished sessions cannot use up the client's memory: set
 `AIPERF_WORKER_SESSION_CACHE_MAX_ENTRIES` to about 8 × users ÷ workers. The default was fine at 16 users.
 

src/aiperf/common/scenario/agentperf_v2.py

@@ -11,6 +11,9 @@
     require_streaming=True,
     forbid_ignore_trace_delays=True,
     forbid_input_truncation=True,
+    # Idle time between requests is part of the recorded workload.
+    forbid_trace_idle_gap_cap=True,
+    forbid_system_idle_gap_cap=True,
     require_loader=("agentperf_v2", "dag_jsonl"),
     min_benchmark_duration_seconds=1,
     default_benchmark_duration_seconds=43200,

src/aiperf/common/scenario/base.py

@@ -142,6 +142,13 @@ class ScenarioSpec(AIPerfBaseModel):
             "trace's original request timeline."
         ),
     )
+    forbid_system_idle_gap_cap: bool = Field(
+        default=False,
+        description=(
+            "Reject system_idle_gap_cap_seconds so the scenario replays globally "
+            "idle time as recorded instead of skipping it."
+        ),
+    )
     forbid_inter_turn_delay_cap: bool = Field(
         default=False,
         description=(

src/aiperf/common/scenario/validator.py

@@ -751,7 +751,23 @@ def _apply_system_idle_gap_cap(
     violations: list[ScenarioViolation],
     applied: list[str],
 ) -> None:
-    """Lock the global idle cap without altering dataset timestamps."""
+    """Forbid or lock the global idle cap without altering dataset timestamps."""
+    if spec.forbid_system_idle_gap_cap:
+        for phase in run.cfg.get_profiling_phases():
+            if phase.system_idle_gap_cap_seconds is not None:
+                violations.append(
+                    ScenarioViolation(
+                        flag="--system-idle-gap-cap-seconds",
+                        current_value=phase.system_idle_gap_cap_seconds,
+                        required_value=None,
+                        message=(
+                            f"scenario {spec.name!r} replays globally idle time "
+                            "as recorded and forbids a system idle-gap cap"
+                        ),
+                    )
+                )
+                return
+        return
     required = spec.system_idle_gap_cap_seconds
     if required is None:
         return

tests/unit/common/scenario/test_scenario_registry.py

@@ -49,3 +49,9 @@ def test_get_scenario_unknown_raises():
     with pytest.raises(UnknownScenarioError) as exc_info:
         get_scenario("nonsense-scenario-v9")
     assert "inferencex-agentx-mvp" in str(exc_info.value)
+
+
+def test_agentperf_v2_forbids_idle_gap_caps() -> None:
+    spec = get_scenario("agentperf-v2")
+    assert spec.forbid_trace_idle_gap_cap is True
+    assert spec.forbid_system_idle_gap_cap is True

tests/unit/common/scenario/test_scenario_validator.py

@@ -652,3 +652,50 @@ def test_random_seed_auto_filled_through_resolver_chain() -> None:
     assert run.random_seed is None
     build_default_resolver_chain().resolve_all(run)
     assert run.random_seed is not None
+
+
+_AGENTPERF_V2_DATASET = {"name": "main", "type": "public", "dataset": "agentperf_v2"}
+
+
+def test_agentperf_v2_accepts_recorded_idle_time() -> None:
+    run = _build_run(
+        scenario="agentperf-v2", dataset=dict(_AGENTPERF_V2_DATASET), streaming=True
+    )
+    outcome = apply_scenario(run)
+    assert outcome.submission_valid is True
+    assert run.cfg.get_profiling_phases()[0].system_idle_gap_cap_seconds is None
+
+
+@pytest.mark.parametrize(
+    "dataset_overrides, profiling_overrides, flag",
+    [
+        param({}, {"system_idle_gap_cap_seconds": 10.0}, "--system-idle-gap-cap-seconds", id="system_idle_gap_cap"),
+        param({"trace_idle_gap_cap_seconds": 10.0}, {}, "--trace-idle-gap-cap-seconds", id="trace_idle_gap_cap"),
+    ],
+)  # fmt: skip
+def test_agentperf_v2_forbids_idle_gap_caps(
+    dataset_overrides: dict[str, Any], profiling_overrides: dict[str, Any], flag: str
+) -> None:
+    run = _build_run(
+        scenario="agentperf-v2",
+        dataset={**_AGENTPERF_V2_DATASET, **dataset_overrides},
+        streaming=True,
+        profiling_overrides=profiling_overrides,
+    )
+    with pytest.raises(ScenarioLockError) as exc:
+        apply_scenario(run)
+    assert flag in str(exc.value)
+
+
+def test_agentperf_v2_idle_gap_caps_invalid_under_unsafe_override() -> None:
+    run = _build_run(
+        scenario="agentperf-v2",
+        unsafe_override=True,
+        dataset={**_AGENTPERF_V2_DATASET, "trace_idle_gap_cap_seconds": 10.0},
+        streaming=True,
+        profiling_overrides={"system_idle_gap_cap_seconds": 10.0},
+    )
+    outcome = apply_scenario(run)
+    assert outcome.submission_valid is False
+    flags = {violation.flag for violation in outcome.violations}
+    assert {"--trace-idle-gap-cap-seconds", "--system-idle-gap-cap-seconds"} <= flags