cosmotron_mcp.tools.jobs

submit_job

submit_job(session_dir: str, argv: list[str], desc: str = 'job', jobspec: dict | None = None) -> dict

Submit a command as a non-blocking background job; return immediately.

argv is the command as a list (e.g. ["cosmosis", "pipeline.ini"]). jobspec is an optional resource dict (backend "local"|"slurm", cpus, mem, walltime, partition, nodes, env); omitted → a local single-process job. For a remote Slurm job pass {"backend": "slurm", "site": "<site>"} (a sites.yaml entry) — resources come from the tool's abstract resource_class translated by the site, so nothing else is usually needed; a remote job starts in status="staging". Logs and a returncode marker land under {session_dir}/jobs/{job_id}/. Poll with check_job(session_dir, job_id) — do NOT block waiting for completion.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory the job belongs to.	required
argv	list[str]	Command as a list, e.g. ["cosmosis", "pipeline.ini"].	required
desc	str	Short human-readable label for the job.	'job'
jobspec	dict | None	Optional resource dict (backend "local"|"slurm", cpus, mem, walltime, partition, nodes, env, site); omitted → a local single-process job.	None

Returns:

Type	Description
dict	{job_id, status, run_dir, backend} (status="staging" for a
dict	fresh Slurm submit).

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def submit_job(
    session_dir: str, argv: list[str], desc: str = "job", jobspec: dict | None = None
) -> dict:
    """Submit a command as a non-blocking background job; return immediately.

    `argv` is the command as a list (e.g. `["cosmosis", "pipeline.ini"]`). `jobspec`
    is an optional resource dict (`backend` "local"|"slurm", `cpus`, `mem`,
    `walltime`, `partition`, `nodes`, `env`); omitted → a local single-process job.
    For a remote Slurm job pass `{"backend": "slurm", "site": "<site>"}` (a
    `sites.yaml` entry) — resources come from the tool's abstract
    `resource_class` translated by the site, so nothing else is usually
    needed; a remote job starts in `status="staging"`. Logs and a
    `returncode` marker land under `{session_dir}/jobs/{job_id}/`. Poll
    with `check_job(session_dir, job_id)` — do NOT block waiting for completion.

    Args:
        session_dir: Session directory the job belongs to.
        argv: Command as a list, e.g. ``["cosmosis", "pipeline.ini"]``.
        desc: Short human-readable label for the job.
        jobspec: Optional resource dict (``backend`` ``"local"``|``"slurm"``,
            ``cpus``, ``mem``, ``walltime``, ``partition``, ``nodes``,
            ``env``, ``site``); omitted → a local single-process job.

    Returns:
        ``{job_id, status, run_dir, backend}`` (``status="staging"`` for a
        fresh Slurm submit).
    """
    _require_human_gates(session_dir)
    result = _submit_job(session_dir, argv, desc=desc, jobspec=jobspec)
    _log_tool_call(session_dir, "submit_job", {"argv": argv, "desc": desc, "jobspec": jobspec})
    return result

check_job

check_job(session_dir: str, job_id: str, wait: float = 0.0) -> dict

Poll a background job's status; return its state + log tails.

Reads the persisted record, polls the backend (returncode marker / pid liveness — the local analogue of squeue/sacct), rewrites the record, and tails the logs. wait=0 (default) returns immediately. Pass wait seconds (e.g. wait=45, clamped to 45) to BLOCK server-side until the job is terminal or the time elapses — use this instead of calling check_job many times in a row while a chain runs (it returns early the moment the job finishes).

Never does network I/O, even for a remote (slurm) job: a detached watcher process polls the cluster and pulls results back on its own, and this call just reads the record that watcher maintains. So it cannot hang, and a long remote run is followed by re-issuing wait=45 until the status is terminal.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory the job belongs to.	required
job_id	str	Job identifier returned by submit_job.	required
wait	float	Seconds to block server-side until the job is terminal (clamped to 45); 0 (default) returns immediately.	0.0

Returns:

Type	Description
dict	``{job_id, status, returncode, run_dir, stdout_tail, stderr_tail,
dict	waited, running_secs, last_activity_secs}.status`` is one of
dict	staging, pending, running, finalizing, done,
dict	failed, cancelled — staging/finalizing only occur for
dict	a remote (slurm) job while its detached watcher is pushing/pulling
dict	files; a local job never reports them. A remote job also carries
dict	resource_warnings when present — e.g. a jobspec["queue"]/
dict	## remote: queue: <name> naming no entry in the site's queues:
dict	map silently falls back to the site's default queue; this field is
dict	the only place that's visible, so check it after any remote submit
dict	that named a non-default queue or partition. A HEAVY_TOOLS job also
dict	carries result (the tool's return dict) once status == "done", or
dict	result_path + result_truncated: true if it exceeds ~256 KB.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def check_job(session_dir: str, job_id: str, wait: float = 0.0) -> dict:
    """Poll a background job's status; return its state + log tails.

    Reads the persisted record, polls the backend (returncode marker / pid
    liveness — the local analogue of squeue/sacct), rewrites the record, and tails
    the logs. `wait=0` (default) returns immediately. **Pass `wait` seconds (e.g.
    `wait=45`, clamped to 45) to BLOCK server-side until the job is terminal or the
    time elapses** — use this instead of calling `check_job` many times in a row
    while a chain runs (it returns early the moment the job finishes).

    Never does network I/O, even for a remote (slurm) job: a detached watcher
    process polls the cluster and pulls results back on its own, and this call
    just reads the record that watcher maintains. So it cannot hang, and a long
    remote run is followed by re-issuing `wait=45` until the status is terminal.

    Args:
        session_dir: Session directory the job belongs to.
        job_id: Job identifier returned by `submit_job`.
        wait: Seconds to block server-side until the job is terminal (clamped
            to 45); `0` (default) returns immediately.

    Returns:
        ``{job_id, status, returncode, run_dir, stdout_tail, stderr_tail,
        waited, running_secs, last_activity_secs}``. ``status`` is one of
        ``staging``, ``pending``, ``running``, ``finalizing``, ``done``,
        ``failed``, ``cancelled`` — ``staging``/``finalizing`` only occur for
        a remote (slurm) job while its detached watcher is pushing/pulling
        files; a local job never reports them. A remote job also carries
        ``resource_warnings`` when present — e.g. a ``jobspec["queue"]``/
        ``## remote: queue: <name>`` naming no entry in the site's ``queues:``
        map silently falls back to the site's ``default`` queue; this field is
        the only place that's visible, so check it after any remote submit
        that named a non-default queue or partition. A `HEAVY_TOOLS` job also
        carries `result` (the tool's return dict) once `status == "done"`, or
        `result_path` + `result_truncated: true` if it exceeds ~256 KB.
    """
    return _check_job(session_dir, job_id, wait=wait)

list_jobs

list_jobs(session_dir: str) -> list[dict]

List all jobs for a session (most recent first), polling each for fresh status.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory to list jobs for.	required

Returns:

Type	Description
list[dict]	A list of {job_id, desc, status, backend, created_at, completed_at},
list[dict]	plus (remote/slurm jobs only) site, slurm_state, and stale
list[dict]	(the watcher hasn't updated the record recently — check check_job
list[dict]	for a fresh poll rather than treating stale as an error).

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def list_jobs(session_dir: str) -> list[dict]:
    """List all jobs for a session (most recent first), polling each for fresh status.

    Args:
        session_dir: Session directory to list jobs for.

    Returns:
        A list of ``{job_id, desc, status, backend, created_at, completed_at}``,
        plus (remote/slurm jobs only) ``site``, ``slurm_state``, and ``stale``
        (the watcher hasn't updated the record recently — check `check_job`
        for a fresh poll rather than treating `stale` as an error).
    """
    return _list_jobs(session_dir)

cancel_job

cancel_job(session_dir: str, job_id: str, force: bool = False) -> dict

Best-effort cancel a running job (local: kill its whole process group).

Escalates SIGTERM → ~5 s grace → SIGKILL so a stuck mpirun/cosmosis MPI run is actually stopped. Pass force=True to SIGKILL immediately (runaway / overlong run).

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory the job belongs to.	required
job_id	str	Job identifier returned by submit_job.	required
force	bool	SIGKILL immediately instead of escalating from SIGTERM.	False

Returns:

Type	Description
dict	{job_id, status} — status="cancelled" on success, or the
dict	terminal status if the job had already finished.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def cancel_job(session_dir: str, job_id: str, force: bool = False) -> dict:
    """Best-effort cancel a running job (local: kill its whole process group).

    Escalates SIGTERM → ~5 s grace → SIGKILL so a stuck mpirun/cosmosis MPI run is
    actually stopped. Pass ``force=True`` to SIGKILL immediately (runaway / overlong run).

    Args:
        session_dir: Session directory the job belongs to.
        job_id: Job identifier returned by `submit_job`.
        force: SIGKILL immediately instead of escalating from SIGTERM.

    Returns:
        ``{job_id, status}`` — ``status="cancelled"`` on success, or the
        terminal status if the job had already finished.
    """
    return _cancel_job(session_dir, job_id, force=force)

list_remote_artefacts

list_remote_artefacts(session_dir: str, job_id: str | None = None) -> list[dict]

List artefacts registered as remote-only (bytes never pulled to the Spark).

Pure local read of results/artefact_registry.json — zero network I/O, always instant. A remote job reporting status=done with no pull_error/merge_error does NOT guarantee every artefact it produced is local: the automatic pull only fetches a fixed set of file types on job completion. If a downstream tool reports a missing artefact that a prior remote job's receipt claims it produced, call this FIRST — the artefact may already exist on the remote site and just need pull_remote_artefacts, not a recompute or resubmit.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory to inspect.	required
job_id	str | None	Optional job id (from submit_job/compute_*_from_session) to filter to that job's outputs (best-effort match on the job's desc).	None

Returns:

Type	Description
list[dict]	A list of {type, path, site, generated_by, bin_index}.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def list_remote_artefacts(session_dir: str, job_id: str | None = None) -> list[dict]:
    """List artefacts registered as remote-only (bytes never pulled to the Spark).

    Pure local read of `results/artefact_registry.json` — zero network I/O, always
    instant. A remote job reporting `status=done` with no `pull_error`/`merge_error`
    does NOT guarantee every artefact it produced is local: the automatic pull only
    fetches a fixed set of file types on job completion. If a downstream tool
    reports a missing artefact that a prior remote job's receipt claims it
    produced, call this FIRST — the artefact may already exist on the remote site
    and just need `pull_remote_artefacts`, not a recompute or resubmit.

    Args:
        session_dir: Session directory to inspect.
        job_id: Optional job id (from `submit_job`/`compute_*_from_session`) to
            filter to that job's outputs (best-effort match on the job's `desc`).

    Returns:
        A list of ``{type, path, site, generated_by, bin_index}``.
    """
    return _list_remote_artefacts(session_dir, job_id=job_id)

pull_remote_artefacts

pull_remote_artefacts(session_dir: str, job_id: str, paths: list[str] | None = None) -> dict

Fetch specific remote-only artefacts for a job back to the Spark; returns immediately.

Spawns a detached pull phase (same idiom as the automatic stage/watch/ finalize lifecycle — no network I/O happens inside this call, so it cannot hang or take the MCP connection down). Poll with check_remote_pull.

Use this instead of recomputing or resubmitting when list_remote_artefacts shows a job produced an artefact that never made it back locally.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory the job belongs to.	required
job_id	str	Job identifier (must be a remote/slurm job that reached the cluster — i.e. has a remote_run_dir).	required
paths	list[str] | None	Session-relative paths to pull (e.g. ["results/maps/delta_map_bin_00.fits"]). Omit to pull every artefact list_remote_artefacts currently tags remote-only for this job.	None

Returns:

Type	Description
dict	{status: "pulling", job_id, next: "check_remote_pull"}, or
dict	{status: "failed", error} if the job never reached the cluster.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def pull_remote_artefacts(session_dir: str, job_id: str, paths: list[str] | None = None) -> dict:
    """Fetch specific remote-only artefacts for a job back to the Spark; returns immediately.

    Spawns a detached ``pull`` phase (same idiom as the automatic stage/watch/
    finalize lifecycle — no network I/O happens inside this call, so it cannot
    hang or take the MCP connection down). Poll with `check_remote_pull`.

    Use this instead of recomputing or resubmitting when `list_remote_artefacts`
    shows a job produced an artefact that never made it back locally.

    Args:
        session_dir: Session directory the job belongs to.
        job_id: Job identifier (must be a remote/slurm job that reached the
            cluster — i.e. has a `remote_run_dir`).
        paths: Session-relative paths to pull (e.g.
            ``["results/maps/delta_map_bin_00.fits"]``). Omit to pull every
            artefact `list_remote_artefacts` currently tags remote-only for
            this job.

    Returns:
        ``{status: "pulling", job_id, next: "check_remote_pull"}``, or
        ``{status: "failed", error}`` if the job never reached the cluster.
    """
    return _pull_remote_artefacts(session_dir, job_id, paths=paths)

check_remote_pull

check_remote_pull(session_dir: str, job_id: str) -> dict

Poll an on-demand pull_remote_artefacts request; local file read only.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory the job belongs to.	required
job_id	str	Job identifier passed to pull_remote_artefacts.	required

Returns:

Type	Description
dict	{status: "pulling"|"done"|"failed", pulled_paths, log_tail}.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def check_remote_pull(session_dir: str, job_id: str) -> dict:
    """Poll an on-demand `pull_remote_artefacts` request; local file read only.

    Args:
        session_dir: Session directory the job belongs to.
        job_id: Job identifier passed to `pull_remote_artefacts`.

    Returns:
        ``{status: "pulling"|"done"|"failed", pulled_paths, log_tail}``.
    """
    return _check_remote_pull(session_dir, job_id)