submit

JobSubmissionFailed

Bases: Exception

Raised when a job submission fails.

Source code in cluv/cli/submit.py

class JobSubmissionFailed(Exception):
    """Raised when a job submission fails."""

SubmissionProgress

Live, mutable tracking of one Submission's progress.

Tracks a submission from before it's even synced (state="SYNCING", no job yet), through submission (job known, state polled from sacct), to running and, possibly, cancellation.

A single flat list of these -- covering every submission, on every cluster, for one submit() call -- is all a live display needs to render the whole picture, from a plain rich.Live table (see render_job_table below) up to, eventually, a table shared across several concurrent submit/submit_first calls (rich.Live only supports one live region per console, so that would fuse several such lists together instead of replacing this one).

Returned by:

• Reference CLI submit wait_for_first_running_job

Used by:

• Reference CLI submit
  • render_job_table
  • run_scancel
  • submit_to_cluster
  • wait_for_first_running_job
  • wait_for_jobs_to_cancel

Source code in cluv/cli/submit.py

@dataclasses.dataclass
class SubmissionProgress:
    """Live, mutable tracking of one `Submission`'s progress.

    Tracks a submission from before it's even synced (``state="SYNCING"``, no `job` yet),
    through submission (``job`` known, state polled from `sacct`), to running and, possibly,
    cancellation.

    A single flat list of these -- covering every submission, on every cluster, for one
    `submit()` call -- is all a live display needs to render the whole picture, from a plain
    `rich.Live` table (see `render_job_table` below) up to, eventually, a table shared across
    several concurrent `submit`/`submit_first` calls (`rich.Live` only supports one live region
    per console, so that would fuse several such lists together instead of replacing this one).
    """

    submission: Submission
    state: JobState = "SYNCING"
    job: Job | None = None
    error: str | None = None

    @property
    def cluster(self) -> str:
        return self.submission.cluster

    @property
    def job_id(self) -> int | None:
        return self.job.job_id if self.job is not None else None

render_job_table

render_job_table(
    rows: list[SubmissionProgress],
    *,
    cancelling: bool = False,
) -> Table

Render the current state of every submission as a single table.

A plain rich.Live for now; the natural place to plug in a registry that fuses several concurrent submit/submit_first calls' rows into one shared live region later on.

Source code in cluv/cli/submit.py

def render_job_table(
    rows: list[SubmissionProgress], *, cancelling: bool = False
) -> rich.table.Table:
    """Render the current state of every submission as a single table.

    A plain `rich.Live` for now; the natural place to plug in a registry that fuses several
    concurrent `submit`/`submit_first` calls' rows into one shared live region later on.
    """
    title = "Waiting for jobs to cancel..." if cancelling else "Submitting jobs..."
    table = rich.table.Table(
        "Cluster",
        "Job ID",
        "Status",
        "Command",
        title=title,
        box=rich.box.ROUNDED,
        show_lines=True,
        header_style="bold white on #1a1a2e",
        title_style="bold cyan",
        expand=True,
    )
    for row in rows:
        table.add_row(
            row.cluster,
            str(row.job_id) if row.job_id is not None else "-",
            rich.text.Text(row.state, style=_state_style(row.state)),
            _short_command(row.submission),
        )
    return table

submit

submit(
    cluster: str,
    job_script: Path | None,
    sbatch_args: list[str],
    program_args: list[str],
    autocommit: bool = False,
    chunking: int | None = None,
    _skip_sync: bool = False,
    sync_datasets: bool = True,
    parsable: bool = False,
) -> Job | None

Submit a job to the given cluster (or all clusters if cluster=="first"), and return the Job object if successful.

If parsable is True, print only the job ID (or ':' when cluster is 'first') to stdout, for programmatic use, instead of the usual human-readable summary. Everything else (logs, the live jobs table, command outputs) is silenced, as with --quiet.

Returns None if the submission failed.

Source code in cluv/cli/submit.py

async def submit(
    cluster: str,
    job_script: Path | None,
    sbatch_args: list[str],
    program_args: list[str],
    autocommit: bool = False,
    chunking: int | None = None,
    _skip_sync: bool = False,
    sync_datasets: bool = True,
    parsable: bool = False,
) -> Job | None:
    """Submit a job to the given cluster (or all clusters if `cluster=="first"`),
    and return the Job object if successful.

    If `parsable` is True, print only the job ID (or '<cluster>:<job_id>' when `cluster`
    is 'first') to stdout, for programmatic use, instead of the usual human-readable summary.
    Everything else (logs, the live jobs table, command outputs) is silenced, as with `--quiet`.

    Returns None if the submission failed.
    """
    # `--parsable` promises that stdout carries nothing but the job id. `--quiet` already suppresses
    # everything the submission writes through the shared console (and the raw command outputs in
    # `cluv.remote.run`, which check `console.quiet` too), so it is all this needs to do.
    if parsable:
        console.quiet = True
    submit_command = build_submit_command(
        cluster=cluster, job_script=job_script, sbatch_args=sbatch_args, program_args=program_args
    )
    git_commit = ensure_clean_git_state(autocommit=autocommit, submit_command=submit_command)
    cluster_to_remote = await get_cluster_to_remote(cluster)

    job_submissions = [
        SubmissionProgress(submission=submission)
        for cluster_name, remote in cluster_to_remote.items()
        for submission in get_submissions(
            cluster_name,
            remote,
            job_script=job_script,
            sbatch_args=sbatch_args,
            program_args=program_args,
            chunking=chunking,
            git_commit=git_commit,
        )
    ]

    if not _skip_sync:
        remotes = [r for r in cluster_to_remote.values() if r]
        await sync_common_part(remotes, sync_datasets=sync_datasets)

    found_running_job = asyncio.Event()
    tasks = [
        asyncio.create_task(
            submit_to_cluster(
                cluster_name,
                remote,
                job_submissions=[
                    job_submission
                    for job_submission in job_submissions
                    if job_submission.cluster == cluster_name
                ],
                found_running_job=found_running_job,
                _skip_sync=_skip_sync,
                sync_datasets=sync_datasets,
            )
        )
        for cluster_name, remote in cluster_to_remote.items()
    ]

    cancelling = False

    def _render() -> rich.table.Table:
        return render_job_table(job_submissions, cancelling=cancelling)

    try:
        with Live(get_renderable=_render, console=console, refresh_per_second=1):
            first_running_row = await wait_for_first_running_job(
                job_submissions, cluster_to_remote, tasks, found_running_job
            )
            if first_running_row is None:
                console.log("All job submissions have failed! Exiting.")
                return None

            cancelling = True
            other_rows = [
                row
                for row in job_submissions
                if row is not first_running_row and row.job_id is not None
            ]
            await wait_for_jobs_to_cancel(other_rows, cluster_to_remote)
    except (KeyboardInterrupt, asyncio.CancelledError):
        # The user stopped `cluv submit` while jobs were still in flight -- cancel everything
        # that got a job id so far instead of leaving them running unattended.
        console.log("Interrupted by user. Cancelling all submitted jobs...")
        submitted_rows = [row for row in job_submissions if row.job_id is not None]
        await run_scancel(submitted_rows)
        raise

    job = first_running_row.job
    assert job is not None

    if parsable:
        print(f"{job.cluster}:{job.job_id}" if cluster == "first" else job.job_id)
    else:
        console.print(
            f"Successfully submitted job {job.job_id} on cluster {job.cluster}.\n"
            f"Use `ssh {job.cluster} sacct -j {job.job_id}` to view its status, and `cluv sync"
            f" {job.cluster}` to fetch results once it is complete.",
            style="green",
        )

    save_job(job)
    return job

wait_for_first_running_job

wait_for_first_running_job(
    job_submissions: list[SubmissionProgress],
    cluster_to_remote: dict[str, Remote | None],
    tasks: list[Task],
    found_running_job: Event,
    max_wait_time_seconds: int = 60,
) -> SubmissionProgress | None

Poll sacct until one submitted job starts running, or every submission has failed.

Mutates rows in place with the latest known job id / state, so a live display can render them at any point during this wait. Sets found_running_job the moment a job starts, so that clusters which haven't submitted their own jobs yet can skip doing so.

Returns the row for the job that started, or None if every submission ended up failing.

Source code in cluv/cli/submit.py

async def wait_for_first_running_job(
    job_submissions: list[SubmissionProgress],
    cluster_to_remote: dict[str, Remote | None],
    tasks: list[asyncio.Task],
    found_running_job: asyncio.Event,
    max_wait_time_seconds: int = 60,
) -> SubmissionProgress | None:
    """Poll `sacct` until one submitted job starts running, or every submission has failed.

    Mutates `rows` in place with the latest known job id / state, so a live display can render
    them at any point during this wait. Sets `found_running_job` the moment a job starts, so
    that clusters which haven't submitted their own jobs yet can skip doing so.

    Returns the row for the job that started, or None if every submission ended up failing.
    """
    delay = 1
    while True:
        all_tasks_done = all(task.done() for task in tasks)
        submitted = [row for row in job_submissions if row.job_id is not None]

        by_cluster = group_by_cluster(submitted)

        n_pending_jobs = 0
        if by_cluster:
            states_per_cluster = await asyncio.gather(
                *(
                    run_saccts(cluster_to_remote[cluster], [row.job_id for row in cluster_rows])
                    for cluster, cluster_rows in by_cluster.items()
                )
            )
            for cluster_rows, states in zip(by_cluster.values(), states_per_cluster):
                for row, state in zip(cluster_rows, states):
                    row.state = state
                    if row.state.startswith(("RUNNING", "COMPLETED")):
                        found_running_job.set()
                        return row
                    elif row.state.startswith(("PENDING")):
                        n_pending_jobs += 1

        # Skip the wait if only one job is pending (if only one job is submitted or all other jobs
        # failed).
        if all_tasks_done and n_pending_jobs == 1:
            console.log("Only one job pending. Skipping wait for a running job.")
            return next(row for row in submitted if row.state.startswith("PENDING"))

        all_failed = bool(submitted) and all(
            row.state.startswith(tuple(FAILED_JOB_STATES)) for row in submitted
        )
        if all_tasks_done and (not submitted or all_failed):
            return None

        await asyncio.sleep(delay)
        delay = min(delay * 2, max_wait_time_seconds)

wait_for_jobs_to_cancel

wait_for_jobs_to_cancel(
    job_submissions: list[SubmissionProgress],
    cluster_to_remote: dict[str, Remote | None],
    max_wait_time_seconds: int = 60,
) -> None

Cancel every (already-submitted) job in rows, and wait until they're all done.

Source code in cluv/cli/submit.py

async def wait_for_jobs_to_cancel(
    job_submissions: list[SubmissionProgress],
    cluster_to_remote: dict[str, Remote | None],
    max_wait_time_seconds: int = 60,
) -> None:
    """Cancel every (already-submitted) job in `rows`, and wait until they're all done."""
    to_cancel = [
        job for job in job_submissions if not job.state.startswith(("CANCELLED", "COMPLETED"))
    ]
    if not to_cancel:
        return

    await run_scancel(to_cancel)

    delay = 1
    while to_cancel:
        by_cluster = group_by_cluster(to_cancel)
        states_per_cluster = await asyncio.gather(
            *(
                run_saccts(cluster_to_remote[cluster], [row.job_id for row in cluster_rows])
                for cluster, cluster_rows in by_cluster.items()
            )
        )
        for cluster_rows, states in zip(by_cluster.values(), states_per_cluster):
            for row, state in zip(cluster_rows, states):
                if state.startswith("CANCELLED") or state == "FAILED":
                    # "CANCELLED by <uid>", and a stray "FAILED" job step on some clusters
                    # (while the rest of the job is "CANCELLED"), both just mean cancelled.
                    state = "CANCELLED"
                row.state = state

        to_cancel = [
            row for row in to_cancel if not row.state.startswith(("CANCELLED", "COMPLETED"))
        ]
        if to_cancel:
            await asyncio.sleep(delay)
            delay = min(delay * 2, max_wait_time_seconds)

    console.log(f"Cancelled {len(job_submissions)} job submission(s).")

run_scancel

run_scancel(rows: list[SubmissionProgress]) -> None

Cancel the (already-submitted) jobs behind rows, grouped by remote.

Source code in cluv/cli/submit.py

async def run_scancel(rows: list[SubmissionProgress]) -> None:
    """Cancel the (already-submitted) jobs behind `rows`, grouped by remote."""
    if not rows:
        return
    by_remote: dict[Remote | None, list[SubmissionProgress]] = {}
    for row in rows:
        by_remote.setdefault(row.submission.remote, []).append(row)

    async def cancel(remote: Remote | None, cluster_rows: list[SubmissionProgress]) -> None:
        job_ids = [row.job_id for row in cluster_rows]
        scancel_command = f"scancel {' '.join(map(str, job_ids))}"
        if remote is not None:
            await remote.get_output(scancel_command, hide=True)
        else:
            await run(tuple(shlex.split(scancel_command)), hide=True)

    await asyncio.gather(
        *(cancel(remote, cluster_rows) for remote, cluster_rows in by_remote.items())
    )

submit_to_cluster

submit_to_cluster(
    cluster: str,
    remote: Remote | None,
    job_submissions: list[SubmissionProgress],
    found_running_job: Event,
    _skip_sync: bool = False,
    sync_datasets: bool = True,
) -> None

Sync then submit every submission for one cluster, in parallel.

Source code in cluv/cli/submit.py

async def submit_to_cluster(
    cluster: str,
    remote: Remote | None,
    job_submissions: list[SubmissionProgress],
    found_running_job: asyncio.Event,
    _skip_sync: bool = False,
    sync_datasets: bool = True,
) -> None:
    """Sync then submit every submission for one cluster, in parallel."""
    if not _skip_sync:
        await sync_per_cluster_part(remote, sync_datasets=sync_datasets)

    if found_running_job.is_set():
        # If a job has already started on another cluster, we don't need to submit more jobs.
        console.log(
            f"Skipping submission of jobs to cluster {cluster} because a job "
            f"has already started on another cluster."
        )
        for row in job_submissions:
            row.state = "SKIPPED"
        return

    for row in job_submissions:
        row.state = "SUBMITTING"

    results = await asyncio.gather(
        *(submit_job(row.submission) for row in job_submissions),
        return_exceptions=True,
    )

    assert len(results) == len(job_submissions)
    for row, result in zip(job_submissions, results):
        if isinstance(result, Job):
            row.job = result
            row.state = "PENDING"
        elif isinstance(result, JobSubmissionFailed):
            row.error = str(result)
            row.state = "FAILED"
            console.log(f"[red]{result}[/red]")
        else:
            assert isinstance(result, BaseException)
            raise result

get_submissions

get_submissions(
    cluster: str,
    remote: Remote | None,
    *,
    job_script: Path | None,
    sbatch_args: list[str],
    program_args: list[str],
    chunking: int | None,
    git_commit: str,
) -> list[Submission]

Expand the possible job configurations for a cluster. Returns a list of Submission objects.

Does not do the actual job submission with sbatch.

Source code in cluv/cli/submit.py

def get_submissions(
    cluster: str,
    remote: Remote | None,
    *,
    job_script: Path | None,
    sbatch_args: list[str],
    program_args: list[str],
    chunking: int | None,
    git_commit: str,
) -> list[Submission]:
    """Expand the possible job configurations for a cluster. Returns a list of `Submission` objects.

    Does *not* do the actual job submission with `sbatch`.
    """
    submissions: list[Submission] = []
    config = get_cluv_config()
    cluster_config = config.get_cluster_config(cluster)
    job_resources_options = cluster_config.sbatch_args

    if job_script is None:
        if cluster_config.job_script_path is None:
            raise ValueError(
                f"No job script specified for cluster {cluster!r}, and no default job script "
                f"path set in the config."
            )
        job_script = Path(os.path.expandvars(str(cluster_config.job_script_path)))
    if not job_script.exists():
        raise ValueError(
            f"The job script ({job_script}) does not exist on this machine. Even though it "
            f"can be customized per cluster, it needs to exist locally, since cluv needs to "
            f"read its header to infer sbatch defaults."
        )

    job_env_vars = get_job_env_vars(
        cluster=cluster, git_commit=git_commit, cluster_config=cluster_config
    )
    project_dir_on_cluster = cluster_config.project_dir
    cluster_job_script_path = get_cluster_job_script_path(
        local_job_script_path=job_script, cluster=cluster, cluster_config=cluster_config
    )
    for job_resources in job_resources_options:
        job_resources = merge_sbatch_args(from_config=job_resources, from_cli=sbatch_args)
        n_chunks, job_resources = apply_chunking(
            job_resources, job_script=job_script, chunking=chunking, env_vars=job_env_vars
        )
        job_resources = add_cluv_sbatch_args(
            job_resources, job_script=job_script, cluster=cluster, cluster_config=cluster_config
        )
        sbatch_command = get_sbatch_command(
            env_vars=job_env_vars,
            job_script=cluster_job_script_path,
            sbatch_args=job_resources,
            program_args=program_args,
            project_dir_on_cluster=project_dir_on_cluster,
        )
        submissions.append(
            Submission(
                cluster=cluster,
                remote=remote,
                job_script=job_script,
                sbatch_args=job_resources,
                program_args=program_args,
                sbatch_command=sbatch_command,
                n_chunks=n_chunks,
                git_commit=git_commit,
            )
        )
    return submissions

merge_sbatch_args

merge_sbatch_args(
    from_config: SbatchArgs, from_cli: list[str]
) -> SbatchArgs

Merge the sbatch args from the config and from the CLI, with CLI args taking precedence.

-t is normalized to time (its long-flag alias) as it's merged in, so --time=1:00:00 -t=2:00:00 -- config or CLI, either order -- resolves to a single time value (the last one written) instead of leaving two separate keys for what's really the same sbatch option.

Source code in cluv/cli/submit.py

def merge_sbatch_args(from_config: SbatchArgs, from_cli: list[str]) -> SbatchArgs:
    """Merge the sbatch args from the config and from the CLI, with CLI args taking precedence.

    `-t` is normalized to `time` (its long-flag alias) as it's merged in, so `--time=1:00:00
    -t=2:00:00` -- config or CLI, either order -- resolves to a single `time` value (the last
    one written) instead of leaving two separate keys for what's really the same sbatch option.
    """
    sbatch_args_from_config = sbatch_args_to_list(from_config)
    return sbatch_args_from_list(sbatch_args_from_config + from_cli)

check_path_is_safe_to_interpolate

check_path_is_safe_to_interpolate(
    path: PurePosixPath | str, setting: str
) -> None

Raise a ValueError if path can't be interpolated into the sbatch command as-is.

Values that may contain environment variables are written into the command unquoted, so that the cluster's login shell is the one that expands them (see get_sbatch_command). Escaping them with shlex.quote first would stop exactly that, so anything else that shell treats specially has to be rejected up front instead: a space would split the value into two sbatch arguments, and a ; would end the command and start another one.

check_path_is_safe_to_interpolate("$SCRATCH/logs/imagenet", "results_path") check_path_is_safe_to_interpolate("/home/me/my logs", "results_path") ... # doctest: +ELLIPSIS Traceback (most recent call last): ... ValueError: The results_path '/home/me/my logs' contains a ' ', ...

Source code in cluv/cli/submit.py

def check_path_is_safe_to_interpolate(path: PurePosixPath | str, setting: str) -> None:
    """Raise a `ValueError` if `path` can't be interpolated into the sbatch command as-is.

    Values that may contain environment variables are written into the command *unquoted*, so that
    the cluster's login shell is the one that expands them (see `get_sbatch_command`). Escaping them
    with `shlex.quote` first would stop exactly that, so anything else that shell treats specially
    has to be rejected up front instead: a space would split the value into two `sbatch` arguments,
    and a `;` would end the command and start another one.

    >>> check_path_is_safe_to_interpolate("$SCRATCH/logs/imagenet", "results_path")
    >>> check_path_is_safe_to_interpolate("/home/me/my logs", "results_path")
    ... # doctest: +ELLIPSIS
    Traceback (most recent call last):
        ...
    ValueError: The results_path '/home/me/my logs' contains a ' ', ...
    """
    if match := _UNSAFE_PATH_CHARS.search(str(path)):
        raise ValueError(
            f"The {setting} {str(path)!r} contains a {match.group()!r}, which cluv can't pass to "
            f"sbatch safely: it goes into a `bash --login -c ...` command unquoted, so that the "
            f"cluster's login shell expands variables like $SCRATCH in it. Please use a value "
            f"without whitespace or shell metacharacters."
        )

add_cluv_sbatch_args

add_cluv_sbatch_args(
    sbatch_args: SbatchArgs,
    job_script: Path,
    cluster: str,
    cluster_config: ClusterConfig,
) -> SbatchArgs

• Add the --output flag (So that outputs are created in the results_path for the run prescribed by Cluv)
• Add the --job-name flag (So that we can identify the cluv jobs later)
• Add the --export=ALL flag (since trillium and trillium-gpu apparently have --export=None as default).
• Add the --chdir flag to move to the project folder when running the command.

Returns a new dict; the one passed in is left alone.

Source code in cluv/cli/submit.py

def add_cluv_sbatch_args(
    sbatch_args: SbatchArgs,
    job_script: Path,
    cluster: str,
    cluster_config: ClusterConfig,
) -> SbatchArgs:
    """
    - Add the --output flag (So that outputs are created in the `results_path` for the run prescribed by Cluv)
    - Add the --job-name flag (So that we can identify the cluv jobs later)
    - Add the --export=ALL flag (since trillium and trillium-gpu apparently have `--export=None` as default).
    - Add the --chdir flag to move to the project folder when running the command.

    Returns a new dict; the one passed in is left alone.
    """
    sbatch_args = sbatch_args.copy()

    base_name = sbatch_args.get("job-name") or Path(job_script).stem
    sbatch_args["job-name"] = f"cluv-{base_name}"

    if "output" not in sbatch_args and (
        _header_output := next(
            (
                line
                for line in job_script.read_text().splitlines()
                if line.strip().startswith("#SBATCH") and "--output" in line
            ),
            None,
        )
    ):
        logger.warning(
            f"[yellow]The job script {job_script} sets {_header_output.strip()!r}, which "
            f"will be overridden by cluv's --output so that results can be synced "
            f"back. Consider using cluv in your Python script to decide where to store "
            f"results instead.[/yellow]"
        )

    # Chunked (job array) jobs need `%A`/`%a` (array job id / task id) instead of `%j`.
    if "array" in sbatch_args:
        sbatch_args["output"] = str(cluster_config.results_path / f"{cluster}_%A/slurm-%A_%a.out")
    else:
        sbatch_args["output"] = str(cluster_config.results_path / f"{cluster}_%j/slurm-%j.out")

    # NOTE: `cluster_config.project_dir` already falls back to the global `project_dir`, so the
    # `$HOME/<project>` default below is only used when neither is set.
    local_project_dir = find_pyproject().parent
    remote_project_dir = cluster_config.project_dir or (
        PurePosixPath("$HOME") / local_project_dir.relative_to(Path.home())
    )
    sbatch_args["chdir"] = str(remote_project_dir)
    # Some clusters (trillium, trillium-gpu) have a wrapper around `sbatch` that sets `--export=NONE` by default,
    # which would discard the environment variables.
    sbatch_args["export"] = "ALL"
    return sbatch_args

get_sbatch_command

get_sbatch_command(
    job_script: PurePosixPath,
    sbatch_args: SbatchArgs,
    program_args: list[str],
    env_vars: dict[str, str],
    project_dir_on_cluster: PurePosixPath | None = None,
) -> str

Generate the command to submit the job via sbatch on the cluster.

Source code in cluv/cli/submit.py

def get_sbatch_command(
    job_script: PurePosixPath,
    sbatch_args: SbatchArgs,
    program_args: list[str],
    env_vars: dict[str, str],
    project_dir_on_cluster: PurePosixPath | None = None,
) -> str:
    """Generate the command to submit the job via `sbatch` on the cluster."""
    if job_script.is_absolute():
        raise RuntimeError(
            f"The job script path {str(job_script)!r} is an absolute path on this machine, but it "
            f"has to be the path of the script *on the cluster* (relative to the project there, or "
            f"starting with a variable like $HOME that the cluster's shell expands). This is what "
            f"`get_cluster_job_script_path` returns."
        )

    sbatch_flags = sbatch_args_to_list(sbatch_args)
    env_vars_prefix = " ".join(f"{k}={v}" for k, v in env_vars.items())

    # These are interpolated unquoted, so the cluster's login shell expands any env vars in them
    # (`$SCRATCH`, `$HOME`); they can't be `shlex`-escaped first, since quoting would both stop
    # that expansion and close the surrounding single-quoted string.
    check_path_is_safe_to_interpolate(job_script, "job_script")
    if isinstance(chdir := sbatch_args.get("chdir"), str):
        check_path_is_safe_to_interpolate(chdir, "chdir")
    if isinstance(output := sbatch_args.get("output"), str):
        check_path_is_safe_to_interpolate(output, "output")
    for name, value in env_vars.items():
        # Same deal: `UV_CACHE_DIR=$SCRATCH/.cache/uv` has to stay unquoted to be expanded on the
        # cluster, so a value with a space in it would make the login shell read the second word
        # as the command to run, and `sbatch` would never be reached.
        check_path_is_safe_to_interpolate(value, f"{name} environment variable")

    # `program_args` is the one part that is *not* meant to be expanded here: it is whatever the
    # user wrote after `--`, so it gets escaped, and a `$SLURM_TMPDIR` in it survives for the job
    # itself to expand. That only holds together because the whole inner command is quoted in one
    # go below - `shlex.join`'s quotes would otherwise close a hand-written `'...'` around it, and
    # an argument containing a space would break apart (POSIX single quotes don't nest).
    if env_vars_prefix:
        env_vars_prefix += "; "
    cd_command = ""
    if project_dir_on_cluster:
        cd_command = f"cd {project_dir_on_cluster} && "
    inner_command = (
        f"{env_vars_prefix}{cd_command}sbatch --parsable {' '.join(sbatch_flags)} {job_script} "
        f"{shlex.join(program_args)}"
    )
    return f"bash --login -c {shlex.quote(inner_command)}"

submit_job

submit_job(submission: Submission) -> Job

Does the actual sbatch call.

Raise a JobSubmissionFailed if the job submission fails for some reason.

Source code in cluv/cli/submit.py

async def submit_job(submission: Submission) -> Job:
    """Does the actual sbatch call.

    Raise a `JobSubmissionFailed` if the job submission fails for some reason.
    """
    display = display_commands.get()
    hide = not display
    warn = not raise_on_command_error.get()

    if submission.remote is not None:
        result = await submission.remote.run(
            submission.sbatch_command, display=display, warn=warn, hide=hide
        )
    else:
        result = await run(
            tuple(shlex.split(submission.sbatch_command)), _display=display, warn=warn, hide=hide
        )

    if result.returncode != 0:
        raise JobSubmissionFailed(
            f"Failed to submit job on cluster {submission.cluster}: "
            f"{result.stderr or result.stdout}"
        )

    return Job(
        cluster=submission.cluster,
        remote=submission.remote,
        job_script=submission.job_script,
        sbatch_args=submission.sbatch_args,
        program_args=submission.program_args,
        sbatch_command=submission.sbatch_command,
        n_chunks=submission.n_chunks,
        git_commit=submission.git_commit,
        job_id=int(result.stdout.strip()),
        submitted_at=datetime.datetime.now(),
    )

build_submit_command

build_submit_command(
    cluster: str,
    job_script: str | Path | PurePosixPath | None,
    sbatch_args: list[str],
    program_args: list[str],
) -> str

Build the local cluv submit command line used to launch the job.

Source code in cluv/cli/submit.py

def build_submit_command(
    cluster: str,
    job_script: str | Path | PurePosixPath | None,
    sbatch_args: list[str],
    program_args: list[str],
) -> str:
    """Build the local `cluv submit` command line used to launch the job."""
    command_parts = ["cluv", "submit", cluster]
    if job_script is not None:
        command_parts.append(str(job_script))
    command_parts.extend(sbatch_args)
    if program_args:
        command_parts.extend(["--", *program_args])
    return shlex.join(command_parts)

create_submit_commit

create_submit_commit(submit_command: str) -> None

Create a commit with tracked changes and include the launched job command in the body.

Source code in cluv/cli/submit.py

def create_submit_commit(submit_command: str) -> None:
    """Create a commit with tracked changes and include the launched job command in the body."""
    try:
        subprocess.run(["git", "add", "-u"], check=True, capture_output=True, text=True)
        subprocess.run(
            [
                "git",
                "commit",
                "-m",
                "cluv submit: auto-commit tracked changes",
                "-m",
                f"Launched job command:\n\n{submit_command}",
            ],
            check=True,
            capture_output=True,
            text=True,
        )
    except subprocess.CalledProcessError as err:
        error_text = (err.stderr or err.stdout or str(err)).strip()
        console.print(
            "[red]Failed to create automatic submit commit before job submission:[/red] "
            f"{error_text}"
        )
        raise

ensure_clean_git_state

ensure_clean_git_state(
    autocommit: bool = False,
    submit_command: str | None = None,
) -> str

Check git is clean locally and return the current commit hash.

Source code in cluv/cli/submit.py

def ensure_clean_git_state(autocommit: bool = False, submit_command: str | None = None) -> str:
    """
    Check git is clean locally and return the current commit hash.
    """
    git_status = subprocess.run(["git", "status", "--porcelain"], capture_output=True, text=True)
    dirty_lines = [line for line in git_status.stdout.splitlines() if not line.startswith("??")]
    if dirty_lines:
        if autocommit:
            if submit_command is None:
                raise ValueError("submit_command is required when autocommit=True")
            create_submit_commit(submit_command)
        elif not (os.environ.get("SKIP_CLEAN_GIT_CHECK", "0") == "1"):
            console.print(
                "Working directory is dirty. Please commit your changes before submitting, "
                "or use `--autocommit` (`hydra.launcher.autocommit=True` when using Hydra).",
                style="red",
            )
            sys.exit(1)

    # In GitHub Actions PR jobs we can be on a detached merge commit that doesn't exist on
    # the synced remote checkout. Prefer the branch tip commit in that case.
    current_branch = subprocess.check_output(
        ["git", "rev-parse", "--abbrev-ref", "HEAD"], text=True
    ).strip()
    if current_branch == "HEAD" and os.environ.get("GITHUB_ACTIONS"):
        github_head_ref = os.environ.get("GITHUB_HEAD_REF", "").strip()
        if github_head_ref:
            remote_head_ref = f"origin/{github_head_ref}"
            remote_head_result = subprocess.run(
                ["git", "rev-parse", "--verify", remote_head_ref],
                capture_output=True,
                text=True,
            )
            if remote_head_result.returncode == 0:
                return remote_head_result.stdout.strip()
            console.log(
                f"Could not resolve {remote_head_ref}. Falling back to local HEAD commit.",
                style="yellow",
            )

    # Capture current commit hash.
    return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()