Repository: oumi-ai/oumi
Pull request: Add a Slurm cluster and cloud to the oumi launcher.
# Description

<!--
Thank you for contributing to Oumi! Before sending your PR out for review, please take a quick read through this template.

When your PR is merged, its title will appear in our release notes. Make sure your title gives a clear description of your change!

After you've updated your title, please replace this section with a detailed description of your change. Include as much context as possible so your reviewers can easily understand *what* you're changing and *why*.
The more information you provide, the faster we can review your change!
-->
<!--↓↓↓↓↓↓↓↓↓↓ Describe your change below ↓↓↓↓↓↓↓↓↓↓-->
This PR adds a cluster and cloud implementation for slurm.

In oumi, we'll consider a "cluster" a logical pair of a `hostname` and a `user`, ex `taenin@myslurmhost`.

This means that running jobs as a second user on the same host would be represented with a second cluster in oumi. Ex: `janedoe@myslurmhost`

Users can set the `OUMI_SLURM_CONNECTIONS` env var to automatically connect to specific slurm hosts in the oumi CLI. This should be a comma-separated list of <user>@<host> pairs.

Ex:
```
export OUMI_SLURM_CONNECTIONS="taenin@myslurmhost,janedoe@myslurmhost"
```
<!--↑↑↑↑↑↑↑↑↑↑ Describe your change above ↑↑↑↑↑↑↑↑↑↑-->

## Related issues

<!--
Make sure to list any relevant related issues to your change. More often than not this will be the single issue fixed by your PR.
-->
<!--↓↓↓↓↓↓↓↓↓↓ List your related issues below ↓↓↓↓↓↓↓↓↓↓-->

Fixes #1382

<!--↑↑↑↑↑↑↑↑↑↑ List your related issues above ↑↑↑↑↑↑↑↑↑↑-->

## Before submitting

- [ ] This PR only changes documentation. (You can ignore the following checks in that case)
- [x] Did you read the [contributor guideline](https://github.com/oumi-ai/oumi/blob/main/CONTRIBUTING.md) Pull Request guidelines?
- [x] Did you link the issue(s) related to this PR in the section above?
- [x] Did you add / update tests where needed?

## Reviewers

At least one review from a member of `oumi-ai/oumi-staff` is required.

<!-- Add `oumi-ai/oumi-staff` as a reviewer when your PR is ready for review.

You are also welcome to add individual members of `oumi-ai/oumi-staff` as reviewers.

If no one has reviewed your PR after several days, feel free to add a comment tagging specific reviewers.

 -->


Problem statement:
[Feature] Add Slurm support in Oumi Launch
### Feature request

We should add support for sshing into a slurm head node and kicking off jobs. This is great for local and dedicated clusters.

### Motivation / references

This greatly simplifies workflows with Oumi when working on local clusters that are already running slurm.

### Your contribution

I will contribute this feature.

Hints from issue discussion:

Commit message: Merge branch 'main' into taenin/slurm_cluster

Patch to review:
diff --git a/src/oumi/cli/env.py b/src/oumi/cli/env.py
index 32dbd2524..c4e93e766 100644
--- a/src/oumi/cli/env.py
+++ b/src/oumi/cli/env.py
@@ -80,6 +80,7 @@ def env():
             "LOCAL_RANK",
             "LOCAL_WORLD_SIZE",
             "OUMI_EXTRA_DEPS_FILE",
+            "OUMI_SLURM_CONNECTIONS",
             "OUMI_USE_SPOT_VM",
             "RANK",
             "WORLD_SIZE",
diff --git a/src/oumi/launcher/clients/slurm_client.py b/src/oumi/launcher/clients/slurm_client.py
index 4ee2fe3f6..a13e45965 100644
--- a/src/oumi/launcher/clients/slurm_client.py
+++ b/src/oumi/launcher/clients/slurm_client.py
@@ -320,7 +320,11 @@ def list_jobs(self) -> list[JobStatus]:
             A list of JobStatus.
         """
         response_format = "JobId%-30,JobName%30,User%30,State%30,Reason%30"
-        command = f"sacct --user={self._user} --format='{response_format}'"
+        # Forcibly list all jobs since Jan 1, 2025.
+        command = (
+            f"sacct --user={self._user} --format='{response_format}' -X "
+            "--starttime 2025-01-01"
+        )
         result = self.run_commands([command])
         if result.exit_code != 0:
             raise RuntimeError(f"Failed to list jobs. stderr: {result.stderr}")
diff --git a/src/oumi/launcher/clouds/__init__.py b/src/oumi/launcher/clouds/__init__.py
index ac4f3096d..cb4efc0ae 100644
--- a/src/oumi/launcher/clouds/__init__.py
+++ b/src/oumi/launcher/clouds/__init__.py
@@ -35,6 +35,7 @@
 from oumi.launcher.clouds.local_cloud import LocalCloud
 from oumi.launcher.clouds.polaris_cloud import PolarisCloud
 from oumi.launcher.clouds.sky_cloud import SkyCloud
+from oumi.launcher.clouds.slurm_cloud import SlurmCloud
 from oumi.utils import logging
 
 logging.configure_dependency_warnings()
@@ -44,4 +45,5 @@
     "LocalCloud",
     "PolarisCloud",
     "SkyCloud",
+    "SlurmCloud",
 ]
diff --git a/src/oumi/launcher/clouds/slurm_cloud.py b/src/oumi/launcher/clouds/slurm_cloud.py
new file mode 100644
index 000000000..e4a17fa93
--- /dev/null
+++ b/src/oumi/launcher/clouds/slurm_cloud.py
@@ -0,0 +1,158 @@
+# Copyright 2025 - Oumi
+#
+# 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 os
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+from oumi.core.configs import JobConfig
+from oumi.core.launcher import BaseCloud, BaseCluster, JobStatus
+from oumi.core.registry import register_cloud_builder
+from oumi.launcher.clients.slurm_client import SlurmClient
+from oumi.launcher.clusters.slurm_cluster import SlurmCluster
+from oumi.utils.logging import logger
+
+_OUMI_SLURM_CONNECTIONS = "OUMI_SLURM_CONNECTIONS"
+
+
+@dataclass
+class _ConnectionInfo:
+    """Dataclass to hold information about a connection."""
+
+    hostname: str
+    user: str
+
+    def name(self):
+        return f"{self.user}@{self.hostname}"
+
+
+def _parse_cluster_name(name: str) -> _ConnectionInfo:
+    """Parses the cluster name into queue and user components.
+
+    Args:
+        name: The name of the cluster.
+
+    Returns:
+        _ConnectionInfo: The parsed cluster information.
+    """
+    # Expected format: <user>@<hostname>
+    connection_regex = r"^([a-zA-Z0-9\.\-\_]+)\@([a-zA-Z0-9\.\-\_]+)"
+    match = re.match(connection_regex, name)
+    if not match:
+        raise ValueError(
+            f"Invalid cluster name: {name}. Must be in the format 'user@hostname'."
+        )
+    return _ConnectionInfo(hostname=match.group(2), user=match.group(1))
+
+
+def _get_slurm_connections() -> list[_ConnectionInfo]:
+    """Gets Slurm connections from the OUMI_SLURM_CONNECTIONS environment variable."""
+    connections_str = os.getenv(_OUMI_SLURM_CONNECTIONS, "")
+    if not connections_str:
+        return []
+    valid_connections = []
+
+    for connection in [h.strip() for h in connections_str.split(",")]:
+        try:
+            valid_connections.append(_parse_cluster_name(connection))
+        except ValueError:
+            logger.warning(f"Invalid Slurm connection string: {connection}. Skipping.")
+    return valid_connections
+
+
+class SlurmCloud(BaseCloud):
+    """A resource pool for managing the Slurm ALCF job queues."""
+
+    def __init__(self):
+        """Initializes a new instance of the SlurmCloud class."""
+        # A mapping from cluster names to Slurm Cluster instances.
+        self._clusters = {}
+
+        # Initialize default connections.
+        self.initialize_clusters()
+
+    def _get_or_create_cluster(self, name: str) -> SlurmCluster:
+        """Gets the cluster with the specified name, or creates one if it doesn't exist.
+
+        Args:
+            name: The name of the cluster.
+
+        Returns:
+            SlurmCluster: The cluster instance.
+        """
+        if name not in self._clusters:
+            cluster_info = _parse_cluster_name(name)
+            self._clusters[name] = SlurmCluster(
+                name,
+                SlurmClient(
+                    user=cluster_info.user,
+                    slurm_host=cluster_info.hostname,
+                    cluster_name=cluster_info.name(),
+                ),
+            )
+        return self._clusters[name]
+
+    def initialize_clusters(self) -> list[BaseCluster]:
+        """Initializes clusters for the specified user for all Slurm queues.
+
+        Returns:
+            List[SlurmCluster]: The list of initialized clusters.
+        """
+        connections = _get_slurm_connections()
+        clusters = []
+        for c in connections:
+            cluster = self._get_or_create_cluster(c.name())
+            clusters.append(cluster)
+        return clusters
+
+    def up_cluster(self, job: JobConfig, name: Optional[str], **kwargs) -> JobStatus:
+        """Creates a cluster and starts the provided Job."""
+        if not job.user:
+            raise ValueError("User must be provided in the job config.")
+        if name:
+            cluster_info = _parse_cluster_name(name)
+            if cluster_info.user != job.user:
+                raise ValueError(
+                    f"Invalid cluster name: `{name}`. "
+                    f"User must match the provided job user: `{job.user}`."
+                )
+        else:
+            raise ValueError(
+                "A cluster name must be provided for Slurm. "
+                "Cluster names are of the form 'user@hostname'."
+            )
+        cluster = self._get_or_create_cluster(cluster_info.name())
+        job_status = cluster.run_job(job)
+        if not job_status:
+            raise RuntimeError("Failed to start job.")
+        return job_status
+
+    def get_cluster(self, name) -> Optional[BaseCluster]:
+        """Gets the cluster with the specified name, or None if not found."""
+        clusters = self.list_clusters()
+        for cluster in clusters:
+            if cluster.name() == name:
+                return cluster
+        return None
+
+    def list_clusters(self) -> list[BaseCluster]:
+        """Lists the active clusters on this cloud."""
+        return list(self._clusters.values())
+
+
+@register_cloud_builder("slurm")
+def slurm_cloud_builder() -> SlurmCloud:
+    """Builds a SlurmCloud instance."""
+    return SlurmCloud()
diff --git a/src/oumi/launcher/clusters/slurm_cluster.py b/src/oumi/launcher/clusters/slurm_cluster.py
new file mode 100644
index 000000000..2e61543fe
--- /dev/null
+++ b/src/oumi/launcher/clusters/slurm_cluster.py
@@ -0,0 +1,236 @@
+# Copyright 2025 - Oumi
+#
+# 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 time
+import uuid
+from datetime import datetime
+from functools import reduce
+from pathlib import Path
+from typing import Any, Optional
+
+from oumi.core.configs import JobConfig
+from oumi.core.launcher import BaseCluster, JobStatus
+from oumi.launcher.clients.slurm_client import SlurmClient
+from oumi.utils.logging import logger
+
+
+def _format_date(date: datetime) -> str:
+    """Formats the provided date as a string.
+
+    Args:
+        date: The date to format.
+
+    Returns:
+        The formatted date.
+    """
+    return date.strftime("%Y%m%d_%H%M%S%f")
+
+
+def _last_sbatch_line(script: list[str]) -> int:
+    """Finds the last SBATCH instruction line in the script.
+
+    Args:
+        script: The lines of the script.
+
+    Returns:
+        The index of the last SBATCH instruction line. -1 if not found.
+    """
+    return reduce(
+        lambda acc, val: val[0] if val[1].startswith("#SBATCH") else acc,
+        enumerate(script),
+        -1,
+    )
+
+
+def _create_job_script(job: JobConfig) -> str:
+    """Creates a job script for the specified job.
+
+    Args:
+        job: The job to create a script for.
+
+    Returns:
+        The script as a string.
+    """
+    setup_lines = [] if not job.setup else job.setup.strip().split("\n")
+    run_lines = job.run.strip().split("\n")
+    # Find the last SBATCH instruction line.
+    last_run_sbatch = _last_sbatch_line(run_lines) + 1
+    last_setup_sbatch = _last_sbatch_line(setup_lines) + 1
+    # Inject environment variables into the script after SBATCH instructions.
+    env_lines = [f"export {key}={value}" for key, value in job.envs.items()]
+    # Pad the environment variables with newlines.
+    env_lines = [""] + env_lines + [""] if env_lines else []
+    # Generate the job script.
+    # The script should have the following structure:
+    # 1. SBATCH instructions from Setup and Run commands (in that order).
+    # 2. Environment variables.
+    # 3. Setup commands.
+    # 4. Run commands.
+    output_lines = (
+        setup_lines[:last_setup_sbatch]
+        + run_lines[:last_run_sbatch]
+        + env_lines
+        + setup_lines[last_setup_sbatch:]
+        + run_lines[last_run_sbatch:]
+    )
+    # Always start the script with #!/bin/bash.
+    script_prefix = "#!/bin/bash"
+    if len(output_lines) > 0:
+        if not output_lines[0].startswith("script_prefix"):
+            output_lines.insert(0, script_prefix)
+    # Join each line. Always end the script with a new line.
+    return "\n".join(output_lines) + "\n"
+
+
+def _validate_job_config(job: JobConfig) -> None:
+    """Validates the provided job configuration.
+
+    Args:
+        job: The job to validate.
+    """
+    if not job.user:
+        raise ValueError("User must be provided for Slurm jobs.")
+    if not job.working_dir:
+        raise ValueError("Working directory must be provided for Slurm jobs.")
+    if not job.run:
+        raise ValueError("Run script must be provided for Slurm jobs.")
+    if job.num_nodes < 1:
+        raise ValueError("Number of nodes must be at least 1.")
+    if job.resources.cloud != "slurm":
+        raise ValueError(
+            f"`Resources.cloud` must be `slurm`. "
+            f"Unsupported cloud: {job.resources.cloud}"
+        )
+    # Warn that other resource parameters are unused for Slurm.
+    if job.resources.region:
+        logger.warning("Region is unused for Slurm jobs.")
+    if job.resources.zone:
+        logger.warning("Zone is unused for Slurm jobs.")
+    if job.resources.accelerators:
+        logger.warning("Accelerators are unused for Slurm jobs.")
+    if job.resources.cpus:
+        logger.warning("CPUs are unused for Slurm jobs.")
+    if job.resources.memory:
+        logger.warning("Memory is unused for Slurm jobs.")
+    if job.resources.instance_type:
+        logger.warning("Instance type is unused for Slurm jobs.")
+    if job.resources.disk_size:
+        logger.warning("Disk size is unused for Slurm jobs.")
+    if job.resources.instance_type:
+        logger.warning("Instance type is unused for Slurm jobs.")
+    # Warn that storage mounts are currently unsupported.
+    if len(job.storage_mounts.items()) > 0:
+        logger.warning("Storage mounts are currently unsupported for Slurm jobs.")
+
+
+class SlurmCluster(BaseCluster):
+    """A cluster implementation backed by a Slurm scheduler."""
+
+    def __init__(self, name: str, client: SlurmClient) -> None:
+        """Initializes a new instance of the SlurmCluster class."""
+        self._name = name
+        self._client = client
+
+    def __eq__(self, other: Any) -> bool:
+        """Checks if two SlurmClusters are equal."""
+        if not isinstance(other, SlurmCluster):
+            return False
+        return self.name() == other.name()
+
+    def name(self) -> str:
+        """Gets the name of the cluster."""
+        return self._name
+
+    def get_job(self, job_id: str) -> Optional[JobStatus]:
+        """Gets the jobs on this cluster if it exists, else returns None."""
+        for job in self.get_jobs():
+            if job.id == job_id:
+                return job
+        return None
+
+    def get_jobs(self) -> list[JobStatus]:
+        """Lists the jobs on this cluster."""
+        jobs = self._client.list_jobs()
+        for job in jobs:
+            job.cluster = self._name
+        return jobs
+
+    def cancel_job(self, job_id: str) -> JobStatus:
+        """Cancels the specified job on this cluster."""
+        self._client.cancel(job_id)
+        job = self.get_job(job_id)
+        if job is None:
+            raise RuntimeError(f"Job {job_id} not found.")
+        return job
+
+    def run_job(self, job: JobConfig) -> JobStatus:
+        """Runs the specified job on this cluster.
+
+        For Slurm this method consists of 5 parts:
+
+        1. Copy the working directory to ~/oumi_launcher/$JOB_NAME.
+        2. Check if there is a conda installation at /home/$USER/miniconda3/envs/oumi.
+           If not, install it.
+        3. Copy all file mounts.
+        4. Create a job script with all env vars, setup, and run commands.
+        5. CD into the working directory and submit the job.
+
+        Args:
+            job: The job to run.
+
+        Returns:
+            JobStatus: The job status.
+        """
+        _validate_job_config(job)
+        job_name = job.name or uuid.uuid1().hex
+        submission_time = _format_date(datetime.now())
+        remote_working_dir = Path(f"~/oumi_launcher/{submission_time}")
+        # Copy the working directory to ~/oumi_launcher/...
+        self._client.put_recursive(job.working_dir, str(remote_working_dir))
+        # Copy all file mounts.
+        for remote_path, local_path in job.file_mounts.items():
+            self._client.put_recursive(local_path, remote_path)
+        # Create the job script by merging envs, setup, and run commands.
+        job_script = _create_job_script(job)
+        script_path = remote_working_dir / "oumi_job.sh"
+        self._client.put(job_script, str(script_path))
+        # Set the proper CHMOD permissions.
+        self._client.run_commands([f"chmod +x {script_path}"])
+        # Submit the job.
+        job_id = self._client.submit_job(
+            str(script_path),
+            str(remote_working_dir),
+            job.num_nodes,
+            job_name,
+        )
+        max_retries = 3
+        wait_time = 5
+        for _ in range(max_retries):
+            job_status = self.get_job(job_id)
+            if job_status is not None:
+                return job_status
+            logger.info(f"Job {job_id} not found. Retrying in {wait_time} seconds.")
+            time.sleep(wait_time)
+        job_status = self.get_job(job_id)
+        if job_status is None:
+            raise RuntimeError(f"Job {job_id} not found after submission.")
+        return job_status
+
+    def stop(self) -> None:
+        """This is a no-op for Slurm clusters."""
+        pass
+
+    def down(self) -> None:
+        """This is a no-op for Slurm clusters."""
+        pass
diff --git a/tests/unit/launcher/clients/test_slurm_client.py b/tests/unit/launcher/clients/test_slurm_client.py
index 14d3071f3..cec4c8f97 100644
--- a/tests/unit/launcher/clients/test_slurm_client.py
+++ b/tests/unit/launcher/clients/test_slurm_client.py
@@ -10,7 +10,8 @@
 
 _CTRL_PATH: str = "-S ~/.ssh/control-%h-%p-%r"
 _SACCT_CMD = (
-    "sacct --user=user --format='JobId%-30,JobName%30,User%30,State%30,Reason%30'"
+    "sacct --user=user --format='JobId%-30,JobName%30,User%30,State%30,Reason%30' "
+    "-X --starttime 2025-01-01"
 )
 
 
diff --git a/tests/unit/launcher/clouds/test_slurm_cloud.py b/tests/unit/launcher/clouds/test_slurm_cloud.py
new file mode 100644
index 000000000..a85fbe310
--- /dev/null
+++ b/tests/unit/launcher/clouds/test_slurm_cloud.py
@@ -0,0 +1,237 @@
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+from oumi.core.configs import JobConfig, JobResources, StorageMount
+from oumi.core.launcher import JobStatus
+from oumi.core.registry import REGISTRY, RegistryType
+from oumi.launcher.clients.slurm_client import SlurmClient
+from oumi.launcher.clouds.slurm_cloud import SlurmCloud
+from oumi.launcher.clusters.slurm_cluster import SlurmCluster
+
+
+#
+# Fixtures
+#
+@pytest.fixture
+def mock_slurm_client():
+    with patch("oumi.launcher.clouds.slurm_cloud.SlurmClient") as client:
+        yield client
+
+
+@pytest.fixture
+def mock_slurm_cluster():
+    with patch("oumi.launcher.clouds.slurm_cloud.SlurmCluster") as cluster:
+        yield cluster
+
+
+@pytest.fixture
+def mock_os():
+    with patch("oumi.launcher.clouds.slurm_cloud.os") as os_mock:
+        os_mock.getenv.return_value = ""
+        yield os_mock
+
+
+def _get_default_job(cloud: str) -> JobConfig:
+    resources = JobResources(
+        cloud=cloud,
+        region="us-central1",
+        zone=None,
+        accelerators="A100-80GB",
+        cpus="4",
+        memory="64",
+        instance_type=None,
+        use_spot=True,
+        disk_size=512,
+        disk_tier="low",
+    )
+    return JobConfig(
+        name="myjob",
+        user="user",
+        working_dir="./",
+        num_nodes=2,
+        resources=resources,
+        envs={"var1": "val1"},
+        file_mounts={},
+        storage_mounts={
+            "~/home/remote/path/gcs/": StorageMount(
+                source="gs://mybucket/", store="gcs"
+            )
+        },
+        setup="pip install -r requirements.txt",
+        run="./hello_world.sh",
+    )
+
+
+#
+# Tests
+#
+def test_slurm_cloud_up_cluster(mock_slurm_client, mock_os, mock_slurm_cluster):
+    cloud = SlurmCloud()
+    mock_client = Mock(spec=SlurmClient)
+    mock_slurm_client.side_effect = [mock_client]
+    mock_cluster = Mock(spec=SlurmCluster)
+    mock_slurm_cluster.side_effect = [mock_cluster]
+    expected_job_status = JobStatus(
+        id="job_id",
+        cluster="user@somehost",
+        name="foo",
+        status="running",
+        metadata="bar",
+        done=False,
+    )
+    mock_cluster.run_job.return_value = expected_job_status
+    job = _get_default_job("slurm")
+    job_status = cloud.up_cluster(job, "user@somehost")
+    mock_slurm_client.assert_called_once_with(
+        user="user", slurm_host="somehost", cluster_name="user@somehost"
+    )
+    mock_cluster.run_job.assert_called_once_with(job)
+    assert job_status == expected_job_status
+
+
+def test_slurm_cloud_up_cluster_fails_mismatched_user(
+    mock_slurm_client, mock_os, mock_slurm_cluster
+):
+    cloud = SlurmCloud()
+    with pytest.raises(
+        ValueError,
+        match=(
+            "Invalid cluster name: `user1@somehost`. "
+            "User must match the provided job user: `user`."
+        ),
+    ):
+        _ = cloud.up_cluster(_get_default_job("slurm"), "user1@somehost")
+
+
+def test_slurm_cloud_init_with_connections(mock_os, mock_slurm_client):
+    mock_client = Mock(spec=SlurmClient)
+    mock_os.getenv.return_value = "user1@host1,user2@host2"
+    mock_slurm_client.side_effect = [mock_client, mock_client]
+    cloud = SlurmCloud()
+    cluster_names = [cluster.name() for cluster in cloud.list_clusters()]
+    cluster_names.sort()
+    assert cluster_names == [
+        "user1@host1",
+        "user2@host2",
+    ]
+    mock_slurm_client.assert_has_calls(
+        [
+            call(user="user1", slurm_host="host1", cluster_name="user1@host1"),
+            call(user="user2", slurm_host="host2", cluster_name="user2@host2"),
+        ]
+    )
+
+
+def test_slurm_cloud_init_skips_malformed_connections(mock_os, mock_slurm_client):
+    mock_client = Mock(spec=SlurmClient)
+    mock_os.getenv.return_value = "user1@host1,foob@@@ar, user2@host2 , user3@host3"
+    mock_slurm_client.side_effect = [mock_client, mock_client, mock_client]
+    cloud = SlurmCloud()
+    cluster_names = [cluster.name() for cluster in cloud.list_clusters()]
+    cluster_names.sort()
+    assert cluster_names == [
+        "user1@host1",
+        "user2@host2",
+        "user3@host3",
+    ]
+    mock_slurm_client.assert_has_calls(
+        [
+            call(user="user1", slurm_host="host1", cluster_name="user1@host1"),
+            call(user="user2", slurm_host="host2", cluster_name="user2@host2"),
+            call(user="user3", slurm_host="host3", cluster_name="user3@host3"),
+        ]
+    )
+
+
+def test_slurm_cloud_initialize_cluster(mock_os, mock_slurm_client):
+    cloud = SlurmCloud()
+    mock_client = Mock(spec=SlurmClient)
+    mock_os.getenv.return_value = "user1@host1,user2@host2,user3@host3"
+    mock_slurm_client.side_effect = [mock_client, mock_client, mock_client]
+    clusters = cloud.initialize_clusters()
+    clusters2 = cloud.initialize_clusters()
+    mock_slurm_client.assert_has_calls(
+        [
+            call(user="user1", slurm_host="host1", cluster_name="user1@host1"),
+            call(user="user2", slurm_host="host2", cluster_name="user2@host2"),
+            call(user="user3", slurm_host="host3", cluster_name="user3@host3"),
+        ]
+    )
+    cluster_names = [cluster.name() for cluster in clusters]
+    cluster_names.sort()
+    assert cluster_names == [
+        "user1@host1",
+        "user2@host2",
+        "user3@host3",
+    ]
+    # Verify that the second initialization returns the same clusters.
+    assert clusters == clusters2
+
+
+def test_slurm_cloud_list_clusters(mock_os, mock_slurm_client):
+    cloud = SlurmCloud()
+    mock_client = Mock(spec=SlurmClient)
+    mock_os.getenv.return_value = "user1@host1,user2@host2,user3@host3"
+    mock_slurm_client.side_effect = [mock_client, mock_client, mock_client]
+    assert [] == cloud.list_clusters()
+    clusters = cloud.initialize_clusters()
+    mock_slurm_client.assert_has_calls(
+        [
+            call(user="user1", slurm_host="host1", cluster_name="user1@host1"),
+            call(user="user2", slurm_host="host2", cluster_name="user2@host2"),
+            call(user="user3", slurm_host="host3", cluster_name="user3@host3"),
+        ]
+    )
+    clusters = cloud.list_clusters()
+    expected_clusters = [
+        "user1@host1",
+        "user2@host2",
+        "user3@host3",
+    ]
+    cluster_names = [cluster.name() for cluster in clusters]
+    cluster_names.sort()
+    assert cluster_names == expected_clusters
+
+
+def test_slurm_cloud_get_cluster_empty(mock_os, mock_slurm_client):
+    cloud = SlurmCloud()
+    # Check that there are no initial clusters.
+    assert cloud.get_cluster("debug.user") is None
+
+
+def test_slurm_cloud_get_cluster_success(mock_os, mock_slurm_client):
+    cloud = SlurmCloud()
+    mock_client = Mock(spec=SlurmClient)
+    mock_os.getenv.return_value = "user1@host1,user2@host2,user3@host3"
+    mock_slurm_client.side_effect = [mock_client, mock_client, mock_client]
+    assert [] == cloud.list_clusters()
+    _ = cloud.initialize_clusters()
+    mock_slurm_client.assert_has_calls(
+        [
+            call(user="user1", slurm_host="host1", cluster_name="user1@host1"),
+            call(user="user2", slurm_host="host2", cluster_name="user2@host2"),
+            call(user="user3", slurm_host="host3", cluster_name="user3@host3"),
+        ]
+    )
+    expected_clusters = [
+        "user1@host1",
+        "user2@host2",
+        "user3@host3",
+    ]
+    for name in expected_clusters:
+        cluster = cloud.get_cluster(name)
+        assert cluster is not None
+        assert cluster.name() == name
+
+
+def test_slurm_cloud_get_cluster_fails(mock_os, mock_slurm_client):
+    cloud = SlurmCloud()
+    mock_client = Mock(spec=SlurmClient)
+    mock_slurm_client.side_effect = [mock_client, mock_client]
+    cloud.initialize_clusters()
+    assert cloud.get_cluster("nonexistent") is None
+
+
+def test_slurm_cloud_builder_registered():
+    assert REGISTRY.contains("slurm", RegistryType.CLOUD)
diff --git a/tests/unit/launcher/clusters/test_slurm_cluster.py b/tests/unit/launcher/clusters/test_slurm_cluster.py
new file mode 100644
index 000000000..c125365a1
--- /dev/null
+++ b/tests/unit/launcher/clusters/test_slurm_cluster.py
@@ -0,0 +1,714 @@
+from datetime import datetime
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+from oumi.core.configs import JobConfig, JobResources, StorageMount
+from oumi.core.launcher import JobStatus
+from oumi.launcher.clients.slurm_client import SlurmClient
+from oumi.launcher.clusters.slurm_cluster import SlurmCluster
+
+
+#
+# Fixtures
+#
+@pytest.fixture
+def mock_slurm_client():
+    yield Mock(spec=SlurmClient)
+
+
+@pytest.fixture
+def mock_time():
+    with patch("oumi.launcher.clusters.slurm_cluster.time") as mock_t:
+        yield mock_t
+
+
+@pytest.fixture
+def mock_datetime():
+    with patch("oumi.launcher.clusters.slurm_cluster.datetime") as mock_dt:
+        mock_dt.now.return_value = datetime(2024, 10, 9, 13, 4, 24, 513094)
+        yield mock_dt
+
+
+def _get_default_job(cloud: str) -> JobConfig:
+    resources = JobResources(
+        cloud=cloud,
+        region="us-central1",
+        zone=None,
+        accelerators="A100-80GB",
+        cpus="4",
+        memory="64",
+        instance_type=None,
+        use_spot=True,
+        disk_size=512,
+        disk_tier="low",
+    )
+    return JobConfig(
+        name="myjob",
+        user="user",
+        working_dir="./",
+        num_nodes=2,
+        resources=resources,
+        envs={"var1": "val1"},
+        file_mounts={
+            "~/home/remote/path.bar": "~/local/path.bar",
+            "~/home/remote/path2.txt": "~/local/path2.txt",
+        },
+        storage_mounts={
+            "~/home/remote/path/gcs/": StorageMount(
+                source="gs://mybucket/", store="gcs"
+            )
+        },
+        setup=(
+            "#SBATCH --gpus-per-task=8 \n#SBATCH --cpus-per-task=4\n"
+            "pip install -r requirements.txt"
+        ),
+        run="./hello_world.sh",
+    )
+
+
+#
+# Tests
+#
+def test_slurm_cluster_name(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("demand.einstein", mock_slurm_client)
+    assert cluster.name() == "demand.einstein"
+
+    cluster = SlurmCluster("debug.einstein", mock_slurm_client)
+    assert cluster.name() == "debug.einstein"
+
+    cluster = SlurmCluster("debug-scaling.einstein", mock_slurm_client)
+    assert cluster.name() == "debug-scaling.einstein"
+
+    cluster = SlurmCluster("preemptable.einstein", mock_slurm_client)
+    assert cluster.name() == "preemptable.einstein"
+
+    cluster = SlurmCluster("prod.einstein", mock_slurm_client)
+    assert cluster.name() == "prod.einstein"
+
+
+def test_slurm_cluster_get_job_valid_id(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="myjob",
+            name="some name",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="final job",
+            name="name3",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+    ]
+    job = cluster.get_job("myjob")
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job is not None
+    assert job.id == "myjob"
+    assert job.cluster == "debug@host"
+
+
+def test_slurm_cluster_get_job_invalid_id_empty(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = []
+    job = cluster.get_job("myjob")
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job is None
+
+
+def test_slurm_cluster_get_job_invalid_id_nonempty(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="myjob",
+            name="some name",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="final job",
+            name="name3",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+    ]
+    job = cluster.get_job("wrong job")
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job is None
+
+
+def test_slurm_cluster_get_jobs_nonempty(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="myjob",
+            name="some name",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+        JobStatus(
+            id="final job",
+            name="name3",
+            status="running",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        ),
+    ]
+    jobs = cluster.get_jobs()
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    expected_jobs = [
+        JobStatus(
+            id="myjob",
+            name="some name",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+        JobStatus(
+            id="final job",
+            name="name3",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+    ]
+    assert jobs == expected_jobs
+
+
+def test_slurm_cluster_get_jobs_empty(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = []
+    jobs = cluster.get_jobs()
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    expected_jobs = []
+    assert jobs == expected_jobs
+
+
+def test_slurm_cluster_cancel_job(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("prod@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="myjob",
+            name="some name",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+        JobStatus(
+            id="final job",
+            name="name3",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+    ]
+    job_status = cluster.cancel_job("job2")
+    expected_status = JobStatus(
+        id="job2",
+        name="some",
+        status="running",
+        metadata="",
+        cluster="prod@host",
+        done=False,
+    )
+    mock_slurm_client.cancel.assert_called_once_with(
+        "job2",
+    )
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_cancel_job_fails(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("prod@host", mock_slurm_client)
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="job2",
+            name="some",
+            status="running",
+            metadata="",
+            cluster="debug@host",
+            done=False,
+        ),
+    ]
+    with pytest.raises(RuntimeError):
+        _ = cluster.cancel_job("myjobid")
+
+
+def test_slurm_cluster_run_job(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_slurm_client.run_commands.return_value = mock_successful_cmd
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job_status = cluster.run_job(_get_default_job("slurm"))
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+            call(
+                "~/local/path.bar",
+                "~/home/remote/path.bar",
+            ),
+            call(
+                "~/local/path2.txt",
+                "~/home/remote/path2.txt",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = (
+        "#!/bin/bash\n#SBATCH --gpus-per-task=8 \n#SBATCH --cpus-per-task=4\n\n"
+        "export var1=val1\n\n"
+        "pip install -r requirements.txt\n./hello_world.sh\n"
+    )
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "myjob",
+    )
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_with_polling_succeeds(
+    mock_time, mock_datetime, mock_slurm_client
+):
+    mock_time.sleep.side_effect = [None, None, None, None, None]
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_failed_cmd = Mock()
+    mock_failed_cmd.exit_code = 1
+    mock_slurm_client.run_commands.side_effect = [
+        mock_failed_cmd,
+        mock_successful_cmd,
+        mock_successful_cmd,
+        mock_successful_cmd,
+    ]
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.side_effect = [
+        [],
+        [
+            JobStatus(
+                id="1",
+                name="some name",
+                status="RUNNING",
+                metadata="",
+                cluster="mycluster",
+                done=False,
+            )
+        ],
+        [
+            JobStatus(
+                id="1234",
+                name="some name",
+                status="RUNNING",
+                metadata="",
+                cluster="mycluster",
+                done=False,
+            )
+        ],
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job_status = cluster.run_job(_get_default_job("slurm"))
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+            call(
+                "~/local/path.bar",
+                "~/home/remote/path.bar",
+            ),
+            call(
+                "~/local/path2.txt",
+                "~/home/remote/path2.txt",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = (
+        "#!/bin/bash\n#SBATCH --gpus-per-task=8 \n#SBATCH --cpus-per-task=4\n\n"
+        "export var1=val1\n\n"
+        "pip install -r requirements.txt\n./hello_world.sh\n"
+    )
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "myjob",
+    )
+    mock_slurm_client.list_jobs.assert_has_calls([call(), call(), call()])
+    mock_time.sleep.assert_has_calls([call(5), call(5)])
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_no_name(mock_datetime, mock_slurm_client):
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_slurm_client.run_commands.return_value = mock_successful_cmd
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job = _get_default_job("slurm")
+    job.name = None
+    with patch("oumi.launcher.clusters.slurm_cluster.uuid") as mock_uuid:
+        mock_hex = Mock()
+        mock_hex.hex = "1-2-3"
+        mock_uuid.uuid1.return_value = mock_hex
+        job_status = cluster.run_job(job)
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+            call(
+                "~/local/path.bar",
+                "~/home/remote/path.bar",
+            ),
+            call(
+                "~/local/path2.txt",
+                "~/home/remote/path2.txt",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = (
+        "#!/bin/bash\n#SBATCH --gpus-per-task=8 \n#SBATCH --cpus-per-task=4\n\n"
+        "export var1=val1\n\n"
+        "pip install -r requirements.txt\n./hello_world.sh\n"
+    )
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "1-2-3",
+    )
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_no_mounts(mock_datetime, mock_slurm_client):
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_slurm_client.run_commands.return_value = mock_successful_cmd
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job = _get_default_job("slurm")
+    job.file_mounts = {}
+    job_status = cluster.run_job(job)
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = (
+        "#!/bin/bash\n#SBATCH --gpus-per-task=8 \n#SBATCH --cpus-per-task=4\n\n"
+        "export var1=val1\n\n"
+        "pip install -r requirements.txt\n./hello_world.sh\n"
+    )
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "myjob",
+    )
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_no_pbs(mock_datetime, mock_slurm_client):
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_slurm_client.run_commands.return_value = mock_successful_cmd
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job = _get_default_job("slurm")
+    job.file_mounts = {}
+    job.setup = "small setup"
+    job.run = "./hello_world.sh"
+    job_status = cluster.run_job(job)
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = (
+        "#!/bin/bash\n\n" "export var1=val1\n\n" "small setup\n./hello_world.sh\n"
+    )
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "myjob",
+    )
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_no_setup(mock_datetime, mock_slurm_client):
+    mock_successful_cmd = Mock()
+    mock_successful_cmd.exit_code = 0
+    mock_slurm_client.run_commands.return_value = mock_successful_cmd
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "1234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    expected_status = JobStatus(
+        id="1234",
+        name="some name",
+        status="RUNNING",
+        metadata="",
+        cluster="debug@host",
+        done=False,
+    )
+    job = _get_default_job("slurm")
+    job.file_mounts = {}
+    job.setup = None
+    job.run = "./hello_world.sh"
+    job_status = cluster.run_job(job)
+    mock_slurm_client.put_recursive.assert_has_calls(
+        [
+            call(
+                "./",
+                "~/oumi_launcher/20241009_130424513094",
+            ),
+        ],
+    )
+    mock_slurm_client.run_commands.assert_has_calls(
+        [
+            call(["chmod +x ~/oumi_launcher/20241009_130424513094/oumi_job.sh"]),
+        ]
+    )
+    job_script = "#!/bin/bash\n\n" "export var1=val1\n\n" "./hello_world.sh\n"
+    mock_slurm_client.put.assert_called_once_with(
+        job_script, "~/oumi_launcher/20241009_130424513094/oumi_job.sh"
+    )
+    mock_slurm_client.submit_job.assert_called_once_with(
+        "~/oumi_launcher/20241009_130424513094/oumi_job.sh",
+        "~/oumi_launcher/20241009_130424513094",
+        2,
+        "myjob",
+    )
+    mock_slurm_client.list_jobs.assert_called_once_with()
+    assert job_status == expected_status
+
+
+def test_slurm_cluster_run_job_fails(mock_time, mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug@host", mock_slurm_client)
+    mock_slurm_client.submit_job.return_value = "234"
+    mock_slurm_client.list_jobs.return_value = [
+        JobStatus(
+            id="1234",
+            name="some name",
+            status="RUNNING",
+            metadata="",
+            cluster="mycluster",
+            done=False,
+        )
+    ]
+    with pytest.raises(RuntimeError):
+        _ = cluster.run_job(_get_default_job("slurm"))
+    mock_time.sleep.assert_has_calls([call(5), call(5), call(5)])
+
+
+def test_slurm_cluster_down(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug-scaling@host", mock_slurm_client)
+    cluster.down()
+    # Nothing to assert, this method is a no-op.
+
+
+def test_slurm_cluster_stop(mock_datetime, mock_slurm_client):
+    cluster = SlurmCluster("debug-scaling@host", mock_slurm_client)
+    cluster.stop()
+    # Nothing to assert, this method is a no-op.


Code review comments:
diff_hunk: @@ -0,0 +1,158 @@
+# Copyright 2025 - Oumi
+#
+# 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 os
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+from oumi.core.configs import JobConfig
+from oumi.core.launcher import BaseCloud, BaseCluster, JobStatus
+from oumi.core.registry import register_cloud_builder
+from oumi.launcher.clients.slurm_client import SlurmClient
+from oumi.launcher.clusters.slurm_cluster import SlurmCluster
+from oumi.utils.logging import logger
+
+_OUMI_SLURM_CONNECTIONS = "OUMI_SLURM_CONNECTIONS"
+
+
+@dataclass
+class _ConnectionInfo:
+    """Dataclass to hold information about a connection."""
+
+    hostname: str
+    user: str
+
+    def name(self):
+        return f"{self.user}@{self.hostname}"
+
+
+def _parse_cluster_name(name: str) -> _ConnectionInfo:
+    """Parses the cluster name into queue and user components.
+
+    Args:
+        name: The name of the cluster.
+
+    Returns:
+        _ConnectionInfo: The parsed cluster information.
+    """
+    # Expected format: <user>@<hostname>
+    connection_regex = r"^([a-zA-Z0-9\.\-\_]+)\@([a-zA-Z0-9\.\-\_]+)"
+    match = re.match(connection_regex, name)
+    if not match:
+        raise ValueError(
+            f"Invalid cluster name: {name}. Must be in the format 'user@hostname'."
+        )
+    return _ConnectionInfo(hostname=match.group(2), user=match.group(1))
+
+
+def _get_slurm_connections() -> list[_ConnectionInfo]:
path: src/oumi/launcher/clouds/slurm_cloud.py
text: @user2:
would it make sense to move these utils into SlurmCloud class as `@user1`-s  to group it all in one place?

@author:
Done!