feat(scenario): replay recorded output lengths in agentperf-v2

docs/tutorials/agentperf-v2.md

@@ -43,6 +43,12 @@ At high concurrency, cap aiperf's session cache so finished sessions cannot use
 Use the recordings of the model your server is serving. Your server must report cached tokens in its usage figures for the
 cache results to appear: for example, `--enable-cache-report` on SGLang.
 
+Every request carries its turn's recorded output length as `max_tokens` and is sent with `ignore_eos`, so each reply is
+exactly as long as the recorded one and reply length cannot move the results. The dataset must set `max_tokens` on every
+turn; the loader rejects a turn without one. The scenario accepts only `--endpoint-type chat`, and rejects
+`--extra-inputs` keys that set output length or stopping, reasoning, chat templates, constrained decoding or the main
+sampling settings (the full list is `forbid_extra_input_keys` in the scenario).
+
 ## What a run does
 
 The dataset's sessions sit end to end on one timeline, which is the loop. `N` users are spaced evenly along it, and each

src/aiperf/common/scenario/agentperf_v2.py

@@ -7,7 +7,41 @@
 AGENTPERF_V2 = ScenarioSpec(
     name="agentperf-v2",
     timing_mode=TimingMode.AGENTIC_REPLAY,
-    require_ignore_eos=False,
+    # Every turn carries its recorded output length as max_tokens, and
+    # ignore_eos makes each reply exactly that long, so reply length cannot
+    # move the score.
+    require_ignore_eos=True,
+    require_turn_max_tokens=True,
+    require_endpoint_types=("chat",),
+    forbid_extra_input_keys=(
+        # Output length and stopping.
+        "max_tokens",
+        "max_completion_tokens",
+        "max_new_tokens",
+        "min_tokens",
+        "min_new_tokens",
+        "stop",
+        "stop_token_ids",
+        "stop_regex",
+        "include_stop_str_in_output",
+        # Reasoning, templates and constrained decoding.
+        "reasoning",
+        "reasoning_effort",
+        "chat_template",
+        "chat_template_kwargs",
+        "response_format",
+        "structured_outputs",
+        "payload_template",
+        "vllm_xargs",
+        # Sampling.
+        "temperature",
+        "top_p",
+        "top_k",
+        "min_p",
+        "repetition_penalty",
+        "presence_penalty",
+        "frequency_penalty",
+    ),
     require_streaming=True,
     forbid_ignore_trace_delays=True,
     forbid_input_truncation=True,

src/aiperf/common/scenario/base.py

@@ -149,6 +149,32 @@ class ScenarioSpec(AIPerfBaseModel):
             "each trace's original inter-turn timing."
         ),
     )
+    require_endpoint_types: tuple[str, ...] = Field(
+        default=(),
+        description=(
+            "Endpoint types the scenario accepts. Empty accepts any. Limiting "
+            "it keeps request bodies built by formatters that write each "
+            "turn's max_tokens, so a custom payload template cannot replace it."
+        ),
+    )
+    require_turn_max_tokens: bool = Field(
+        default=False,
+        description=(
+            "Reject a DAG dataset in which any request turn lacks max_tokens, "
+            "or carries a forbidden extra key or ignore_eos=false. Pair with "
+            "require_ignore_eos so every reply has exactly the length the "
+            "dataset sets for that turn (its recorded output length). This is "
+            "a load error, so --unsafe-override does not bypass it."
+        ),
+    )
+    forbid_extra_input_keys: tuple[str, ...] = Field(
+        default=(),
+        description=(
+            "Keys rejected in --extra-inputs. Extra inputs are merged into "
+            "every request body after the per-turn max_tokens, so these keys "
+            "would override the dataset's output length or decoding settings."
+        ),
+    )
     loop: LoopRules | None = Field(
         default=None,
         description="Space lanes evenly on one loop of the dataset's sessions, and "

src/aiperf/common/scenario/validator.py

@@ -84,6 +84,8 @@ def apply_scenario(run: BenchmarkRun) -> ScenarioOutcome:
     _apply_timing_mode(run, spec, violations, applied)
     _apply_require_streaming(run, spec, violations, applied)
     _apply_require_ignore_eos(run, spec, violations, applied)
+    _apply_forbid_extra_input_keys(run, spec, violations, applied)
+    _apply_require_endpoint_types(run, spec, violations, applied)
     _apply_forbid_ignore_trace_delays(run, spec, violations, applied)
     _apply_use_think_time_only(run, spec, violations, applied)
     _apply_forbid_input_truncation(run, spec, violations, applied)
@@ -263,6 +265,64 @@ def _apply_require_streaming(
         applied.append("streaming")
 
 
+def _apply_forbid_extra_input_keys(
+    run: BenchmarkRun,
+    spec: ScenarioSpec,
+    violations: list[ScenarioViolation],
+    applied: list[str],
+) -> None:
+    """Reject ``--extra-inputs`` keys that would override per-turn settings.
+
+    ``EndpointConfig.extra`` is merged into every request body after the
+    per-turn ``max_tokens``, so a key such as ``max_tokens`` or ``min_tokens``
+    there would replace the dataset's output length for every request.
+    """
+    if not spec.forbid_extra_input_keys:
+        return
+    extra = run.cfg.endpoint.extra
+    found = [key for key in spec.forbid_extra_input_keys if key in extra]
+    for key in found:
+        violations.append(
+            ScenarioViolation(
+                flag=f"extra_inputs.{key}",
+                current_value=extra[key],
+                required_value=None,
+                message=(
+                    f"scenario {spec.name!r} takes {key!r} from the dataset and "
+                    "forbids overriding it with --extra-inputs"
+                ),
+            )
+        )
+    if not found:
+        applied.append("forbid_extra_input_keys")
+
+
+def _apply_require_endpoint_types(
+    run: BenchmarkRun,
+    spec: ScenarioSpec,
+    violations: list[ScenarioViolation],
+    applied: list[str],
+) -> None:
+    """Reject endpoint types outside ``spec.require_endpoint_types``."""
+    if not spec.require_endpoint_types:
+        return
+    endpoint_type = str(run.cfg.endpoint.type)
+    if endpoint_type in spec.require_endpoint_types:
+        applied.append("endpoint_type")
+        return
+    violations.append(
+        ScenarioViolation(
+            flag="--endpoint-type",
+            current_value=endpoint_type,
+            required_value=", ".join(spec.require_endpoint_types),
+            message=(
+                f"scenario {spec.name!r} accepts only endpoint types "
+                f"{', '.join(spec.require_endpoint_types)}"
+            ),
+        )
+    )
+
+
 def _apply_require_ignore_eos(
     run: BenchmarkRun,
     spec: ScenarioSpec,

src/aiperf/dataset/loader/dag_jsonl.py

@@ -17,6 +17,7 @@
 from aiperf.common.models import DatasetMetadata, TurnPrerequisite
 from aiperf.common.models.branch import ConversationBranchInfo
 from aiperf.common.models.dataset_models import Conversation, ThinkTimeSpec, Turn
+from aiperf.common.scenario.registry import get_scenario
 from aiperf.common.validators.orchestrator_v1 import validate_for_orchestrator_v1
 from aiperf.dataset.loader._dag_jsonl_helpers import (
     DagLoadError,
@@ -195,6 +196,7 @@ def load_dataset(self) -> dict[str, list[Conversation]]:
             self._parse_lines()
             self._desugar_forks()
             self._resolve_and_validate()
+            self._check_turn_max_tokens()
             self._roots = self._compute_roots()
             for sid, conv in self._conversations.items():
                 # Respect an author-declared context mode (e.g.
@@ -393,6 +395,47 @@ def _check_pre_session_duplicates(
             seen.add(child)
         return list(children)
 
+    def _check_turn_max_tokens(self) -> None:
+        """Reject request turns that would not get the dataset's output length.
+
+        Under ``require_ignore_eos`` a request turn without ``max_tokens``
+        would decode until the server's own limit, and a per-turn ``extra``
+        key from the scenario's forbid list, or ``ignore_eos=false``, would
+        override the length. Turns that send no request are skipped.
+        """
+        scenario_name = self.run.cfg.scenario if self.run is not None else None
+        if scenario_name is None:
+            return
+        spec = get_scenario(scenario_name)
+        if not spec.require_turn_max_tokens:
+            return
+        missing: list[str] = []
+        overridden: list[str] = []
+        for sid, conv in self._conversations.items():
+            for index, turn in enumerate(conv.turns):
+                if turn.no_request:
+                    continue
+                if turn.max_tokens is None:
+                    missing.append(f"{sid}[{index}]")
+                extra = turn.extra_body or {}
+                if extra.get("ignore_eos") is False or any(
+                    key in extra for key in spec.forbid_extra_input_keys
+                ):
+                    overridden.append(f"{sid}[{index}]")
+        if missing:
+            raise DagLoadError(
+                f"{self._path}: scenario {scenario_name!r} requires max_tokens on "
+                f"every request turn, but {len(missing)} turn(s) have none "
+                f"(first: {', '.join(missing[:5])})"
+            )
+        if overridden:
+            raise DagLoadError(
+                f"{self._path}: scenario {scenario_name!r} takes output length "
+                "from max_tokens, but the extra of "
+                f"{len(overridden)} turn(s) overrides it "
+                f"(first: {', '.join(overridden[:5])})"
+            )
+
     def _build_turn(self, t: DagTurn) -> Turn:
         return Turn(
             raw_messages=list(t.messages),

tests/unit/common/scenario/test_agentperf_v2_output_length.py

@@ -0,0 +1,86 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""agentperf-v2 locks output length to the dataset's per-turn max_tokens."""
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+from pytest import param
+
+from aiperf.common.scenario import ScenarioLockError, apply_scenario
+from aiperf.config.config import BenchmarkConfig
+from aiperf.config.resolution.plan import BenchmarkRun
+
+
+def _build_run(
+    extra: dict[str, Any] | None = None, endpoint_type: str = "chat"
+) -> BenchmarkRun:
+    endpoint: dict[str, Any] = {
+        "urls": ["http://localhost:8000/v1/chat/completions"],
+        "type": endpoint_type,
+        "streaming": True,
+    }
+    if extra is not None:
+        endpoint["extra"] = extra
+    if endpoint_type == "template":
+        endpoint["template"] = {
+            "body": '{"messages": {{ messages | tojson }}, "max_tokens": 16}'
+        }
+    cfg = BenchmarkConfig.model_validate(
+        {
+            "models": ["my-model"],
+            "endpoint": endpoint,
+            "datasets": [{"name": "main", "type": "public", "dataset": "agentperf_v2"}],
+            "phases": [
+                {
+                    "name": "profiling",
+                    "type": "concurrency",
+                    "concurrency": 8,
+                    "duration": 1800,
+                }
+            ],
+            "scenario": "agentperf-v2",
+        }
+    )
+    return BenchmarkRun(
+        benchmark_id="test-run",
+        cfg=cfg,
+        artifact_dir=Path("/tmp/aiperf-scenario-test"),
+    )
+
+
+def test_agentperf_v2_injects_ignore_eos() -> None:
+    run = _build_run()
+    outcome = apply_scenario(run)
+    assert outcome.submission_valid is True
+    assert run.cfg.endpoint.extra["ignore_eos"] is True
+
+
+@pytest.mark.parametrize(
+    "key, value",
+    [
+        param("max_tokens", 4096, id="max_tokens"),
+        param("max_completion_tokens", 4096, id="max_completion_tokens"),
+        param("min_tokens", 16, id="min_tokens"),
+        param("stop", ["</s>"], id="stop"),
+        param("reasoning_effort", "low", id="reasoning_effort"),
+        param("chat_template_kwargs", {"enable_thinking": False}, id="chat_template_kwargs"),
+        param("temperature", 0.0, id="temperature"),
+        param("payload_template", "{}", id="payload_template"),
+        param("stop_regex", "x", id="stop_regex"),
+    ],
+)  # fmt: skip
+def test_agentperf_v2_rejects_output_shaping_extra_inputs(key: str, value: Any) -> None:
+    run = _build_run(extra={key: value})
+    with pytest.raises(ScenarioLockError) as exc:
+        apply_scenario(run)
+    assert f"extra_inputs.{key}" in str(exc.value)
+
+
+@pytest.mark.parametrize("endpoint_type", ["template", "completions"])
+def test_agentperf_v2_accepts_only_chat_endpoint(endpoint_type: str) -> None:
+    run = _build_run(endpoint_type=endpoint_type)
+    with pytest.raises(ScenarioLockError) as exc:
+        apply_scenario(run)
+    assert "--endpoint-type" in str(exc.value)