[API] Run, get, list experiments

doc/api.rst

@@ -0,0 +1,39 @@
+Python API
+==========
+
+``cloudai.api`` lets a Python program run a scenario and find its saved results.
+Pass configuration file paths as ``pathlib.Path`` objects:
+
+.. code-block:: python
+
+   from pathlib import Path
+
+   import cloudai.api
+
+   system = Path("conf/common/system/example_slurm_cluster.toml")
+   scenario = Path("conf/common/test_scenario/sleep.toml")
+
+   experiment = cloudai.api.run_experiment(
+       scenario, system, tests_dir=Path("conf/common/test")
+   )
+   print(experiment.status, experiment.path)
+
+``run_experiment`` waits for the scenario to finish and returns an ``Experiment``
+model with the same data saved in ``experiment.json``. Workload failures appear in
+its status. Configuration and setup errors raise exceptions. Use
+``mode="dry-run"`` to generate commands without running workloads. Use
+``single_sbatch=True`` for a single Slurm allocation. Pass ``hook_dir`` to use a
+custom hook directory. Pass an ``on_start`` callback to receive the initial
+``Experiment`` and a cancellation function after ``experiment.json`` is created
+and before execution begins. The function remains synchronous after calling the
+callback.
+
+``list_experiments(system)`` returns ``(experiment_id, result_directory)`` pairs
+from the system's configured output directory. Directories without
+``experiment.json`` are skipped. A missing output directory gives an empty list;
+experiment file contents are not parsed.
+
+``get_experiment(experiment_id, system)`` loads an ``Experiment`` from that
+system's output directory. Pass a ``Path`` to a result directory to load it
+directly. Invalid IDs and missing, unreadable, or malformed experiment files
+raise ``ValueError``.

doc/index.rst

@@ -13,6 +13,7 @@ This document contains the following chapters:
 
    Tutorial
    USER_GUIDE
+   api
    reporting
    systems
    workloads/index
@@ -21,6 +22,7 @@ This document contains the following chapters:
 
 - :doc:`Tutorial`
 - :doc:`USER_GUIDE`
+- :doc:`api`
 - :doc:`reporting`
 - :doc:`systems`
 - :doc:`workloads/index`

src/cloudai/_core/runner.py

@@ -16,6 +16,7 @@
 
 import datetime
 import logging
+import warnings
 from types import FrameType
 from typing import Optional
 
@@ -44,18 +45,42 @@ class Runner:
         test_scenario (TestScenario): The test scenario to be executed.
     """
 
-    def __init__(self, mode: str, system: System, test_scenario: TestScenario):
+    def __init__(
+        self,
+        mode: str,
+        system: System,
+        test_scenario: TestScenario,
+        *,
+        runner_class: type[BaseRunner] | None = None,
+    ):
         logging.info(f"Initializing Runner [{mode.upper()}] mode")
-        self.runner = self.create_runner(mode, system, test_scenario)
-
-    def create_runner(self, mode: str, system: System, test_scenario: TestScenario) -> BaseRunner:
+        if runner_class is None:
+            warnings.warn(
+                "Runner.create_runner() compatibility with its former three-argument signature is deprecated and "
+                "will be removed in CloudAI 1.9.1.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            self.runner = self.create_runner(mode, system, test_scenario)
+        else:
+            self.runner = self.create_runner(mode, system, test_scenario, runner_class=runner_class)
+
+    def create_runner(
+        self,
+        mode: str,
+        system: System,
+        test_scenario: TestScenario,
+        *,
+        runner_class: type[BaseRunner] | None = None,
+    ) -> BaseRunner:
         """
         Dynamically create a runner instance based on the system's scheduler type.
 
         Args:
             mode (str): The operation mode ('dry-run', 'run').
             system (System): The system configuration.
             test_scenario (TestScenario): The test scenario to run.
+            runner_class (type[BaseRunner] | None): Override for this invocation.
 
         Returns:
             BaseRunner: A runner instance suitable for the system.
@@ -65,11 +90,12 @@ def create_runner(self, mode: str, system: System, test_scenario: TestScenario)
         """
         registry = Registry()
         scheduler_type = system.scheduler.lower()
-        if scheduler_type not in registry.runners_map:
-            msg = f"No runner registered for scheduler: {scheduler_type}"
-            logging.error(msg)
-            raise NotImplementedError(msg)
-        runner_class = registry.runners_map[scheduler_type]
+        if runner_class is None:
+            if scheduler_type not in registry.runners_map:
+                msg = f"No runner registered for scheduler: {scheduler_type}"
+                logging.error(msg)
+                raise NotImplementedError(msg)
+            runner_class = registry.runners_map[scheduler_type]
         logging.info(f"Creating {runner_class.__name__}")
 
         if not system.output_path.exists():

src/cloudai/api.py

@@ -0,0 +1,97 @@
+# 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 collections.abc
+import pathlib
+
+import pydantic
+
+import cloudai.core
+import cloudai.handlers
+import cloudai.models.output
+
+
+def run_experiment(
+    scenario: pathlib.Path,
+    system: pathlib.Path,
+    *,
+    mode: str = "run",
+    single_sbatch: bool = False,
+    tests_dir: pathlib.Path | None = None,
+    hook_dir: pathlib.Path | None = None,
+    on_start: (
+        collections.abc.Callable[[cloudai.models.output.Experiment, collections.abc.Callable[[], None]], None] | None
+    ) = None,
+) -> cloudai.models.output.Experiment:
+    """
+    Run a scenario synchronously and return its experiment snapshot.
+
+    Configuration and setup errors raise exceptions. Workload failures are
+    reflected in the returned experiment status.
+    """
+    parsed_system, tests, parsed_scenario = cloudai.handlers.load_experiment(
+        scenario.expanduser().resolve(),
+        system.expanduser().resolve(),
+        tests_dir=tests_dir,
+        hook_dir=hook_dir,
+    )
+    runner = cloudai.handlers.create_experiment_runner(
+        parsed_system,
+        parsed_scenario,
+        mode=mode,
+        single_sbatch=single_sbatch,
+    )
+    cloudai.handlers.execute_experiment(runner, tests, on_start=on_start)
+    return runner.runner.experiment_output.snapshot()
+
+
+def list_experiments(system: pathlib.Path) -> list[tuple[str, pathlib.Path]]:
+    """List saved experiment IDs and result directories for a system."""
+    output_dir = cloudai.core.Parser.parse_system(system.expanduser().resolve()).output_path.expanduser().resolve()
+    if not output_dir.exists():
+        return []
+    if not output_dir.is_dir():
+        raise NotADirectoryError(output_dir)
+
+    return [
+        (experiment_file.parent.name, experiment_file.parent)
+        for experiment_file in sorted(output_dir.glob("*/experiment.json"))
+        if experiment_file.is_file()
+    ]
+
+
+def get_experiment(exp_id: str | pathlib.Path, system: str | pathlib.Path) -> cloudai.models.output.Experiment:
+    """Load an experiment by ID or result directory."""
+    parsed_system = cloudai.core.Parser.parse_system(pathlib.Path(system).expanduser().resolve())
+    system_output = parsed_system.output_path.expanduser().resolve()
+
+    experiment_path = system_output / exp_id if isinstance(exp_id, str) else exp_id
+    experiment_path = experiment_path.expanduser().resolve()
+    experiment_file = experiment_path / "experiment.json"
+    if experiment_path.parent != system_output or not experiment_path.is_dir() or not experiment_file.is_file():
+        raise ValueError(f"Invalid experiment ID: {experiment_path!r}")
+
+    try:
+        experiment_data = experiment_file.read_text(encoding="utf-8")
+    except (OSError, ValueError) as error:
+        raise ValueError(f"Invalid experiment ID: {experiment_path!r}") from error
+
+    try:
+        exp = cloudai.models.output.Experiment.model_validate_json(experiment_data)
+    except pydantic.ValidationError as error:
+        raise ValueError(f"Cannot read experiment: {experiment_path!r}") from error
+
+    return exp

src/cloudai/cli/cli.py

@@ -17,17 +17,69 @@
 import argparse
 import logging
 import logging.config
+import signal
+import types
 from pathlib import Path
+from typing import Callable
 
 import click
 
-from cloudai.handlers import (
-    handle_dry_run_and_run,
-    handle_generate_report,
-    handle_install_and_uninstall,
-    handle_list_registered_items,
-    handle_verify_all_configs,
-)
+import cloudai.core
+import cloudai.handlers
+
+
+def register_signal_handlers(signal_handler: Callable) -> None:
+    """Register the CLI's termination signals for a running experiment."""
+    for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT):
+        signal.signal(sig, signal_handler)
+
+
+def handle_dry_run_and_run(args: argparse.Namespace) -> int:
+    """Translate CLI options and experiment outcomes into a process exit code."""
+    try:
+        system, tests, scenario = cloudai.handlers.load_experiment(
+            args.test_scenario, args.system_config, tests_dir=args.tests_dir, hook_dir=args.hook_dir
+        )
+        if args.output_dir is not None:
+            system.output_path = args.output_dir.absolute()
+        runner = cloudai.handlers.create_experiment_runner(
+            system,
+            scenario,
+            mode=args.mode,
+            single_sbatch=args.single_sbatch,
+        )
+    except (
+        cloudai.core.MissingTestError,
+        cloudai.core.SystemConfigParsingError,
+        cloudai.core.TestConfigParsingError,
+        cloudai.core.TestScenarioParsingError,
+        FileNotFoundError,
+        OSError,
+        ValueError,
+    ) as exc:
+        logging.error(str(exc))
+        return 1
+
+    def on_start(_experiment: object, cancel: Callable[[], None]) -> None:
+        def cancel_on_signal(signum: int, _frame: types.FrameType | None) -> None:
+            logging.info(f"Signal {signum} received, shutting down...")
+            cancel()
+
+        register_signal_handlers(cancel_on_signal)
+
+    try:
+        successful = cloudai.handlers.execute_experiment(
+            runner,
+            tests,
+            enable_cache_without_check=args.enable_cache_without_check,
+            on_start=on_start,
+        )
+    except cloudai.handlers.InstallationError:
+        return 1
+
+    if args.single_sbatch or not any(tr.is_dse_job for tr in scenario.test_runs):
+        return 0
+    return 0 if successful else 1
 
 
 def setup_logging(log_file: str, log_level: str) -> None:
@@ -153,7 +205,7 @@ def install(system_cfg: Path, tests_dir: Path, scenario_cfg: Path, hook_dir: Pat
         hook_dir=hook_dir,
         mode="install",
     )
-    exit(handle_install_and_uninstall(args))
+    exit(cloudai.handlers.handle_install_and_uninstall(args))
 
 
 @main.command()
@@ -167,7 +219,7 @@ def uninstall(system_cfg: Path, tests_dir: Path, scenario_cfg: Path, hook_dir: P
         hook_dir=hook_dir,
         mode="uninstall",
     )
-    exit(handle_install_and_uninstall(args))
+    exit(cloudai.handlers.handle_install_and_uninstall(args))
 
 
 @main.command()
@@ -252,7 +304,7 @@ def generate_report(system_cfg: Path, tests_dir: Path, scenario_cfg: Path, hook_
         hook_dir=hook_dir,
         result_dir=result_dir,
     )
-    exit(handle_generate_report(args))
+    exit(cloudai.handlers.handle_generate_report(args))
 
 
 @main.command()
@@ -268,12 +320,12 @@ def generate_report(system_cfg: Path, tests_dir: Path, scenario_cfg: Path, hook_
 def verify_configs(configs_dir: Path, tests_dir: Path):
     """Verify the configuration TOML files."""
     args = argparse.Namespace(configs_dir=configs_dir, tests_dir=tests_dir)
-    raise click.exceptions.Exit(1 if handle_verify_all_configs(args) else 0)
+    raise click.exceptions.Exit(1 if cloudai.handlers.handle_verify_all_configs(args) else 0)
 
 
 @main.command()
 @click.argument("type", type=click.Choice(["reports", "agents", "reward-functions"], case_sensitive=False))
 @click.option("-v", "--verbose", is_flag=True, default=False, help="Verbose output.")
 def list(type: str, verbose: bool):
     """List available in Registry items."""
-    handle_list_registered_items(type, verbose)
+    cloudai.handlers.handle_list_registered_items(type, verbose)