[Reporting] Use shared inference reports for Pareto

doc/pareto-frontiers.rst

@@ -5,6 +5,10 @@ The ``pareto`` scenario report writes ``pareto.html`` in the scenario results di
 metrics. Successful iterations and sweep steps supply the points: one test case can contribute several points,
 and several cases can contribute to one frontier.
 
+Point details show serving instances, nodes per instance, and TP/PP/DP within each instance. Disaggregated
+runs show separate prefill and decode layouts and whether they share nodes; combined serving uses the
+``Serving`` label.
+
 Usage Example
 -------------
 

doc/reporting.rst

@@ -176,6 +176,26 @@ Report registration is done via ``Registry`` class:
 
    Registry().add_scenario_report("per_test", PerTestReporter, ReportConfig(enable=True))
 
+Inference Report Providers
+--------------------------
+
+Test definitions can implement ``InferenceReportProvider.inference_reports(system, tr)`` from
+``cloudai.report_generator.inference_report``. It returns a ``list[InferenceReport]`` with one entry per
+independent benchmark measurement, or an empty list when no results are available.
+
+Each report contains an ``InferenceConfig`` with resolved model, benchmark, load, hardware, and serving-role
+settings, plus ``MetricObservation`` outputs with their units and dimensions. Multiple measurements from the
+same execution have separate reports, even when their metric names are identical. Unknown optional settings
+remain ``None``.
+
+vLLM and SGLang return one report per execution.
+The Pareto reporter treats each returned report as one candidate point and applies ``match`` to its metric
+observations. ``metric_observations()`` remains available and returns the combined observations from these reports.
+
+Each serving role records its instance count and assigned nodes per instance. An instance is one serving launch,
+which can span nodes; TP, PP, and DP describe parallelism within it. ``shared_nodes`` identifies overlapping
+prefill/decode placement. A combined prefill/decode instance is displayed as ``Serving``.
+
 .. _reporting-configuration-implementation:
 
 Reporting Configuration Implementation

src/cloudai/report_generator/inference_report.py

@@ -0,0 +1,128 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# 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.
+
+from __future__ import annotations
+
+import abc
+import dataclasses
+import typing
+
+import cloudai.metrics
+
+if typing.TYPE_CHECKING:
+    import cloudai.core
+
+
+@dataclasses.dataclass(kw_only=True)
+class InferenceRoleConfig:
+    """Configured serving layout for one role, such as prefill or decode."""
+
+    model: str | None = None
+    num_nodes: int | None = None
+    num_instances: int | None = None
+    nodes_per_instance: int | None = None
+    instances_per_node: int | None = None
+    gpu_ids: list[int] | None = None
+    tensor_parallel_size: int | None = None
+    pipeline_parallel_size: int | None = None
+    data_parallel_size: int | None = None
+    expert_parallel: bool | None = None
+    precision: str | None = None
+    quantization: str | None = None
+
+
+@dataclasses.dataclass(kw_only=True)
+class InferenceConfig:
+    """Resolved input settings for one measurement; unknown settings remain None."""
+
+    model: str
+    num_nodes: int
+    backend: str | None = None
+    benchmark: str | None = None
+    phase: str | None = None
+    container_image: str | None = None
+    gpus_per_node: int | None = None
+    serving_gpu_count: int | None = None
+    input_tokens: int | None = None
+    output_tokens: int | None = None
+    concurrency: int | None = None
+    users: int | None = None
+    request_count: int | None = None
+    request_rate: float | None = None
+    prefill: InferenceRoleConfig | None = None
+    decode: InferenceRoleConfig | None = None
+    shared_nodes: bool | None = None
+
+    def context(self) -> dict[str, typing.Any]:
+        """Display resolved input settings without inspecting workload-specific configurations."""
+        labels = {
+            "model": "Model",
+            "num_nodes": "Nodes",
+            "backend": "Backend",
+            "container_image": "Container image",
+            "serving_gpu_count": "Serving GPUs",
+            "input_tokens": "Input tokens",
+            "output_tokens": "Output tokens",
+            "concurrency": "Concurrency",
+            "users": "Users",
+            "request_count": "Requests",
+            "request_rate": "Request rate",
+            "shared_nodes": "Prefill and decode share nodes",
+        }
+        context = {label: getattr(self, key) for key, label in labels.items() if getattr(self, key) is not None}
+        if self.gpus_per_node is not None and self.gpus_per_node > 0:
+            context["Allocated GPUs"] = self.num_nodes * self.gpus_per_node
+        role_labels = {
+            "model": "Model",
+            "num_nodes": "Nodes",
+            "num_instances": "Instances",
+            "nodes_per_instance": "Nodes per instance",
+            "instances_per_node": "Instances per node",
+            "gpu_ids": "GPU IDs",
+            "tensor_parallel_size": "TP",
+            "pipeline_parallel_size": "PP",
+            "data_parallel_size": "DP",
+            "expert_parallel": "Expert parallel",
+            "precision": "Precision",
+            "quantization": "Quantization",
+        }
+        decode_label = "Decode" if self.prefill is not None else "Serving"
+        for name, role in (("Prefill", self.prefill), (decode_label, self.decode)):
+            if role is not None:
+                context.update(
+                    {
+                        f"{name} / {label}": getattr(role, key)
+                        for key, label in role_labels.items()
+                        if getattr(role, key) is not None
+                    }
+                )
+        return context
+
+
+@dataclasses.dataclass(kw_only=True)
+class InferenceReport:
+    """One independently measured configuration, including all metric dimensions."""
+
+    config: InferenceConfig
+    observations: list[cloudai.metrics.MetricObservation] = dataclasses.field(default_factory=list)
+
+
+class InferenceReportProvider(abc.ABC):
+    """Interface implemented by inference test definitions."""
+
+    @abc.abstractmethod
+    def inference_reports(self, system: cloudai.core.System, tr: cloudai.core.TestRun) -> list[InferenceReport]:
+        """Return one report per measurement, or an empty list when no results are available."""

src/cloudai/report_generator/pareto_report.py

@@ -32,6 +32,7 @@
 import cloudai.core
 import cloudai.metrics
 import cloudai.models.scenario
+import cloudai.report_generator.inference_report as inference_report
 
 
 class MetricSelector(pydantic.BaseModel):
@@ -162,6 +163,7 @@ class RunObservations:
     observations: list[cloudai.metrics.MetricObservation] = dataclasses.field(default_factory=list)
     error: str = ""
     config: str = ""
+    inference_config: inference_report.InferenceConfig | None = None
 
     @property
     def label(self) -> str:
@@ -178,50 +180,6 @@ def format_number(value: float) -> str:
     return f"{value:,.2f}".rstrip("0").rstrip(".")
 
 
-def inference_context(tr: cloudai.core.TestRun, system: cloudai.core.System) -> dict[str, typing.Any]:
-    """Summarize explicitly configured inference settings, including serving roles."""
-    labels = {
-        "model": "Model",
-        "max_concurrency": "Concurrency",
-        "concurrency": "Concurrency",
-        "random_input_len": "Input tokens",
-        "random_input": "Input tokens",
-        "random_output_len": "Output tokens",
-        "random_output": "Output tokens",
-        "num_prompts": "Requests",
-        "request_rate": "Request rate",
-        "tensor_parallel_size": "TP",
-        "tp_size": "TP",
-        "pipeline_parallel_size": "PP",
-        "pp_size": "PP",
-        "data_parallel_size": "DP",
-        "dp_size": "DP",
-        "enable_expert_parallel": "Expert parallel",
-        "dtype": "Precision",
-        "quantization": "Quantization",
-        "num_nodes": "Nodes",
-        "gpu_ids": "GPU IDs",
-    }
-    context: dict[str, typing.Any] = {"Workload": tr.test.test_template_name, "Nodes": tr.nnodes}
-    gpus_per_node = getattr(system, "gpus_per_node", None)
-    if gpus_per_node is not None and gpus_per_node > 0:
-        context["Allocated GPUs"] = tr.nnodes * gpus_per_node
-    if serving_gpus := getattr(tr.test, "serving_gpu_count", None):
-        context["Serving GPUs"] = serving_gpus
-
-    def collect(values: dict[str, typing.Any], scope: str = "") -> None:
-        for key, value in values.items():
-            if isinstance(value, dict):
-                collect(value, f"{scope}{key.replace('_', ' ').title()} / ")
-            elif key in labels and value is not None:
-                context[f"{scope}{labels[key]}"] = value
-
-    definition = tr.test.model_dump(mode="json", exclude_none=True)
-    collect(definition.get("cmd_args", {}))
-    collect(definition.get("bench_cmd_args", {}))
-    return context
-
-
 class ParetoReporter(cloudai.core.Reporter):
     """Render all successful iterations and sweep steps in selected cases."""
 
@@ -253,10 +211,10 @@ def collect_runs(self) -> list[RunObservations]:
                     tr.current_iteration = int(iteration.name)
                     tr.step = int(path.name) if case.is_dse_job else 0
                     tr.output_path = path
-                    runs.append(self._collect_run(tr))
+                    runs.extend(self._collect_run(tr))
         return runs
 
-    def _collect_run(self, tr: cloudai.core.TestRun) -> RunObservations:
+    def _collect_run(self, tr: cloudai.core.TestRun) -> list[RunObservations]:
         run = RunObservations(tr)
         try:
             dump = tr.output_path / "test-run.toml"
@@ -278,11 +236,31 @@ def _collect_run(self, tr: cloudai.core.TestRun) -> RunObservations:
             status = tr.test.was_run_successful(tr)
             if not status.is_successful:
                 raise ValueError(status.error_message or "Execution failed or incomplete")
+            if isinstance(tr.test, inference_report.InferenceReportProvider):
+                reports = tr.test.inference_reports(self.system, tr)
+                return [
+                    dataclasses.replace(
+                        run,
+                        observations=report.observations,
+                        inference_config=report.config,
+                    )
+                    for report in reports
+                ] or [dataclasses.replace(run, error="No inference measurements found")]
             run.observations = tr.test.metric_observations(self.system, tr)
         except Exception as exc:
             run.error = str(exc)
             logging.debug("Cannot collect Pareto results at %s: %s", tr.output_path, exc)
-        return run
+        return [run]
+
+    def point_context(self, run: RunObservations) -> dict[str, typing.Any]:
+        context: dict[str, typing.Any] = {"Workload": run.tr.test.test_template_name}
+        if run.inference_config is not None:
+            return {**context, **run.inference_config.context()}
+        context["Nodes"] = run.tr.nnodes
+        gpus_per_node = getattr(self.system, "gpus_per_node", None)
+        if gpus_per_node is not None and gpus_per_node > 0:
+            context["Allocated GPUs"] = run.tr.nnodes * gpus_per_node
+        return context
 
     @staticmethod
     def select_pair(
@@ -308,6 +286,15 @@ def build_sections(self, runs: list[RunObservations]) -> list[dict[str, typing.A
                 for run in runs:
                     if frontier.cases and run.tr.name not in frontier.cases:
                         continue
+                    if (
+                        frontier.match
+                        and not run.error
+                        and not any(
+                            all(obs.dimensions.get(key) == value for key, value in frontier.match.items())
+                            for obs in run.observations
+                        )
+                    ):
+                        continue
                     try:
                         x, y = self.select_pair(run, plot, frontier)
                     except ValueError as exc:
@@ -324,7 +311,7 @@ def build_sections(self, runs: list[RunObservations]) -> list[dict[str, typing.A
                             "y": y.value,
                             "run": run.label,
                             "group": frontier.name,
-                            "context": inference_context(run.tr, self.system),
+                            "context": self.point_context(run),
                             "dimensions": dimensions,
                             "config": run.config,
                             "artifact": run.tr.output_path.relative_to(self.results_root).as_posix(),