fix(config): propagate export level to benchmark plans

docs/tutorials/multi-run-confidence.md

@@ -722,6 +722,22 @@ aiperf profile \
 
 > Distribution mode requires `--export-level records` or `--export-level raw` because it reads per-request JSONL data. It is rejected with `--export-level summary`.
 
+For YAML configuration, use `benchmark.artifacts.records: [jsonl]`; add
+`benchmark.artifacts.raw: true` to also export raw payloads. Setting both
+`records: false` and `raw: false` selects summary-only export, which is rejected
+for distribution convergence. CLI and YAML configurations use the resolved
+artifact export level for this validation.
+
+For sweeps, every expanded benchmark config is checked, including per-variation
+artifact overrides. A summary-only variation is rejected before execution even
+if the base config uses `records` or `raw`. Conversely, a summary-only base is
+valid when every expanded config enables per-request records. Mixing `records`
+and `raw` export levels across variations is allowed.
+
+Dynamically proposed search configs are checked again before their trials start.
+A summary-only proposal is rejected with its search iteration index, so runtime
+artifact overrides do not bypass the distribution-convergence requirement.
+
 ### Threshold Semantics
 
 For `ci_width` and `cv`, a lower threshold is stricter (harder to converge). For `distribution`, the threshold is a KS test p-value — convergence triggers when `p_value > threshold`, so a higher threshold is stricter. AIPerf logs this at runtime:

src/aiperf/cli_runner/_strategy.py

@@ -25,33 +25,40 @@
 
 if TYPE_CHECKING:
     from aiperf.common.aiperf_logger import AIPerfLogger
-    from aiperf.config import BenchmarkPlan
+    from aiperf.config import BenchmarkConfig, BenchmarkPlan
     from aiperf.orchestrator.convergence.base import ConvergenceCriterion
     from aiperf.orchestrator.strategies import ExecutionStrategy
 
 
 def validate_convergence_config(plan: BenchmarkPlan) -> None:
     """Raise ValueError for invalid adaptive/convergence plan configurations."""
-    from aiperf.common.enums import ExportLevel
-    from aiperf.plugin.enums import ConvergenceCriterionType
-
     if not plan.use_adaptive:
         return
     if plan.trials <= 1:
         raise ValueError(
             "--convergence-metric requires --num-profile-runs > 1. "
             "Set --num-profile-runs to at least 2 to enable adaptive convergence."
         )
+    for index, config in enumerate(plan.configs, start=1):
+        validate_convergence_export(plan, config, label=f"benchmark config {index}")
+
+
+def validate_convergence_export(
+    plan: BenchmarkPlan, config: BenchmarkConfig, *, label: str
+) -> None:
+    """Reject summary-only exports when distribution convergence needs request data."""
+    from aiperf.common.enums import ExportLevel
+    from aiperf.plugin.enums import ConvergenceCriterionType
+
     convergence = plan.multi_run.convergence
-    assert convergence is not None  # use_adaptive guards this
-    if (
-        convergence.mode == ConvergenceCriterionType.DISTRIBUTION
-        and plan.export_level == ExportLevel.SUMMARY
-    ):
+    if convergence is None or convergence.mode != ConvergenceCriterionType.DISTRIBUTION:
+        return
+    if config.artifacts.export_level == ExportLevel.SUMMARY:
         raise ValueError(
             "--convergence-mode distribution requires per-request JSONL data, "
-            "but --export-level is set to 'summary'. "
-            "Use --export-level records or --export-level raw."
+            f"but {label} has export level 'summary'. "
+            "Enable benchmark.artifacts.records: [jsonl] for this config "
+            "(CLI: --export-level records or --export-level raw)."
         )
 
 

src/aiperf/config/loader/plan.py

@@ -113,6 +113,7 @@ def _assemble_plan_from_aiperf_config(
         set_consistent_seed=config.multi_run.set_consistent_seed,
         disable_warmup_after_first=config.multi_run.disable_warmup_after_first,
         no_sweep_table=config.no_sweep_table,
+        export_level=config.benchmark.artifacts.export_level,
         multi_run=config.multi_run,
         sweep=config.sweep,
         failure_policy=None,

src/aiperf/orchestrator/orchestrator.py

@@ -475,6 +475,7 @@ async def execute_adaptive_search(
         for it via :meth:`_run_independent_cell`, feed results back to the
         planner, write search_history.json incrementally.
         """
+        from aiperf.cli_runner._strategy import validate_convergence_export
         from aiperf.exporters.search_history import write_search_history
         from aiperf.orchestrator.search_planner import write_search_checkpoint
 
@@ -527,6 +528,9 @@ async def _flush_history(reason: str | None) -> None:
                 await _flush_history(reason)
                 return all_results
             cfg, variation = proposal
+            validate_convergence_export(
+                plan, cfg, label=f"search iteration {variation.index}"
+            )
             strategy = _build_strategy(plan)
             strategy.validate_config(cfg)
 

tests/unit/config/test_benchmark_plan.py

@@ -10,12 +10,16 @@
 from pydantic import ValidationError
 from pytest import param
 
+from aiperf.cli_runner._strategy import validate_convergence_config
+from aiperf.common.enums import ExportLevel
 from aiperf.config import (
     AIPerfConfig,
     BenchmarkConfig,
     BenchmarkPlan,
     BenchmarkRun,
 )
+from aiperf.config.flags import CLIConfig
+from aiperf.config.flags.resolver import resolve_config
 from aiperf.config.loader import build_benchmark_plan, load_benchmark_plan
 from aiperf.config.resolution.plan import FailurePolicy
 from aiperf.config.sweep import (
@@ -249,6 +253,138 @@ def test_multi_run_only(self) -> None:
         assert plan.confidence_level == 0.99
         assert not plan.is_single_run
 
+    @pytest.mark.parametrize("source", [param("cli"), param("yaml")])
+    @pytest.mark.parametrize(
+        "export_level",
+        [param(ExportLevel.SUMMARY), param(ExportLevel.RECORDS), param(ExportLevel.RAW)],
+    )  # fmt: skip
+    def test_distribution_convergence_respects_export_level(
+        self, source: str, export_level: ExportLevel, tmp_path: Path
+    ) -> None:
+        if source == "cli":
+            config = resolve_config(
+                CLIConfig(
+                    model_names=["test-model"],
+                    url="http://localhost:8000",
+                    num_profile_runs=3,
+                    convergence_metric="request_latency",
+                    convergence_mode="distribution",
+                    export_level=export_level,
+                )
+            )
+            plan = build_benchmark_plan(config)
+        else:
+            path = tmp_path / "benchmark.yaml"
+            path.write_text(
+                yaml.safe_dump(
+                    {
+                        "benchmark": {
+                            **_MINIMAL_CONFIG_KWARGS,
+                            "artifacts": {
+                                "records": False
+                                if export_level == ExportLevel.SUMMARY
+                                else ["jsonl"],
+                                "raw": export_level == ExportLevel.RAW,
+                            },
+                        },
+                        "multi_run": {
+                            "num_runs": 3,
+                            "convergence": {
+                                "metric": "request_latency",
+                                "mode": "distribution",
+                            },
+                        },
+                    }
+                )
+            )
+            plan = load_benchmark_plan(path, substitute_env=False)
+
+        assert plan.configs[0].artifacts.export_level == export_level
+        assert plan.export_level == export_level
+        if export_level == ExportLevel.SUMMARY:
+            with pytest.raises(ValueError, match="requires per-request JSONL data"):
+                validate_convergence_config(plan)
+        else:
+            validate_convergence_config(plan)
+
+    @pytest.mark.parametrize(
+        "sweep_type", [param("grid"), param("zip"), param("scenarios")]
+    )  # fmt: skip
+    @pytest.mark.parametrize(
+        "base_records, swept_field, values, expected_levels",
+        [
+            param(
+                True, "records", [["jsonl"], False], ["records", "summary"],
+                id="summary-last",
+            ),
+            param(
+                True, "records", [False, ["jsonl"]], ["summary", "records"],
+                id="summary-first",
+            ),
+            param(
+                False, "records", [["jsonl"], ["jsonl"]], ["records", "records"],
+                id="summary-base-overridden",
+            ),
+            param(
+                True, "raw", [False, True], ["records", "raw"],
+                id="mixed-records-raw",
+            ),
+        ],
+    )  # fmt: skip
+    def test_distribution_convergence_checks_expanded_sweep_exports(
+        self,
+        sweep_type: str,
+        base_records: bool,
+        swept_field: str,
+        values: list[object],
+        expected_levels: list[str],
+        tmp_path: Path,
+    ) -> None:
+        sweep: dict[str, object] = {
+            "type": sweep_type,
+            "iteration_order": "independent",
+        }
+        if sweep_type == "scenarios":
+            sweep["runs"] = [
+                {"name": f"point-{i}", "benchmark": {"artifacts": {swept_field: value}}}
+                for i, value in enumerate(values)
+            ]
+        else:
+            sweep["parameters"] = {f"artifacts.{swept_field}": values}
+        path = tmp_path / "sweep.yaml"
+        path.write_text(
+            yaml.safe_dump(
+                {
+                    "benchmark": {
+                        **_MINIMAL_CONFIG_KWARGS,
+                        "artifacts": {
+                            "records": ["jsonl"] if base_records else False,
+                            "raw": False,
+                        },
+                    },
+                    "sweep": sweep,
+                    "multi_run": {
+                        "num_runs": 3,
+                        "convergence": {
+                            "metric": "request_latency",
+                            "mode": "distribution",
+                        },
+                    },
+                }
+            )
+        )
+        plan = load_benchmark_plan(path, substitute_env=False)
+
+        assert [cfg.artifacts.export_level for cfg in plan.configs] == expected_levels
+        if "summary" in expected_levels:
+            with pytest.raises(
+                ValueError, match="requires per-request JSONL data"
+            ) as exc:
+                validate_convergence_config(plan)
+            assert f"config {expected_levels.index('summary') + 1}" in str(exc.value)
+        else:
+            validate_convergence_config(plan)
+
     def test_grid_sweep(self) -> None:
         config = _make_aiperf_config(
             sweep={

tests/unit/orchestrator/test_adaptive_convergence_exports.py

@@ -0,0 +1,126 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Validate dynamically proposed exports without running an optimizer or benchmark."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from pytest import param
+
+from aiperf.cli_runner._strategy import validate_convergence_config
+from aiperf.config import AIPerfConfig, BenchmarkConfig
+from aiperf.config.loader import build_benchmark_plan
+from aiperf.config.sweep import SweepVariation, _set_nested_value
+from aiperf.orchestrator.executor import RunExecutor
+from aiperf.orchestrator.orchestrator import MultiRunOrchestrator
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "mode, records, raw_values, rejected_index",
+    [
+        param("distribution", False, [0], 0, id="summary-first"),
+        param("distribution", False, [1, 0], 1, id="summary-after-valid-point"),
+        param("distribution", False, [1, 1], None, id="raw-points"),
+        param("distribution", True, [0, 1], None, id="mixed-records-raw"),
+        param("ci_width", False, [1, 0], None, id="ci-width-allows-summary"),
+        param("cv", False, [1, 0], None, id="cv-allows-summary"),
+        param(None, False, [1, 0], None, id="fixed-trials-allow-summary"),
+    ],
+)  # fmt: skip
+async def test_adaptive_search_validates_each_proposed_export(
+    mode: str | None,
+    records: bool,
+    raw_values: list[int],
+    rejected_index: int | None,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Reject an invalid proposal before any of its trials can be dispatched."""
+    config = AIPerfConfig.model_validate(
+        {
+            "benchmark": {
+                "models": ["test-model"],
+                "endpoint": {"urls": ["http://localhost:8000/v1/chat/completions"]},
+                "datasets": [{"name": "default", "type": "synthetic"}],
+                "phases": [
+                    {
+                        "name": "profiling",
+                        "type": "concurrency",
+                        "concurrency": 1,
+                        "requests": 10,
+                    }
+                ],
+                "artifacts": {"records": ["jsonl"] if records else False, "raw": True},
+            },
+            "multi_run": {
+                "num_runs": 3,
+                "convergence": {"metric": "request_latency", "mode": mode}
+                if mode is not None
+                else None,
+            },
+            "sweep": {
+                "type": "adaptive_search",
+                "search_space": [
+                    {"path": "artifacts.raw", "lo": 0, "hi": 1, "kind": "int"}
+                ],
+                "objectives": [
+                    {
+                        "metric": "output_token_throughput",
+                        "stat": "avg",
+                        "direction": "maximize",
+                    }
+                ],
+                "max_iterations": 10,
+            },
+        }
+    )
+    plan = build_benchmark_plan(config)
+    validate_convergence_config(plan)
+
+    proposals = []
+    for index, raw in enumerate(raw_values):
+        cfg_dict = plan.configs[0].model_dump(mode="python", exclude_none=True)
+        _set_nested_value(cfg_dict, "artifacts.raw", raw)
+        proposals.append(
+            (
+                BenchmarkConfig.model_validate(cfg_dict),
+                SweepVariation(
+                    index=index,
+                    label=f"search_iter_{index:04d}",
+                    values={"artifacts.raw": raw},
+                ),
+            )
+        )
+    planner = MagicMock()
+    planner.history.return_value = []
+    planner.ask.side_effect = [*proposals, None]
+    planner.iter_count = len(proposals)
+    planner.convergence_reason.return_value = "max_iterations"
+
+    orchestrator = MultiRunOrchestrator(base_dir=tmp_path)
+    run_cell = AsyncMock(return_value=([], False))
+    monkeypatch.setattr(orchestrator, "_run_independent_cell", run_cell)
+    monkeypatch.setattr(
+        "aiperf.exporters.search_history.write_search_history", MagicMock()
+    )
+    monkeypatch.setattr(
+        "aiperf.orchestrator.search_planner.write_search_checkpoint", MagicMock()
+    )
+    executor = MagicMock(spec=RunExecutor)
+    if rejected_index is not None:
+        with pytest.raises(
+            ValueError,
+            match=f"search iteration {rejected_index} has export level 'summary'",
+        ):
+            await orchestrator.execute(plan, executor, search_planner=planner)
+    else:
+        await orchestrator.execute(plan, executor, search_planner=planner)
+
+    executed_count = rejected_index if rejected_index is not None else len(proposals)
+    assert run_cell.await_count == executed_count
+    assert [
+        call.kwargs["variation"].index for call in run_cell.await_args_list
+    ] == list(range(executed_count))
+    assert planner.tell.call_count == executed_count