[Reporting] Unified output for standalone scenarios

doc/reporting.rst

@@ -31,6 +31,23 @@ 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
+-------------------------
+
+CloudAI writes ``experiment.json`` in each scenario's results directory.
+
+The file contains scenario details, the configured system name, test cases, status, timing, and result paths. Standalone
+execution also records each run's process ID, status, iteration, timing, and workload metrics.
+
+CloudAI updates the file when standalone runs start and finish, then finalizes it when scenario execution succeeds or
+fails. Timestamps use UTC; durations use seconds. Unknown timestamps are ``null``. Dry runs also produce scenario and
+test-case details without launching workloads.
+
+Metrics come from ``TestDefinition.metric_observations()``, independently of reporter settings. A completed ordinary
+test with one run also includes those metrics on the test case. Each file update replaces the previous snapshot
+atomically. Metric extraction or write errors produce warnings without affecting execution.
+
+
 .. _general-flow:
 
 General Flow

src/cloudai/_core/base_runner.py

@@ -14,12 +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
 
+import cloudai.models.output
+import cloudai.output
+
 from .base_job import BaseJob
 from .command_gen_strategy import CommandGenStrategy
 from .exceptions import JobFailureError, JobSubmissionError
@@ -69,6 +73,49 @@ 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
+        output_path = self.scenario_root.absolute()
+        experiment = cloudai.models.output.Experiment(
+            id=output_path.name,
+            name=self.test_scenario.name,
+            system_name=self.system.name,
+            status="running",
+            path=str(output_path),
+            start=datetime.datetime.now(datetime.timezone.utc),
+            tests=[
+                cloudai.models.output.Test(
+                    id=str(tr.name),
+                    name=tr.test.name,
+                    description=tr.test.description,
+                    path=str(output_path / str(tr.name)),
+                )
+                for tr in self.test_scenario.test_runs
+            ],
+        )
+        self.experiment_output = cloudai.output.ExperimentOutput(experiment, output_path)
+
+    def get_run_output(
+        self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None
+    ) -> cloudai.models.output.Run | None:
+        """Normalize one logical execution; result is absent at submission."""
+        return None
+
+    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":
+            return
+        for tr in self.completed_test_runs(job):
+            run = self.get_run_output(job, tr, result)
+            if run is not None:
+                self.experiment_output.update_run(str(tr.name), run)
+        self.experiment_output.write()
+
+    def finish_output(self, successful: bool) -> 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 +157,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 +299,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 +313,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.experiment_output.write()
+        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,90 @@
+# 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
+from typing import Literal
+
+import pydantic
+
+Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"]
+
+
+class Dimension(pydantic.BaseModel):
+    """One coordinate of a metric measurement."""
+
+    name: str
+    value: str
+    unit: str = ""
+    is_x: bool = False
+
+
+class Metric(pydantic.BaseModel):
+    """A canonical measurement at a dimension point."""
+
+    name: str
+    value: str | int | pydantic.FiniteFloat | bool
+    unit: str = ""
+    dimensions: list[Dimension] = pydantic.Field(default_factory=list)
+
+
+class Run(pydantic.BaseModel):
+    """A logical execution with normalized metadata and original measurements."""
+
+    path: str
+    jobid: str
+    status: Status = "unknown"
+    metrics: list[Metric] = pydantic.Field(default_factory=list)
+    start: datetime.datetime | None = None
+    finish: datetime.datetime | None = None
+    duration: pydantic.FiniteFloat | None = None
+    iteration: int | None = None
+    step: int | None = None
+
+
+class DSE(pydantic.BaseModel):
+    """Search space and recommendation for one test case."""
+
+    space: dict[str, list[str | int | pydantic.FiniteFloat]]
+    best_config: dict[str, str | int | pydantic.FiniteFloat] | None = None
+    best_step: int | None = None
+
+
+class Test(pydantic.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] = pydantic.Field(default_factory=list)
+    runs: list[Run] = pydantic.Field(default_factory=list)
+    dse: DSE | None = None
+
+
+class Experiment(pydantic.BaseModel):
+    """Full snapshot spanning the entire scenario, including all DSE trials."""
+
+    id: str
+    name: str
+    system_name: str
+    description: str | None = None
+    status: Status = "pending"
+    path: str
+    start: datetime.datetime | None = None
+    finish: datetime.datetime | None = None
+    duration: pydantic.FiniteFloat | None = None
+    tests: list[Test] = pydantic.Field(default_factory=list)