[API] Support background experiment execution

doc/api.rst

@@ -47,6 +47,26 @@ Optional keyword arguments:
 * ``enable_cache_without_check=True`` uses installed workload components without
   checking them first, as with the CLI option.
 
+Background execution
+--------------------
+
+Set ``wait=False`` to launch a separate worker process and return a pending
+experiment. The worker continues after the calling process exits. Read the latest
+saved snapshot using the returned path:
+
+.. code-block:: python
+
+   experiment = cloudai.api.run_experiment(scenario, system, wait=False, tests_dir=tests_dir)
+   # Later, including from another process:
+   experiment = cloudai.api.get_experiment(experiment.path)
+   print(experiment.status)
+
+The result directory contains ``experiment.json``, the worker's ``controller.log``,
+and its ``request.json``.
+Keep referenced config files available until the worker has read them. The worker
+loads the installed CloudAI package and plugins in a fresh Python process;
+registrations made only in the caller's memory are not transferred.
+
 Find and read results
 ---------------------
 
@@ -69,7 +89,9 @@ execution constraints without installing workloads, creating a results directory
 or querying cluster availability.
 
 ``run_experiment`` raises exceptions for invalid configuration and setup errors.
-Workload failures appear in the experiment and run statuses.
+Workload failures appear in the experiment and run statuses. For background runs,
+errors after launch are recorded as a failed experiment, with details in
+``controller.log``.
 
 Missing result files raise ``FileNotFoundError``. Invalid experiment JSON raises
 ``pydantic.ValidationError``. Listing also raises for unreadable or invalid result

src/cloudai/api.py

@@ -23,6 +23,7 @@
 def run_experiment(
     scenario: str | Path,
     system: str | Path,
+    wait: bool = True,
     *,
     tests_dir: Path | None = None,
     hook_dir: Path | None = None,
@@ -36,7 +37,8 @@ def run_experiment(
 
     Pass TOML content as str or configuration files as Path. Relative test paths
     require a scenario file. Optional directories have the same meaning as in
-    the CLI. The call waits for execution to finish.
+    the CLI. With wait=False, a separate process executes the experiment and
+    returns an initial pending snapshot; use get_experiment to read later results.
 
     Configuration and setup errors raise exceptions. Workload failures are
     recorded in the returned experiment's status. This function does not invoke
@@ -45,6 +47,7 @@ def run_experiment(
     return cloudai.handlers.run_experiment(
         scenario,
         system,
+        wait,
         tests_dir=tests_dir,
         hook_dir=hook_dir,
         output_dir=output_dir,

src/cloudai/handlers.py

@@ -16,8 +16,12 @@
 
 import copy
 import datetime
+import json
 import logging
+import subprocess
+import sys
 import tempfile
+import threading
 import traceback
 from contextlib import contextmanager
 from pathlib import Path
@@ -321,6 +325,7 @@ def _configuration_files(system: str | Path, scenario: str | Path | None = None)
 def run_experiment(
     scenario: str | Path,
     system: str | Path,
+    wait: bool = True,
     *,
     tests_dir: Path | None = None,
     hook_dir: Path | None = None,
@@ -329,7 +334,7 @@ def run_experiment(
     single_sbatch: bool = False,
     enable_cache_without_check: bool = False,
 ) -> cloudai.models.output.Experiment:
-    """Run an experiment from configuration files or TOML text."""
+    """Run an experiment synchronously or hand it to an independent worker."""
     with _configuration_files(system, scenario) as (system_path, scenario_path):
         parsed_system, tests, parsed_scenario = load_scenario(
             scenario_path, system_path, tests_dir=tests_dir, hook_dir=hook_dir
@@ -357,8 +362,72 @@ def run_experiment(
             parsed_system, parsed_scenario, dry_run=dry_run, single_sbatch=single_sbatch, result_dir=result_dir
         )
         output = runner.runner.experiment_output
-        execute_experiment(runner, tests, enable_cache_without_check)
-        return output.snapshot()
+        if wait:
+            execute_experiment(runner, tests, enable_cache_without_check)
+            return output.snapshot()
+        output.experiment.status = "pending"
+        output.experiment.start = None
+        output.write()
+        initial = get_experiment(result_dir)
+        request = {
+            "system": str(system_path),
+            "scenario": str(scenario_path),
+            "tests_dir": str(tests_dir.resolve()) if tests_dir is not None else None,
+            "hook_dir": str((hook_dir or HOOK_ROOT).resolve()),
+            "output_dir": str(parsed_system.output_path.resolve()),
+            "dry_run": dry_run,
+            "single_sbatch": single_sbatch,
+            "enable_cache_without_check": enable_cache_without_check,
+        }
+        request_path = result_dir / "request.json"
+        try:
+            request_path.write_text(json.dumps(request), encoding="utf-8")
+            command = [
+                sys.executable,
+                "-c",
+                "import logging, pathlib, sys; import cloudai.handlers; "
+                "logging.basicConfig(level=logging.INFO); "
+                "cloudai.handlers.run_background(pathlib.Path(sys.argv[1]))",
+                str(request_path),
+            ]
+            with (result_dir / "controller.log").open("a", encoding="utf-8") as log:
+                process = subprocess.Popen(
+                    command, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True
+                )
+            threading.Thread(target=process.wait, daemon=True).start()
+        except Exception:
+            runner.runner.finish_output(False)
+            raise
+        return initial
+
+
+def run_background(request_path: Path) -> None:
+    """Execute a persisted request in its worker process."""
+    try:
+        request = json.loads(request_path.read_text(encoding="utf-8"))
+        system, tests, scenario = load_scenario(
+            Path(request["scenario"]),
+            Path(request["system"]),
+            tests_dir=Path(request["tests_dir"]) if request["tests_dir"] else None,
+            hook_dir=Path(request["hook_dir"]),
+        )
+        system.output_path = Path(request["output_dir"])
+        runner = create_experiment_runner(
+            system,
+            scenario,
+            dry_run=request["dry_run"],
+            single_sbatch=request["single_sbatch"],
+            result_dir=request_path.parent,
+        )
+        execute_experiment(runner, tests, request["enable_cache_without_check"])
+    except BaseException:
+        logging.exception("Background experiment failed")
+        experiment = get_experiment(request_path.parent)
+        if experiment.status in ("pending", "running"):
+            cloudai.output.ExperimentOutput(experiment, request_path.parent).finish(
+                "failed", datetime.datetime.now(datetime.timezone.utc)
+            )
+        raise
 
 
 def get_experiment(exp: str | Path) -> cloudai.models.output.Experiment:

tests/test_api.py

@@ -63,11 +63,20 @@ def configs(standalone_system: StandaloneSystem) -> tuple[str, str]:
     return scenario, system
 
 
-@pytest.mark.parametrize("successful,as_text", [(True, False), (True, True), (False, True)])
+@pytest.mark.parametrize(
+    "wait,successful,as_text", [(True, True, False), (True, True, True), (True, False, True), (False, True, True)]
+)
 def test_experiment_api(
-    configs: tuple[str, str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch, successful: bool, as_text: bool
+    configs: tuple[str, str],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    wait: bool,
+    successful: bool,
+    as_text: bool,
 ):
     scenario, system = configs
+    process = Mock()
+    monkeypatch.setattr("cloudai.handlers.subprocess.Popen", process)
     monkeypatch.setattr(StandaloneSystem, "is_job_completed", lambda *_: True)
     monkeypatch.setattr(StandaloneSystem, "is_job_running", lambda *_: False)
     monkeypatch.setattr(
@@ -97,7 +106,17 @@ def test_experiment_api(
         system_input = tmp_path / "system.toml"
         scenario_input.write_text(scenario)
         system_input.write_text(system)
-    experiment = cloudai.api.run_experiment(scenario_input, system_input)
+    experiment = cloudai.api.run_experiment(scenario_input, system_input, wait=wait)
+    if not wait:
+        assert experiment.status == "pending"
+        assert experiment.start is None
+        assert cloudai.api.get_experiment(experiment.path) == experiment
+        request_path = Path(experiment.path) / "request.json"
+        assert process.call_args.args[0][-1] == str(request_path)
+        cloudai.handlers.run_background(request_path)
+        experiment = cloudai.api.get_experiment(experiment.path)
+    else:
+        process.assert_not_called()
     assert logging.getLogger().handlers == handlers
     assert signal.getsignal(signal.SIGINT) == sigint
     assert experiment.status == ("completed" if successful else "failed")
@@ -158,12 +177,14 @@ def test_validate_scenario(configs: tuple[str, str], tmp_path: Path):
         cloudai.api.get_experiment(tmp_path / "missing")
 
 
-@pytest.mark.parametrize("failure", ["setup", "submission"])
+@pytest.mark.parametrize("failure", ["setup", "spawn", "worker-parse", "submission"])
 def test_failed_experiment_is_saved(configs: tuple[str, str], monkeypatch: pytest.MonkeyPatch, failure: str):
     scenario, system = configs
     if failure == "setup":
         monkeypatch.setattr(StandaloneSystem, "update", Mock(side_effect=RuntimeError("setup failed")))
-    else:
+    elif failure == "spawn":
+        monkeypatch.setattr("cloudai.handlers.subprocess.Popen", Mock(side_effect=OSError("spawn failed")))
+    elif failure == "submission":
         monkeypatch.setattr(
             StandaloneRunner,
             "_submit_test",
@@ -173,8 +194,15 @@ def test_failed_experiment_is_saved(configs: tuple[str, str], monkeypatch: pytes
                 )
             ),
         )
-    with pytest.raises((RuntimeError, cloudai.core.JobSubmissionError)):
-        cloudai.api.run_experiment(scenario, system)
+    if failure == "worker-parse":
+        monkeypatch.setattr("cloudai.handlers.subprocess.Popen", Mock())
+        pending = cloudai.api.run_experiment(scenario, system, wait=False)
+        (Path(pending.path) / "scenario.toml").write_text("name = [")
+        with pytest.raises(cloudai.core.TestScenarioParsingError):
+            cloudai.handlers.run_background(Path(pending.path) / "request.json")
+    else:
+        with pytest.raises((RuntimeError, OSError, cloudai.core.JobSubmissionError)):
+            cloudai.api.run_experiment(scenario, system, wait=failure != "spawn")
     experiments = cloudai.api.list_experiments(system)
     assert len(experiments) == 1
     failed = cloudai.api.get_experiment(experiments[0][1])