compute: process-tree cleanup, run handles keep their settings, SkyPilot SDK and copy transport

CHANGELOG.md

@@ -54,9 +54,38 @@ and uses semantic versioning while the public API stabilizes.
 
   The other inputs cover what differs between repositories: `sync-args`, `docs-sync-args`,
   `python-versions`, `pytest-args`, `test-env`, `import-modules`, `os` and `uv-version`.
+- The `skypilot` backend's copy transport, `[tool.substrax.compute.backends.skypilot.copy]`, for
+  providers that cannot mount buckets, such as RunPod and Vast. SkyPilot copies each mount's
+  local source onto the cluster, read-only, and the worker runs under bubblewrap with the system,
+  the project and the inputs read-only (`network_isolation = true` also takes its network away).
+  `fetch` copies the outputs over rsync and verifies them against manifests of the remote
+  directory taken before and after the copy; the cluster stays up until
+  `SkyPilotBackend.teardown(handle)` or its `lifetime_minutes` timer removes it.
 
 ### Changed
 
+- **Breaking:** a run handle records the backend settings it was submitted with
+  (`RunHandle.settings`, required), and `substrax-compute` rebuilds the backend from them, never
+  from the project's current configuration, so editing a project no longer redirects its earlier
+  runs. A backend passes its validated settings to `new_run(spec, project, settings=...)`.
+  Handles are of version 2, and a version-1 handle is refused, since it holds no settings to
+  rebuild its backend from: remove the records of runs submitted before this release from the
+  state directory's `compute/runs/`. `substrax-compute status` lists every run it can read,
+  names each record it cannot with the reason, and exits 1. `RunStore.handles()` is replaced by
+  `RunStore.read_all()`, which returns both.
+- **Breaking:** the `skypilot` backend runs SkyPilot through its public SDK, in SkyPilot's own
+  interpreter (`sky_python`, by default the `python` beside the `sky` command), and reads its
+  native request and queue records instead of the `sky` command's output. A launch records its
+  intent and SkyPilot's request id before it waits, so an interrupted one is reconciled rather
+  than submitted again. `cancel` cancels that request and the job and tears the cluster down,
+  attempting every step and raising an `ExceptionGroup` of the failures, which
+  `substrax-compute cancel` reports with exit code 1; a run that already ended is left as it is,
+  and `teardown(handle)` removes its cluster. A job status the backend does not know raises
+  instead of reading as failed; `FAILED_DRIVER` reads as failed. The transport's settings move
+  into a table of their own: `outputs_bucket` and `buckets` go under
+  `[tool.substrax.compute.backends.skypilot.bucket]`, and a backend configures exactly one of
+  `bucket` and `copy`.
+
 - substrax's CI is one workflow, `ci.yml`. It calls `python-gates.yml`, adds the Orbax-floor
   checkpoint tests and the derived-status check, and ends in the `required` check. The Quality
   Checks, Security, Documentation and Build Verification workflows are folded into it, so a pull
@@ -76,6 +105,10 @@ and uses semantic versioning while the public API stabilizes.
 
 ### Fixed
 
+- A compute task runs in its own process group, and the worker kills the whole group when the
+  task ends, times out or is cancelled, so a descendant no longer outlives its task holding its
+  log pipes open. The worker turns SIGTERM and SIGINT into that cleanup, so cancelling a local
+  run stops its task tree, a task that ignores SIGTERM included.
 - The lockfile resolves the fixed releases of nine packages pip-audit reports: aiohttp 3.14.5,
   certifi 2026.10.23, filelock 4.2.4, pillow 13.0.0 and virtualenv 21.14.13, and gitpython 3.2.1,
   mlflow 3.17.2, opentelemetry-api 1.45.1 and werkzeug 3.1.9, which only the `mlflow` and `wandb`

docs/api/compute.md

@@ -76,7 +76,7 @@ itself never starts a JAX backend.
 | --- | --- | --- |
 | `local` | this machine, in the project's current environment | nothing |
 | `modal` | Modal, one detached function per run, outputs on a Volume | `substrax[modal]`, a Modal account, and a workspace image builder of 2025.06 or later |
-| `skypilot` | a cluster SkyPilot provisions for the run on any cloud it supports, torn down after it; outputs on a `gs://` or `s3://` bucket | SkyPilot installed as its documentation says (`uv tool install --with pip "skypilot[gcp]"`), and `gcloud` or `aws` for the bucket |
+| `skypilot` | a cluster SkyPilot provisions for the run on any cloud it supports; inputs and outputs on buckets, or copied | SkyPilot installed as its documentation says (`uv tool install --with pip "skypilot[gcp]"`, or the provider's extra), and the transport's tools: `gcloud` or `aws` for buckets, `ssh` and `rsync` for copies |
 
 The Modal image is built in layers keyed on what they depend on: the locked dependencies
 (`Image.uv_sync`), then the project's source, then the project installed without
@@ -85,9 +85,67 @@ builder 2025.06 or later, set for the workspace in Modal's dashboard under Setti
 config; an older builder fails the build with `Image builder version <= 2024.10 requires modal
 to be specified in your pyproject.toml file`. The image runs the Python this interpreter runs,
 which the serialized worker requires, and which `uv sync` is told to use so a project's own
-`python-preference` cannot choose another. SkyPilot is never imported:
-the backend writes a task file and runs `sky launch`, `sky queue -o json`, `sky logs`,
-`sky cancel` and `sky down`.
+`python-preference` cannot choose another.
+
+### SkyPilot
+
+SkyPilot is never imported into the project's environment. Every operation (launch, queue, logs,
+cancel, teardown) runs `substrax/compute/sky_sdk.py` in SkyPilot's own interpreter, which calls
+SkyPilot's public SDK (0.11.1) and reads its native request and queue records. Before a launch
+waits for provisioning, it writes the run's intent and SkyPilot's request id beside the task
+file; an interrupted launch is reconciled from them, not submitted again. `cancel` leaves a run
+that already ended as it is; otherwise it cancels the request and the job and tears the cluster
+down, attempting every step and reporting each failure (`substrax-compute cancel` exits 1).
+
+The settings under `[tool.substrax.compute.backends.skypilot]`, all optional:
+
+| Setting | Default | Meaning |
+| --- | --- | --- |
+| `infra` | any enabled cloud | where to run, in SkyPilot's `infra` syntax: `gcp`, `gcp/us-central1`, `runpod` |
+| `use_spot` | `false` | run on spot instances |
+| `setup_commands` | `[]` | commands that prepare the image before the locked environment is synced |
+| `sky` | `"sky"` | the `sky` command, used to find SkyPilot's interpreter beside it |
+| `sky_python` | the `python` beside `sky` | SkyPilot's interpreter |
+| `worker_python` | `"uv run --no-sync python"` | the command that starts Python in the synced environment |
+| `state_dir` | the default state directory | where task files, launch records and cancellations are kept |
+
+A run's files move one of two ways, configured as exactly one of these tables.
+
+**Bucket transport** (`[...skypilot.bucket]`): SkyPilot mounts the buckets, so the outputs
+outlive the cluster, which SkyPilot removes when the job ends. Providers that cannot mount
+buckets (RunPod, Vast) are refused.
+
+```toml
+[tool.substrax.compute.backends.skypilot.bucket]
+outputs_bucket = "gs://avitai-runs"                # required; gs:// or s3://
+buckets = { datasets = "gs://avitai-datasets" }    # each mount's bucket, by mount name
+```
+
+**Copy transport** (`[...skypilot.copy]`): SkyPilot copies each mount's local source onto the
+cluster, read-only; a writable mount is refused, since a copy cannot be written back. The worker
+runs under bubblewrap, so the image has to provide `bwrap` with working user namespaces: the
+system, the project and the inputs are read-only, `/tmp` and the outputs writable. `fetch`
+copies the outputs over `rsync` and checks them against manifests of the remote directory taken
+before and after; a difference keeps the cluster up. Fetch the outputs before
+`SkyPilotBackend.teardown(handle)`, which removes the cluster, and with it the outputs on it.
+`lifetime_minutes` starts once setup ends; it is not a provisioning deadline or a cost cap.
+
+```toml
+[tool.substrax.compute.backends.skypilot.copy]
+sources = { datasets = "/data/datasets" }   # each mount's local path, by mount name
+disk_size = 60                              # GB
+lifetime_minutes = 60                       # idle time after setup before the cluster removes itself
+network_isolation = false                   # true runs the worker without network access
+```
+
+## Run handles
+
+A run's handle records the backend settings it was submitted with (`RunHandle.settings`), and
+`status`, `logs`, `fetch` and `cancel` rebuild the backend from them, never from the project's
+current configuration; a backend passes them to `new_run(spec, project, settings=...)`. Handles
+are of version 2. One of another version is refused: `substrax-compute status RUN_ID` exits 2,
+and `substrax-compute status` lists every run it can read, names each record it cannot with the
+reason, and exits 1.
 
 ## Adding a provider
 
@@ -130,3 +188,5 @@ class TestMyBackend(BackendContract):
 ::: substrax.compute.modal_backend
 
 ::: substrax.compute.skypilot_backend
+
+::: substrax.compute.skypilot_transports

src/substrax/compute/backend.py

@@ -23,7 +23,7 @@
 from substrax.typing import JsonValue
 
 
-HANDLE_VERSION = 1
+HANDLE_VERSION = 2
 SPEC_NAME = "spec.json"
 """The file a staged run's spec is written to."""
 STATE_DIR_ENV = "SUBSTRAX_STATE_DIR"
@@ -56,6 +56,7 @@ class RunHandle:
         job: The job's name.
         project: The project root it was submitted from.
         submitted_at: When it was submitted, in UTC.
+        settings: Frozen provider configuration used for every recovery operation.
         details: The provider's own references to the run, such as an app or cluster id.
     """
 
@@ -67,6 +68,7 @@ class RunHandle:
     job: str
     project: str
     submitted_at: datetime
+    settings: Mapping[str, JsonValue]
     details: Mapping[str, str] = field(default_factory=dict)
 
 
@@ -78,11 +80,13 @@ class NewRun:
         run_id: The run's id, from :func:`new_run_id`.
         spec: The job.
         project: The resolved project root.
+        settings: Frozen provider configuration used to construct the run handle.
     """
 
     run_id: str
     spec: JobSpec
     project: Path
+    settings: Mapping[str, JsonValue]
 
     def handle(self, backend: str, details: Mapping[str, str]) -> RunHandle:
         """The handle of this run, submitted now.
@@ -100,6 +104,7 @@ def handle(self, backend: str, details: Mapping[str, str]) -> RunHandle:
             job=self.spec.name,
             project=str(self.project),
             submitted_at=datetime.now(UTC),
+            settings=self.settings,
             details=details,
         )
 
@@ -181,17 +186,23 @@ def backend_state_dir(state_dir: str | None, backend: str) -> Path:
     return root / "compute" / backend
 
 
-def new_run(spec: JobSpec, project: Path) -> NewRun:
+def new_run(spec: JobSpec, project: Path, *, settings: Mapping[str, JsonValue]) -> NewRun:
     """Start submitting a run of ``spec`` from ``project``.
 
     Args:
         spec: The job.
         project: The project root.
+        settings: Validated provider settings to freeze into the handle.
 
     Returns:
         The run, named by :func:`new_run_id`.
     """
-    return NewRun(run_id=new_run_id(spec.name), spec=spec, project=project.resolve())
+    return NewRun(
+        run_id=new_run_id(spec.name),
+        spec=spec,
+        project=project.resolve(),
+        settings=json.loads(json.dumps(settings)),
+    )
 
 
 def stage_run(runs: Path, run: NewRun) -> Path:
@@ -248,21 +259,60 @@ def load(self, run_id: str) -> RunHandle:
         if not path.is_file():
             msg = f"no run {run_id!r} in {self.root}"
             raise KeyError(msg)
-        return read_record(RunHandle, json.loads(path.read_text(encoding="utf-8")))
+        return _read_handle(path)
 
-    def handles(self) -> list[RunHandle]:
-        """Every kept handle, newest first.
+    def read_all(self) -> StoredRuns:
+        """Every kept handle, newest first, and each record that cannot be read, with the reason.
+
+        A record an older substrax wrote, or a damaged one, is named on its own rather than
+        hiding the runs that can still be read.
 
         Returns:
-            The handles.
+            The readable handles and the unreadable records.
         """
-        if not self.root.is_dir():
-            return []
-        found = [
-            read_record(RunHandle, json.loads(path.read_text(encoding="utf-8")))
-            for path in self.root.glob("*.json")
-        ]
-        return sorted(found, key=lambda handle: handle.submitted_at, reverse=True)
+        handles: list[RunHandle] = []
+        unreadable: dict[Path, str] = {}
+        for path in sorted(self.root.glob("*.json")) if self.root.is_dir() else []:
+            try:
+                handles.append(_read_handle(path))
+            except ValueError as error:
+                unreadable[path] = str(error)
+        handles.sort(key=lambda handle: handle.submitted_at, reverse=True)
+        return StoredRuns(handles=tuple(handles), unreadable=unreadable)
 
     def _path(self, run_id: str) -> Path:
         return self.root / f"{run_id}.json"
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class StoredRuns:
+    """What a :class:`RunStore` holds: the handles it can read, and why it cannot read the rest.
+
+    Attributes:
+        handles: The readable handles, newest first.
+        unreadable: Each record that cannot be read, by path, with the reason.
+    """
+
+    handles: tuple[RunHandle, ...]
+    unreadable: Mapping[Path, str]
+
+
+def _read_handle(path: Path) -> RunHandle:
+    """Read one stored handle; one of another handle version is refused, never reinterpreted.
+
+    Args:
+        path: The record.
+
+    Returns:
+        The handle.
+
+    Raises:
+        ValueError: If the record is of another handle version; its JSON and its fields raise
+            ``ValueError`` subclasses too (``json.JSONDecodeError``, pydantic's validation error).
+    """
+    data = json.loads(path.read_text(encoding="utf-8"))
+    version = data.get("schema_version") if isinstance(data, dict) else None
+    if version != HANDLE_VERSION:
+        msg = f"run handle version {version} cannot be read; this substrax reads {HANDLE_VERSION}"
+        raise ValueError(msg)
+    return read_record(RunHandle, data)

src/substrax/compute/cli.py

@@ -12,7 +12,8 @@
 ``run`` submits the job, follows its log, waits for it and fetches its outputs; with ``--detach``
 it returns once the job is submitted, and the other commands find the run again by the id it
 prints. Fetched outputs go to ``--into``, or under the state directory, never into the project.
-Exit codes: 0 for success, 1 for a run that did not succeed, 2 for a refused command.
+Exit codes: 0 for success, 1 for unsuccessful execution or cleanup, or a stored run ``status``
+cannot read, 2 for a refused command.
 """
 
 from __future__ import annotations
@@ -28,9 +29,9 @@
     RunHandle,
     RunState,
     RunStore,
+    StoredRuns,
 )
 from substrax.compute.config import (
-    ComputeConfig,
     parse_accelerator,
     read_compute_config,
     require_locked_substrax,
@@ -55,12 +56,23 @@ def main(argv: Sequence[str] | None = None) -> int:
     store = RunStore()
     try:
         return arguments.command(arguments, store)
+    except ExceptionGroup as error:
+        for message in _failure_messages(error):
+            sys.stderr.write(f"substrax-compute: {message}\n")
+        return _UNSUCCESSFUL
     except (LookupError, FileNotFoundError, ValueError) as error:
         message = error.args[0] if isinstance(error, KeyError) and error.args else str(error)
         sys.stderr.write(f"substrax-compute: {message}\n")
         return _REFUSED
 
 
+def _failure_messages(error: Exception) -> list[str]:
+    """Retain each cleanup cause without an unhandled exception traceback."""
+    if isinstance(error, ExceptionGroup):
+        return [message for child in error.exceptions for message in _failure_messages(child)]
+    return [str(error)]
+
+
 def _run(arguments: argparse.Namespace, store: RunStore) -> int:
     project = arguments.project.resolve()
     config = read_compute_config(project)
@@ -83,11 +95,16 @@ def _run(arguments: argparse.Namespace, store: RunStore) -> int:
 
 
 def _status(arguments: argparse.Namespace, store: RunStore) -> int:
-    handles = [store.load(arguments.run_id)] if arguments.run_id else store.handles()
-    for handle in handles:
+    if arguments.run_id:
+        stored = StoredRuns(handles=(store.load(arguments.run_id),), unreadable={})
+    else:
+        stored = store.read_all()
+    for handle in stored.handles:
         state = _backend_for(handle).status(handle)
         sys.stdout.write(f"{handle.run_id}  {handle.backend}  {handle.job}  {state}\n")
-    return _SUCCESS
+    for path, reason in stored.unreadable.items():
+        sys.stderr.write(f"substrax-compute: {path}: {reason}\n")
+    return _UNSUCCESSFUL if stored.unreadable else _SUCCESS
 
 
 def _logs(arguments: argparse.Namespace, store: RunStore) -> int:
@@ -119,12 +136,8 @@ def _finish(backend: ComputeBackend, handle: RunHandle, into: Path | None) -> in
 
 
 def _backend_for(handle: RunHandle) -> ComputeBackend:
-    """Rebuild the backend of a stored run with the settings its project configures now."""
-    project = Path(handle.project)
-    config = (
-        read_compute_config(project) if (project / "pyproject.toml").is_file() else ComputeConfig()
-    )
-    return load_backend(handle.backend, config.backends.get(handle.backend, {}))
+    """Recover every provider from the settings bound to this run."""
+    return load_backend(handle.backend, handle.settings)
 
 
 def _destination(into: Path | None) -> Path: