diff --git a/docs/tutorials/agentperf-v2.md b/docs/tutorials/agentperf-v2.md
index fc311b0320..c40c29de6c 100644
--- a/docs/tutorials/agentperf-v2.md
+++ b/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
diff --git a/src/aiperf/common/scenario/agentperf_v2.py b/src/aiperf/common/scenario/agentperf_v2.py
index 3e79adfc52..3f6420c352 100644
--- a/src/aiperf/common/scenario/agentperf_v2.py
+++ b/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,
diff --git a/src/aiperf/common/scenario/base.py b/src/aiperf/common/scenario/base.py
index c7ccc92559..4972480d23 100644
--- a/src/aiperf/common/scenario/base.py
+++ b/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 "
diff --git a/src/aiperf/common/scenario/validator.py b/src/aiperf/common/scenario/validator.py
index c82945efe8..7d3901b4f9 100644
--- a/src/aiperf/common/scenario/validator.py
+++ b/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,
diff --git a/src/aiperf/dataset/loader/dag_jsonl.py b/src/aiperf/dataset/loader/dag_jsonl.py
index f19ea3f0fc..b270d71320 100644
--- a/src/aiperf/dataset/loader/dag_jsonl.py
+++ b/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),
diff --git a/tests/unit/common/scenario/test_agentperf_v2_output_length.py b/tests/unit/common/scenario/test_agentperf_v2_output_length.py
new file mode 100644
index 0000000000..764e33e772
--- /dev/null
+++ b/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)
diff --git a/tests/unit/dataset/loader/test_dag_jsonl_turn_max_tokens.py b/tests/unit/dataset/loader/test_dag_jsonl_turn_max_tokens.py
new file mode 100644
index 0000000000..f85b2d3b64
--- /dev/null
+++ b/tests/unit/dataset/loader/test_dag_jsonl_turn_max_tokens.py
@@ -0,0 +1,99 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Scenario-required per-turn max_tokens in the DAG JSONL loader."""
+
+import json
+from pathlib import Path
+
+import pytest
+from pytest import param
+
+from aiperf.dataset.loader.dag_jsonl import DagJsonlLoader, DagLoadError
+from tests.unit.dataset.loader.conftest import make_weka_run
+
+
+def _write(tmp_path: Path, max_tokens: list[int | None]) -> Path:
+    turns = []
+    for value in max_tokens:
+        turn: dict = {"messages": [{"role": "user", "content": "hi"}]}
+        if value is not None:
+            turn["max_tokens"] = value
+        turns.append(turn)
+    path = tmp_path / "dag.jsonl"
+    path.write_text(json.dumps({"session_id": "s1", "turns": turns}) + "\n")
+    return path
+
+
+def _loader(path: Path, scenario: str | None) -> DagJsonlLoader:
+    run = make_weka_run()
+    run.cfg.scenario = scenario
+    return DagJsonlLoader(filename=str(path), run=run)
+
+
+def test_agentperf_v2_accepts_turns_with_max_tokens(tmp_path: Path) -> None:
+    loader = _loader(_write(tmp_path, [120, 45]), "agentperf-v2")
+    convs = loader.convert_to_conversations(loader.load_dataset())
+    assert [turn.max_tokens for turn in convs[0].turns] == [120, 45]
+
+
+def test_agentperf_v2_rejects_turn_without_max_tokens(tmp_path: Path) -> None:
+    loader = _loader(_write(tmp_path, [120, None]), "agentperf-v2")
+    with pytest.raises(
+        DagLoadError, match=r"requires max_tokens on every request turn.*s1\[1\]"
+    ):
+        loader.load_dataset()
+
+
+@pytest.mark.parametrize(
+    "scenario",
+    [
+        param(None, id="no_scenario"),
+        param("inferencex-agentx-mvp", id="other_scenario"),
+    ],
+)  # fmt: skip
+def test_missing_max_tokens_allowed_without_requirement(
+    tmp_path: Path, scenario: str | None
+) -> None:
+    loader = _loader(_write(tmp_path, [None, None]), scenario)
+    convs = loader.convert_to_conversations(loader.load_dataset())
+    assert [turn.max_tokens for turn in convs[0].turns] == [None, None]
+
+
+def test_agentperf_v2_skips_turns_without_requests(tmp_path: Path) -> None:
+    rows = [
+        {"session_id": "orch", "orchestrator": True, "spawns": ["child"]},
+        {
+            "session_id": "child",
+            "turns": [
+                {"messages": [{"role": "user", "content": "hi"}], "max_tokens": 8}
+            ],
+        },
+    ]
+    path = tmp_path / "dag.jsonl"
+    path.write_text("".join(json.dumps(row) + "\n" for row in rows))
+    loader = _loader(path, "agentperf-v2")
+    convs = loader.convert_to_conversations(loader.load_dataset())
+    assert {conv.session_id for conv in convs} == {"orch", "child"}
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [
+        param({"ignore_eos": False}, id="ignore_eos_false"),
+        param({"min_tokens": 4}, id="min_tokens"),
+        param({"stop": ["\\n"]}, id="stop"),
+    ],
+)  # fmt: skip
+def test_agentperf_v2_rejects_turn_extra_overriding_length(
+    tmp_path: Path, extra: dict
+) -> None:
+    turn = {
+        "messages": [{"role": "user", "content": "hi"}],
+        "max_tokens": 8,
+        "extra": extra,
+    }
+    path = tmp_path / "dag.jsonl"
+    path.write_text(json.dumps({"session_id": "s1", "turns": [turn]}) + "\n")
+    loader = _loader(path, "agentperf-v2")
+    with pytest.raises(DagLoadError, match=r"overrides it.*s1\[0\]"):
+        loader.load_dataset()