Add unified results for completed Slurm scenarios

doc/reporting.rst

@@ -31,6 +31,21 @@ Per-test reports are linked to a particular workload type (e.g. ``NcclTest``). A
 To list all available reports, users can use ``cloudai list-reports``. Use verbose output to also print report configurations.
 
 
+Unified experiment output
+-------------------------
+
+Ordinary Slurm scenarios write ``experiment.json`` in the scenario results directory when execution finishes or fails.
+The file contains experiment metadata, test cases, submitted runs, statuses, timing, and canonical metrics from
+``TestDefinition.metric_observations()``. NCCL and NIXLBench provide these metrics; other workloads have empty metric lists.
+Test-level metrics are arithmetic means of successful iterations at matching metric and dimension points. Per-run
+measurements retain their original values, and missing measurements are not treated as zero.
+
+Experiment timing covers scenario execution, including gaps between jobs. Run timestamps without timezone information
+are null. Metric extraction and output-write failures produce warnings without changing execution behavior. The file is
+replaced atomically and is independent of reporter configuration. Dry runs, DSE, and single-sbatch execution do not produce
+this artifact.
+
+
 .. _general-flow:
 
 General Flow

src/cloudai/_core/base_runner.py

@@ -14,11 +14,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import logging
 import time
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Dict, List
+from typing import TYPE_CHECKING, Dict, List
+
+if TYPE_CHECKING:
+    from cloudai.models.output import Run
+    from cloudai.output import ExperimentOutput
 
 from .base_job import BaseJob
 from .command_gen_strategy import CommandGenStrategy
@@ -69,6 +74,34 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu
         self.testrun_to_job_map: Dict[TestRun, BaseJob] = {}
         logging.debug(f"{self.__class__.__name__} initialized")
         self.shutting_down = False
+        self.experiment_output: ExperimentOutput | None = self.create_experiment_output()
+
+    def create_experiment_output(self) -> "ExperimentOutput | None":
+        return None
+
+    def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run":
+        """Normalize one logical execution; result is absent at submission."""
+        raise NotImplementedError
+
+    def completed_test_runs(self, job: BaseJob) -> list[TestRun]:
+        """Return logical executions represented by a scheduler job."""
+        return [job.test_run]
+
+    def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) -> None:
+        """Capture logical runs before iteration or DSE state advances, then publish a snapshot."""
+        if self.mode == "run" and self.experiment_output is not None:
+            for tr in self.completed_test_runs(job):
+                self.experiment_output.update_run(str(tr.name), self.get_run_output(job, tr, result))
+            self.write_output()
+
+    def write_output(self) -> None:
+        if self.mode == "run" and self.experiment_output is not None:
+            self.experiment_output.write()
+
+    def finish_output(self, successful: bool) -> None:
+        if self.mode == "run" and self.experiment_output is not None:
+            status = "completed" if successful else "failed"
+            self.experiment_output.finish(status=status, finish=datetime.datetime.now(datetime.timezone.utc))
 
     def shutdown(self):
         """Gracefully shut down the runner, terminating all outstanding jobs."""
@@ -110,6 +143,7 @@ def submit_test(self, tr: TestRun):
             job = self._submit_test(tr)
             self.jobs.append(job)
             self.testrun_to_job_map[tr] = job
+            self.update_run_output(job)
         except JobSubmissionError as e:
             logging.error(e)
             exit(1)
@@ -251,6 +285,7 @@ def monitor_jobs(self) -> int:
                 else:
                     if self.test_scenario.job_status_check:
                         job_status_result = self.get_job_status(job)
+                        self.update_run_output(job, job_status_result)
                         if job_status_result.is_successful:
                             successful_jobs_count += 1
                             self.handle_job_completion(job)
@@ -264,6 +299,7 @@ def monitor_jobs(self) -> int:
                             raise JobFailureError(job.test_run.name, error_message, job_status_result.error_message)
                     else:
                         job_status_result = self.get_job_status(job)
+                        self.update_run_output(job, job_status_result)
                         if not job_status_result.is_successful:
                             error_message = (
                                 f"Job {job.id} for test {job.test_run.name} failed: {job_status_result.error_message}"

src/cloudai/_core/runner.py

@@ -79,13 +79,15 @@ def create_runner(self, mode: str, system: System, test_scenario: TestScenario)
 
         return runner_class(mode, system, test_scenario, results_root)
 
-    def run(self):
+    def run(self) -> bool:
         """Run the test scenario using the instantiated runner."""
         try:
             self.runner.run()
             logging.debug("All jobs finished successfully.")
+            return True
         except JobFailureError as exc:
             logging.debug(f"Runner failed JobFailure exception: {exc}", exc_info=True)
+            return False
 
     def cancel_on_signal(
         self,

src/cloudai/cli/handlers.py

@@ -239,10 +239,11 @@ def generate_reports(
             logging.debug(e, exc_info=True)
 
 
-def handle_non_dse_job(runner: Runner, args: argparse.Namespace) -> None:
-    runner.run()
+def handle_non_dse_job(runner: Runner, args: argparse.Namespace) -> bool:
+    successful = runner.run()
     generate_reports(runner.runner.system, runner.runner.test_scenario, runner.runner.scenario_root)
     logging.info("All jobs are complete.")
+    return successful
 
 
 def register_signal_handlers(signal_handler: Callable) -> None:
@@ -352,16 +353,23 @@ def handle_dry_run_and_run(args: argparse.Namespace) -> int:
     register_signal_handlers(runner.cancel_on_signal)
     logging.info(f"Scenario results will be stored at: {runner.runner.scenario_root}")
 
-    has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs)
-    if args.single_sbatch or not has_dse:  # in this mode cases are unrolled using grid search
-        handle_non_dse_job(runner, args)
-        return 0
-
-    if all(tr.is_dse_job for tr in test_scenario.test_runs):
-        return handle_dse_job(runner, args)
-
-    logging.error("Mixing DSE and non-DSE jobs is not allowed.")
-    return 1
+    successful = False
+    try:
+        runner.runner.write_output()
+        has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs)
+        if args.single_sbatch or not has_dse:  # in this mode cases are unrolled using grid search
+            successful = handle_non_dse_job(runner, args)
+            return 0
+
+        if all(tr.is_dse_job for tr in test_scenario.test_runs):
+            result = handle_dse_job(runner, args)
+            successful = result == 0
+            return result
+
+        logging.error("Mixing DSE and non-DSE jobs is not allowed.")
+        return 1
+    finally:
+        runner.runner.finish_output(successful)
 
 
 def handle_generate_report(args: argparse.Namespace) -> int:

src/cloudai/models/output.py

@@ -0,0 +1,89 @@
+# 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 datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field, FiniteFloat
+
+Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"]
+
+
+class Dimension(BaseModel):
+    """One coordinate of a metric measurement."""
+
+    name: str
+    value: str
+    unit: str = ""
+    is_x: bool = False
+
+
+class Metric(BaseModel):
+    """A canonical measurement at a dimension point."""
+
+    name: str
+    value: str | int | FiniteFloat | bool
+    unit: str = ""
+    dimensions: list[Dimension] = Field(default_factory=list)
+
+
+class Run(BaseModel):
+    """A logical execution with normalized metadata and original measurements."""
+
+    path: str
+    jobid: str
+    status: Status = "unknown"
+    metrics: list[Metric] = Field(default_factory=list)
+    start: datetime | None = None
+    finish: datetime | None = None
+    duration: FiniteFloat | None = None
+    iteration: int | None = None
+    step: int | None = None
+
+
+class DSE(BaseModel):
+    """Search space and recommendation for one test case."""
+
+    space: dict[str, list[str | int | FiniteFloat]]
+    best_config: dict[str, str | int | FiniteFloat] | None = None
+    best_step: int | None = None
+
+
+class Test(BaseModel):
+    """A test case with all logical executions and optional DSE metadata."""
+
+    id: str
+    name: str
+    description: str | None = None
+    status: Status = "pending"
+    path: str
+    metrics: list[Metric] = Field(default_factory=list)
+    runs: list[Run] = Field(default_factory=list)
+    dse: DSE | None = None
+
+
+class Experiment(BaseModel):
+    """Full snapshot spanning the entire scenario, including all DSE trials."""
+
+    id: str
+    name: str
+    description: str | None = None
+    status: Status = "pending"
+    path: str
+    start: datetime | None = None
+    finish: datetime | None = None
+    duration: FiniteFloat | None = None
+    tests: list[Test] = Field(default_factory=list)

src/cloudai/output.py

@@ -0,0 +1,123 @@
+# 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.
+
+import datetime
+import logging
+import pathlib
+import tempfile
+
+from cloudai.models import output as output_models
+
+
+class ExperimentOutput:
+    """Collect experiment results and publish snapshots."""
+
+    def __init__(self, experiment: output_models.Experiment, output_path: pathlib.Path) -> None:
+        self.experiment = experiment.model_copy(deep=True)
+        self.output_path = output_path
+
+    def update_run(self, test_id: str, run: output_models.Run) -> None:
+        """Store the latest state of a run, identified by test and output path."""
+        test = next((test for test in self.experiment.tests if test.id == test_id), None)
+        if test is None:
+            raise KeyError(f"Unknown experiment test: {test_id}")
+        recorded = run.model_copy(deep=True)
+        for index, current in enumerate(test.runs):
+            if current.path == run.path:
+                test.runs[index] = recorded
+                self._update_test_status(test)
+                return
+        test.runs.append(recorded)
+        self._update_test_status(test)
+
+    def update_test(self, test: output_models.Test) -> None:
+        """Update a test while retaining its previously recorded runs."""
+        recorded = test.model_copy(deep=True)
+        for index, current in enumerate(self.experiment.tests):
+            if current.id == test.id:
+                incoming_runs = recorded.runs
+                recorded.runs = current.runs
+                self.experiment.tests[index] = recorded
+                for run in incoming_runs:
+                    self.update_run(test.id, run)
+                return
+        self.experiment.tests.append(recorded)
+
+    def snapshot(self) -> output_models.Experiment:
+        """Return an independent snapshot without finalizing the experiment."""
+        full = self.experiment.model_copy(deep=True)
+        self._update_timing(full)
+        for test in full.tests:
+            for run in test.runs:
+                self._update_timing(run)
+        return full
+
+    def write(self) -> None:
+        """Atomically replace experiment.json, warning on failure."""
+        temporary_path: pathlib.Path | None = None
+        try:
+            content = self.snapshot().model_dump_json(indent=2)
+            self.output_path.mkdir(parents=True, exist_ok=True)
+            with tempfile.NamedTemporaryFile(
+                mode="w", encoding="utf-8", dir=self.output_path, prefix=".experiment.json.", delete=False
+            ) as temporary:
+                temporary_path = pathlib.Path(temporary.name)
+                temporary.write(content + "\n")
+            temporary_path.replace(self.output_path / "experiment.json")
+        except Exception as exc:
+            logging.warning("Cannot write experiment output: %s", exc)
+        finally:
+            if temporary_path is not None:
+                try:
+                    temporary_path.unlink(missing_ok=True)
+                except OSError as exc:
+                    logging.warning("Cannot remove temporary experiment output %s: %s", temporary_path, exc)
+
+    def finish(self, status: output_models.Status, finish: datetime.datetime | None) -> None:
+        self.experiment.status = status
+        self.experiment.finish = finish
+        if status == "completed":
+            for test in self.experiment.tests:
+                if test.status not in ("failed", "cancelled"):
+                    test.status = "completed"
+        self._update_timing(self.experiment)
+        self.write()
+
+    @staticmethod
+    def _update_test_status(test: output_models.Test) -> None:
+        statuses = {run.status for run in test.runs}
+        if "failed" in statuses:
+            test.status = "failed"
+        elif "cancelled" in statuses:
+            test.status = "cancelled"
+        elif "running" in statuses:
+            test.status = "running"
+        elif statuses == {"completed"}:
+            test.status = "completed"
+
+    @staticmethod
+    def _update_timing(record: output_models.Experiment | output_models.Run) -> None:
+        for field in ("start", "finish"):
+            value = getattr(record, field)
+            if value is not None:
+                setattr(
+                    record, field, value.astimezone(datetime.timezone.utc) if value.utcoffset() is not None else None
+                )
+        end = record.finish
+        if end is None and record.status == "running":
+            end = datetime.datetime.now(datetime.timezone.utc)
+        if record.start is not None and end is not None:
+            record.duration = max((end - record.start).total_seconds(), 0.0)

src/cloudai/systems/slurm/single_sbatch_runner.py

@@ -191,6 +191,7 @@ def run(self):
         tr = self.test_scenario.test_runs[0]
         job = self._submit_test(tr)
         self.jobs.append(job)
+        self.update_run_output(job)
 
         if self.shutting_down:
             self.system.kill(job)
@@ -208,6 +209,7 @@ def run(self):
         self.handle_dse()
 
         self.on_job_completion(job)
+        self.update_run_output(job)
 
     def handle_dse(self):
         registry = Registry()