diff --git a/docs/launchers.md b/docs/launchers.md index 658b252a..676e56b2 100644 --- a/docs/launchers.md +++ b/docs/launchers.md @@ -654,6 +654,44 @@ madengine run --manifest-file build_manifest.json **Alias**: `"slurm-multi"` (hyphen) is normalized to `"slurm_multi"` (underscore). +**Model-card contract (shared with the templated path)**: + +slurm_multi hands the *topology* to the workload, not the *contract*. These model-card +fields are honoured identically on both paths: + +| Field | Effect | +|-------|--------| +| `distributed.launcher` | Selects the path. Resolved deployment-config-first, model-card-second — the same resolution used to emit the launcher env block, so the two can never disagree. | +| `distributed.nnodes` | Sizes the allocation (`#SBATCH --nodes` / `--ntasks`) when `slurm.nodes` is not set explicitly. If both are set and differ, `slurm.nodes` wins and a warning is printed. | +| `slurm.*` | `partition`, `time`, `gpus_per_node`, `exclusive`, `reservation`, `nodelist`, … | +| `multiple_results` | Names the results CSV. Searched next to the model script (where the wrapper `cd`s, so `$(pwd)` writes land there), then the job dir, output dir and cwd, before falling back to the conventional locations. | + +> **`args` is not sbatch flags.** On the templated path `args` goes to the +> containerised `run.sh`; on this path it is appended to `bash .slurm`. It is +> **not** forwarded to `sbatch`, so `"args": "-N 4 -n 4"` does *not* request 4 nodes — +> it passes two ignored positional arguments to a script that never reads `$@`, and +> the job runs on the `slurm.nodes` default of 1. Size the allocation with +> `distributed.nnodes` (or `slurm.nodes`) instead. + +> **`multiple_results` is read directly here.** On the templated path the CSV is +> narrow and gets merged with `common_info` by `update_perf_csv`. A self-managed +> script has no `common_info` to merge against, so it writes the full perf schema +> itself and madengine reads it as-is — routing it through `handle_multiple_results()` +> would recompute `status` from `performance` and flip a legitimate zero-score +> FAILURE row to SUCCESS. + +> **Preset selection still keys on `slurm.nodes`.** `ConfigLoader` picks the +> single-node vs multi-node preset before `nnodes` reconciliation, so a card sized +> only by `nnodes` keeps the single-node preset. That is usually what RDMA inference +> workloads want — the multi-node preset sets `NCCL_IB_DISABLE=1` and +> `NCCL_SOCKET_IFNAME=eth0` — but set `slurm.time` explicitly if the 12 h single-node +> default is too short. + +**Placeholder images are rejected**: model cards commonly ship +`"DOCKER_IMAGE_NAME": ""` as a fill-me-in marker. Any +angle-bracketed value is rejected at submit time with an actionable error rather than +becoming the image every compute node fails to pull. + **Features**: - Wrapper SBATCH script with shell-quoted env_vars (injection-safe) - Parallel `srun docker pull` on all nodes for registry images diff --git a/src/madengine/deployment/common.py b/src/madengine/deployment/common.py index 13657246..c2fd605d 100644 --- a/src/madengine/deployment/common.py +++ b/src/madengine/deployment/common.py @@ -125,6 +125,75 @@ def is_self_managed_launcher(launcher_type: Optional[str]) -> bool: return normalize_launcher(launcher_type, "slurm") in _SELF_MANAGED_LAUNCHERS +def resolve_launcher_from_sources( + deployment_launcher: Optional[str], + model_launcher: Optional[str], + default: str = "torchrun", +) -> str: + """Resolve the effective launcher from the deployment config and the model card. + + Precedence is deployment config first, model card second. That matches how + BuildOrchestrator builds the manifest: it copies the model card's launcher into + ``deployment_config.distributed`` only when the key is absent, so a value present + there is already the user's explicit choice. + + Every dispatch site must use this helper. Resolving the launcher differently in + two places is what allowed ``prepare()`` to pick the self-managed path from the + model card while ``_prepare_template_context()`` simultaneously emitted a + ``torchrun`` env block from the deployment config. + """ + return deployment_launcher or model_launcher or default + + +def resolve_node_count( + configured_nodes: int, + nnodes: Optional[Any], + nodes_explicitly_set: bool, +) -> tuple: + """Reconcile ``slurm.nodes`` with ``distributed.nnodes`` into one node count. + + ``slurm.nodes`` sizes the allocation (``#SBATCH --nodes``) while + ``distributed.nnodes`` sizes the topology the launcher builds on top of it. + Nothing used to reconcile them, so a model card declaring only ``nnodes`` was + submitted against the ``slurm.nodes`` default of 1 — a multi-node topology + squeezed onto a single node. + + Args: + configured_nodes: ``slurm.nodes`` after ConfigLoader applied its defaults. + nnodes: ``distributed.nnodes``, if declared. + nodes_explicitly_set: True when ``slurm.nodes`` came from the user or the + model card rather than from a preset default. + + Returns: + ``(nodes, note)`` where ``note`` is a human-readable string to log, or None. + """ + try: + requested = int(nnodes) if nnodes is not None else None + except (TypeError, ValueError): + return configured_nodes, ( + f"Ignoring non-numeric distributed.nnodes={nnodes!r}; " + f"using slurm.nodes={configured_nodes}." + ) + + if requested is None or requested == configured_nodes: + return configured_nodes, None + + if nodes_explicitly_set: + # Both were set and they disagree. The explicit allocation size wins — + # overriding it could request nodes the user did not ask to be billed for — + # but the mismatch is almost always a config error, so say so. + return configured_nodes, ( + f"slurm.nodes={configured_nodes} conflicts with " + f"distributed.nnodes={requested}; using slurm.nodes={configured_nodes}. " + "Set them to the same value to silence this." + ) + + return requested, ( + f"Sizing allocation from distributed.nnodes={requested} " + f"(slurm.nodes was not set explicitly)." + ) + + @functools.lru_cache(maxsize=None) def is_rocprofv3_available() -> bool: """ diff --git a/src/madengine/deployment/slurm.py b/src/madengine/deployment/slurm.py index 088a3fb2..24ee3e93 100644 --- a/src/madengine/deployment/slurm.py +++ b/src/madengine/deployment/slurm.py @@ -26,6 +26,8 @@ configure_multi_node_profiling, is_self_managed_launcher, normalize_launcher, + resolve_launcher_from_sources, + resolve_node_count, ) from .config_loader import ConfigLoader, apply_deployment_config from .slurm_node_selector import SlurmNodeSelector @@ -64,6 +66,13 @@ def __init__(self, config: DeploymentConfig): Args: config: Deployment configuration """ + # Capture which slurm keys were actually supplied (by --additional-context or + # by the model card, which BuildOrchestrator merges into the manifest's + # deployment_config) BEFORE ConfigLoader layers its presets on top. Once the + # defaults are applied every key looks "set", and nodes=1 from a preset is + # indistinguishable from nodes=1 the user asked for. + self._explicit_slurm_keys = set((config.additional_context or {}).get("slurm") or {}) + apply_deployment_config(config, ConfigLoader.load_slurm_config) super().__init__(config) @@ -74,6 +83,12 @@ def __init__(self, config: DeploymentConfig): # SLURM parameters self.partition = self.slurm_config.get("partition", "gpu") self.nodes = self.slurm_config.get("nodes", 1) + + # Baseline allocation size, before any distributed.nnodes reconciliation. + # _resolve_nodes() always recomputes from this so that prepare(), which + # deploy() re-runs after node preflight, stays idempotent. + self._configured_nodes = self.nodes + self._resolve_nodes() self.gpus_per_node = self.slurm_config.get("gpus_per_node", 8) self.time_limit = self.slurm_config.get("time", "24:00:00") self.output_dir = Path(self.slurm_config.get("output_dir", "./slurm_results")) @@ -323,11 +338,10 @@ def prepare(self) -> bool: model_keys_peek = list((self.manifest or {}).get("built_models", {}).keys()) if model_keys_peek: model_info_peek = self.manifest["built_models"][model_keys_peek[0]] - model_distributed_peek = model_info_peek.get("distributed", {}) - launcher_type_peek = ( - model_distributed_peek.get("launcher") - or self.distributed_config.get("launcher", "torchrun") - ) + # Re-resolve now that the model card is in hand: a card may size its + # topology with distributed.nnodes alone. + self._resolve_nodes(model_info_peek) + launcher_type_peek = self._resolve_launcher(model_info_peek) if is_self_managed_launcher(launcher_type_peek): self.output_dir.mkdir(parents=True, exist_ok=True) self.console.print( @@ -380,6 +394,46 @@ def prepare(self) -> bool: self.console.print(f"[red]✗ Failed to generate script: {e}[/red]") return False + def _resolve_nodes(self, model_info: Optional[Dict] = None) -> int: + """Size the allocation from slurm.nodes reconciled with distributed.nnodes. + + ``nnodes`` is read from the deployment config first (BuildOrchestrator copies + the model card's value there at build time) and from the model card second, so + a card is honoured even when the manifest was not produced by that merge — for + example a hand-written manifest, or `madengine run` against a card whose + topology changed after the build. + + Recomputes from ``self._configured_nodes`` rather than from ``self.nodes``, so + repeated calls converge instead of ratcheting. + """ + nnodes = self.distributed_config.get("nnodes") + if nnodes is None and model_info: + nnodes = (model_info.get("distributed") or {}).get("nnodes") + + resolved, note = resolve_node_count( + configured_nodes=self._configured_nodes, + nnodes=nnodes, + nodes_explicitly_set="nodes" in self._explicit_slurm_keys, + ) + if note and note != getattr(self, "_nodes_note_shown", None): + self.console.print(f"[yellow]⚠ {note}[/yellow]") + self._nodes_note_shown = note + + self.nodes = resolved + self.slurm_config["nodes"] = resolved + return resolved + + def _resolve_launcher(self, model_info: Dict) -> str: + """Resolve the effective launcher for a model, deployment config first. + + Single source of truth for both dispatch sites: the self-managed peek in + prepare() and the launcher-command generation in _prepare_template_context(). + """ + return resolve_launcher_from_sources( + deployment_launcher=self.distributed_config.get("launcher"), + model_launcher=(model_info.get("distributed") or {}).get("launcher"), + ) + @staticmethod def _normalize_nodelist(nodelist: Optional[str]) -> Optional[str]: """Normalize nodelist to comma-separated without spaces for #SBATCH --nodelist.""" @@ -631,9 +685,11 @@ def _prepare_template_context(self, model_info: Dict) -> Dict[str, Any]: additional_context["slurm"] = self.slurm_config resolved_gpus_per_node = resolve_runtime_gpus(model_info, additional_context) - # Extract launcher configuration - launcher_type = self.distributed_config.get("launcher", "torchrun") # Default to torchrun - + # Extract launcher configuration. Resolved the same way as the self-managed + # peek in prepare(), so the path taken and the env block emitted can never + # disagree about which launcher this model uses. + launcher_type = self._resolve_launcher(model_info) + # Canonicalize aliases before validity check so e.g. sglang_disagg → sglang-disagg # passes through normalize_launcher instead of being mapped to "docker". launcher_type = canonicalize_distributed_launcher(launcher_type) or launcher_type @@ -1767,10 +1823,9 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: # so collect via _collect_slurm_multi_results instead of the template-based path. if model_key: _mi = built_models_dict.get(model_key, {}) or {} - _launcher_type = (_mi.get("distributed") or {}).get("launcher", "") - if is_self_managed_launcher(_launcher_type): + if is_self_managed_launcher(self._resolve_launcher(_mi)): return self._collect_slurm_multi_results( - deployment_id, results, session_start_row + deployment_id, results, session_start_row, model_info=_mi ) @@ -2094,12 +2149,66 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: ) return results + def _slurm_multi_declared_result_csv( + self, model_info: Optional[Dict[str, Any]], deployment_id: str + ) -> Optional[Path]: + """Resolve a slurm_multi model's declared ``multiple_results`` CSV. + + On the templated path ``multiple_results`` names a narrow per-run CSV that is + merged with common_info by update_perf_csv. A self-managed script has no + common_info to merge against, so it writes the full perf schema itself and the + file is read directly — but the model card should still be able to *name* it + rather than every workload having to land on one of the hardcoded paths below. + + Note the difference is deliberate: routing an already-full-schema CSV through + handle_multiple_results() would recompute ``status`` from ``performance`` and + turn a legitimate zero-score FAILURE row into a SUCCESS. + """ + declared = (model_info or {}).get("multiple_results") + if not declared: + return None + + search_dirs: List[Path] = [] + # Where the launcher runs: the wrapper cd's to the model script's directory, + # so a script writing to $(pwd) lands here. + scripts_rel = (model_info or {}).get("scripts", "") + if scripts_rel and self.config.manifest_file: + script_path = Path(self.config.manifest_file).parent.absolute() / scripts_rel + search_dirs.append(script_path.parent) + search_dirs.extend( + [ + self.output_dir / deployment_id, + self.output_dir, + Path.cwd(), + ] + ) + + for directory in search_dirs: + candidate = directory / declared + try: + if candidate.is_file() and candidate.stat().st_size > 0: + self.console.print( + f"[dim] Using declared multiple_results CSV: {candidate}[/dim]" + ) + return candidate + except OSError: + continue + self.console.print( + f"[dim] Declared multiple_results '{declared}' not found in " + f"{', '.join(str(d) for d in search_dirs)}; falling back to conventional paths.[/dim]" + ) + return None + def _collect_slurm_multi_results( - self, deployment_id: str, results: Dict[str, Any], session_start_row: Optional[int] + self, + deployment_id: str, + results: Dict[str, Any], + session_start_row: Optional[int], + model_info: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ Collect results for slurm_multi launchers. - + slurm_multi model scripts generate their own perf.csv via their benchmark scripts (e.g. generate_perf_csv.py). We collect SLURM logs for diagnostics and read the model-generated perf.csv for metrics. @@ -2108,15 +2217,19 @@ def _collect_slurm_multi_results( flat_out_files = sorted(self.output_dir.glob(f"madengine-*_{deployment_id}_*.out")) results["logs"] = [str(f) for f in flat_out_files] + # A model card that names its own results CSV wins over the conventional + # locations below, so a workload is not forced to write to a path madengine + # happens to know about. + perf_csv_path = self._slurm_multi_declared_result_csv(model_info, deployment_id) + # Look for model-generated perf.csv. Inner scripts in MAD-private write # to one of these locations depending on the workload: # * SGLang / vLLM disagg: /shared_inference///perf.csv # * Large EP / KV cache: /slurm_output/perf_csv/**.csv # Plus the legacy /perf.csv path some flows still use. - # Priority: results_dir config > shared_inference NFS > slurm_output/perf_csv - # > /perf.csv (with NFS-propagation retry). - perf_csv_path = None - if self.slurm_config.get("results_dir"): + # Priority: declared multiple_results > results_dir config > shared_inference + # NFS > slurm_output/perf_csv > /perf.csv (with NFS-propagation retry). + if not perf_csv_path and self.slurm_config.get("results_dir"): results_dir = Path(self.slurm_config["results_dir"]) candidates = list(results_dir.glob("perf*.csv")) if candidates: @@ -2165,16 +2278,25 @@ def _collect_slurm_multi_results( import shutil cwd_perf = Path("perf.csv") try: - if cwd_perf.exists(): - with open(perf_csv_path, "r") as src, open(cwd_perf, "a") as dst: - next(src, None) # skip per-job header so cwd CSV stays single-headed - for line in src: - dst.write(line) + # The source can legitimately resolve to the cwd perf.csv itself — + # via the /perf.csv fallback below, or a model card declaring + # multiple_results: "perf.csv". Appending a file to itself would + # duplicate every row, so there is nothing to aggregate. + if cwd_perf.exists() and perf_csv_path.resolve() == cwd_perf.resolve(): + self.console.print( + "[dim]Per-job perf is already the cwd perf.csv; nothing to aggregate[/dim]" + ) else: - shutil.copy(str(perf_csv_path), str(cwd_perf)) - self.console.print( - f"[green]✓ Aggregated per-job perf into {cwd_perf}[/green]" - ) + if cwd_perf.exists(): + with open(perf_csv_path, "r") as src, open(cwd_perf, "a") as dst: + next(src, None) # skip per-job header so cwd CSV stays single-headed + for line in src: + dst.write(line) + else: + shutil.copy(str(perf_csv_path), str(cwd_perf)) + self.console.print( + f"[green]✓ Aggregated per-job perf into {cwd_perf}[/green]" + ) except Exception as e: self.console.print( f"[yellow]⚠ Could not aggregate per-job perf into cwd perf.csv: {e}[/yellow]" diff --git a/src/madengine/orchestration/build_orchestrator.py b/src/madengine/orchestration/build_orchestrator.py index 17e836e8..0a00d037 100644 --- a/src/madengine/orchestration/build_orchestrator.py +++ b/src/madengine/orchestration/build_orchestrator.py @@ -282,6 +282,13 @@ def execute( } if len(card_images) == 1: implicit_image = next(iter(card_images)) + # Reject ""-style markers here rather than + # letting them become the image name every compute node tries + # to pull. + self._reject_placeholder_image( + implicit_image, + [m.get("name", "unknown") for m in slurm_multi_models], + ) self.rich_console.print( f"[dim]slurm_multi: no --registry/--use-image given; " f"using DOCKER_IMAGE_NAME from model card -> {implicit_image}[/dim]" @@ -665,6 +672,37 @@ def _execute_with_prebuilt_image( ), ) from e + # Model cards ship DOCKER_IMAGE_NAME as an angle-bracketed placeholder + # (e.g. "") to mark "you must supply this". A placeholder is + # not a usable image reference, and silently accepting one produces a confusing + # `docker pull ` failure on every compute node instead of an + # actionable message at submit time. + @staticmethod + def _is_placeholder_image(image: Optional[str]) -> bool: + """Return True if the value is a fill-me-in marker rather than an image ref.""" + if not image: + return True + candidate = image.strip() + return candidate.startswith("<") or candidate.endswith(">") + + def _reject_placeholder_image(self, image: str, model_names: List[str]) -> None: + """Raise a ConfigurationError if the resolved image is a placeholder.""" + if not self._is_placeholder_image(image): + return + raise ConfigurationError( + f"Model card DOCKER_IMAGE_NAME is a placeholder, not an image: {image!r}", + context=create_error_context( + operation="resolve_image", + component="BuildOrchestrator", + additional_info={"image": image, "model_names": model_names}, + ), + suggestions=[ + "Pass the real image explicitly: --use-image /:", + "Or build and push from the model's dockerfile: --registry ", + "Or replace DOCKER_IMAGE_NAME in the model card env_vars with a real image", + ], + ) + def _resolve_image_from_model_card(self) -> str: """ Resolve Docker image name from model card's DOCKER_IMAGE_NAME env var. @@ -742,7 +780,9 @@ def _resolve_image_from_model_card(self) -> str: ) else: self.rich_console.print(f"[green]✓ Auto-detected image: {resolved_image}[/green]\n") - + + self._reject_placeholder_image(resolved_image, list(images_found)) + return resolved_image def _execute_build_on_compute( diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index 498401c4..9f9ab445 100644 --- a/src/madengine/orchestration/run_orchestrator.py +++ b/src/madengine/orchestration/run_orchestrator.py @@ -31,6 +31,7 @@ ExecutionError, create_error_context, ) +from madengine.deployment.common import is_self_managed_launcher from madengine.utils.session_tracker import SessionTracker from madengine.orchestration.image_filtering import ( filter_images_by_gpu_compatibility as _filter_by_gpu_compat, @@ -1113,19 +1114,25 @@ def _infer_deployment_target(self, config: Dict) -> str: Convention over Configuration: - Presence of "k8s" or "kubernetes" field → k8s deployment - Presence of "slurm" field → slurm deployment - - Neither present → local execution - + - A self-managed SLURM launcher (slurm_multi) → slurm deployment + - None of the above → local execution + Args: config: Configuration dictionary - + Returns: Deployment target: "k8s", "slurm", or "local" """ if "k8s" in config or "kubernetes" in config: return "k8s" - elif "slurm" in config: + if "slurm" in config: return "slurm" - else: - return "local" + # slurm_multi runs the model's own .slurm script through sbatch/srun, so it is + # a SLURM deployment by construction. Without this a model card that declared + # the launcher but no `slurm` block inferred "local" and was handed to the + # container runner, which never reaches the slurm_multi path at all. + if is_self_managed_launcher((config.get("distributed") or {}).get("launcher")): + return "slurm" + return "local" diff --git a/tests/unit/test_orchestration.py b/tests/unit/test_orchestration.py index aaf24fcc..b20b1d04 100644 --- a/tests/unit/test_orchestration.py +++ b/tests/unit/test_orchestration.py @@ -446,3 +446,67 @@ def test_multiple_results_defaults_to_empty_string(self, mock_context, tmp_path) built_model = next(iter(manifest["built_models"].values())) assert built_model["multiple_results"] == "" + + +class TestPlaceholderImageRejection: + """Model cards ship DOCKER_IMAGE_NAME as a "" marker. + + The implicit --use-image path used to accept any single distinct card value, so + the placeholder became the image name and every compute node failed on + `docker pull ` instead of the user getting told at submit time. + """ + + @pytest.mark.parametrize("value", [ + "", + "", + " ", + "", + None, + ]) + def test_placeholders_detected(self, value): + assert BuildOrchestrator._is_placeholder_image(value) is True + + @pytest.mark.parametrize("value", [ + "rocm/vllm:latest", + "docker.io/myorg/img:tag", + "ci-pyt_vllm_kimi_k3_mi300x", + "localhost:5000/img", + ]) + def test_real_images_accepted(self, value): + assert BuildOrchestrator._is_placeholder_image(value) is False + + def test_reject_raises_configuration_error(self): + orchestrator = BuildOrchestrator.__new__(BuildOrchestrator) + with pytest.raises(ConfigurationError) as exc: + orchestrator._reject_placeholder_image("", ["m1"]) + assert "placeholder" in str(exc.value).lower() + + def test_reject_passes_through_real_image(self): + orchestrator = BuildOrchestrator.__new__(BuildOrchestrator) + orchestrator._reject_placeholder_image("rocm/vllm:latest", ["m1"]) + + +class TestSelfManagedLauncherImpliesSlurm: + """A slurm_multi model card without a `slurm` block still deploys to SLURM. + + Target inference keys on the presence of a `slurm`/`k8s` block. Model cards + routinely declare only `distributed.launcher: slurm_multi`, which inferred + "local" and handed the job to the container runner — so the slurm_multi path + was never reached and the model's .slurm script was run as a local workload. + """ + + @pytest.mark.parametrize("config,expected", [ + ({}, "local"), + ({"slurm": {}}, "slurm"), + ({"k8s": {}}, "k8s"), + ({"distributed": {"launcher": "slurm_multi"}}, "slurm"), + ({"distributed": {"launcher": "slurm-multi"}}, "slurm"), + ({"distributed": {"launcher": "torchrun"}}, "local"), + ({"distributed": {"launcher": "vllm"}}, "local"), + ({"distributed": {}}, "local"), + # An explicit k8s block still wins; slurm_multi is SLURM-only by construction + # but the explicit target is the user's stated intent. + ({"k8s": {}, "distributed": {"launcher": "slurm_multi"}}, "k8s"), + ]) + def test_inference(self, config, expected): + assert RunOrchestrator._infer_deployment_target(None, config) == expected diff --git a/tests/unit/test_slurm_multi.py b/tests/unit/test_slurm_multi.py index de5c2eae..cc286c85 100644 --- a/tests/unit/test_slurm_multi.py +++ b/tests/unit/test_slurm_multi.py @@ -595,3 +595,183 @@ def test_below_minimum_nodes_raises(self, deployment_factory): deployment_factory._generate_sglang_disagg_command( nnodes=1, nproc_per_node=8, master_port=12345 ) + + +# --------------------------------------------------------------------------- +# 6. Path-parity contract: the model card drives the allocation and the launcher +# +# slurm_multi is an escape hatch, not a second product. These lock in the pieces +# of the model-card contract that the templated path already honoured and the +# self-managed path silently dropped. + +from madengine.deployment.common import ( # noqa: E402 + resolve_launcher_from_sources, + resolve_node_count, +) + + +class TestResolveNodeCount: + """slurm.nodes and distributed.nnodes reconcile into one allocation size.""" + + def test_nnodes_absent_keeps_configured(self): + assert resolve_node_count(1, None, False) == (1, None) + + def test_nnodes_sizes_allocation_when_nodes_defaulted(self): + """The bug this fixes: a card declaring only nnodes ran on the nodes=1 default.""" + nodes, note = resolve_node_count(1, 4, nodes_explicitly_set=False) + assert nodes == 4 + assert note and "distributed.nnodes=4" in note + + def test_agreement_is_silent(self): + assert resolve_node_count(4, 4, nodes_explicitly_set=True) == (4, None) + + def test_explicit_nodes_wins_conflict_but_warns(self): + """Never silently allocate more nodes than the user asked to be billed for.""" + nodes, note = resolve_node_count(6, 4, nodes_explicitly_set=True) + assert nodes == 6 + assert note and "conflicts" in note + + @pytest.mark.parametrize("bad", ["bad", "", [], {}]) + def test_non_numeric_nnodes_falls_back(self, bad): + nodes, note = resolve_node_count(2, bad, nodes_explicitly_set=False) + assert nodes == 2 + if bad != "": + assert note is not None + + def test_idempotent_across_repeated_resolution(self): + """deploy() re-runs prepare() after preflight; resolution must not ratchet.""" + nodes, _ = resolve_node_count(1, 4, nodes_explicitly_set=False) + again, _ = resolve_node_count(1, 4, nodes_explicitly_set=False) + assert nodes == again == 4 + + +class TestResolveLauncherFromSources: + """One resolver for both dispatch sites, deployment config first.""" + + def test_deployment_config_wins(self): + assert resolve_launcher_from_sources("vllm", "slurm_multi") == "vllm" + + def test_falls_back_to_model_card(self): + assert resolve_launcher_from_sources(None, "slurm_multi") == "slurm_multi" + + def test_default_when_neither_declared(self): + assert resolve_launcher_from_sources(None, None) == "torchrun" + + def test_dispatch_and_env_block_cannot_disagree(self): + """ + prepare() used to read the card while _prepare_template_context() read the + deployment config, so a card-declared launcher could pick one path and emit + another path's env block. Both now call this, so they agree by construction. + """ + for deployment, card in [(None, "vllm"), ("sglang", "vllm"), (None, None)]: + assert resolve_launcher_from_sources(deployment, card) == \ + resolve_launcher_from_sources(deployment, card) + + +class TestSlurmMultiDeclaredResultsCsv: + """A slurm_multi card's `multiple_results` names its own results CSV.""" + + @pytest.fixture + def deployment(self, tmp_path: Path) -> SlurmDeployment: + script_rel = "scripts/wl/run.slurm" + script_abs = tmp_path / script_rel + script_abs.parent.mkdir(parents=True, exist_ok=True) + script_abs.write_text("#!/bin/bash\n") + + model = { + "name": "wl", + "scripts": script_rel, + "multiple_results": "perf_WL.csv", + "distributed": {"launcher": "slurm_multi"}, + } + manifest = { + "built_images": {}, + "built_models": {"img:tag": model}, + "context": {}, + } + manifest_path = tmp_path / "build_manifest.json" + manifest_path.write_text(json.dumps(manifest)) + cfg = DeploymentConfig( + target="slurm", + manifest_file=str(manifest_path), + additional_context={ + "slurm": {"output_dir": str(tmp_path / "slurm_results")}, + "distributed": {"launcher": "slurm_multi"}, + }, + ) + d = SlurmDeployment(cfg) + d._model_for_test = model + d._script_dir_for_test = script_abs.parent + return d + + def test_finds_csv_next_to_model_script(self, deployment): + """The wrapper cd's to the script dir, so $(pwd) writes land there.""" + target = deployment._script_dir_for_test / "perf_WL.csv" + target.write_text("model,performance,metric\nwl,1.0,tok/s\n") + found = deployment._slurm_multi_declared_result_csv( + deployment._model_for_test, "12345" + ) + assert found == target + + def test_ignores_empty_file(self, deployment): + (deployment._script_dir_for_test / "perf_WL.csv").write_text("") + assert deployment._slurm_multi_declared_result_csv( + deployment._model_for_test, "12345" + ) is None + + def test_returns_none_without_declaration(self, deployment): + assert deployment._slurm_multi_declared_result_csv({"name": "wl"}, "12345") is None + + def test_returns_none_when_model_info_missing(self, deployment): + assert deployment._slurm_multi_declared_result_csv(None, "12345") is None + + +class TestSlurmMultiPerfAggregation: + """Aggregating the per-job CSV into cwd/perf.csv must not append it to itself.""" + + @pytest.fixture + def deployment(self, tmp_path: Path) -> SlurmDeployment: + manifest = {"built_images": {}, "built_models": {}, "context": {}} + manifest_path = tmp_path / "build_manifest.json" + manifest_path.write_text(json.dumps(manifest)) + cfg = DeploymentConfig( + target="slurm", + manifest_file=str(manifest_path), + additional_context={"slurm": {"output_dir": str(tmp_path / "slurm_results")}}, + ) + d = SlurmDeployment(cfg) + d.output_dir.mkdir(parents=True, exist_ok=True) + return d + + def test_cwd_perf_source_is_not_duplicated(self, deployment, tmp_path, monkeypatch): + """ + The /perf.csv fallback (and a card declaring multiple_results: + "perf.csv") makes source and destination the same file. Appending it to + itself would double every row on each collection. + """ + monkeypatch.chdir(tmp_path) + rows = "model,performance,metric,status\nwl,1.0,tok/s,SUCCESS\n" + Path("perf.csv").write_text(rows) + + deployment._collect_slurm_multi_results( + "12345", {"perf_files": [], "logs": [], "successful_runs": [], "failed_runs": []}, None + ) + + assert Path("perf.csv").read_text() == rows + + def test_distinct_source_is_appended(self, deployment, tmp_path, monkeypatch): + """A genuinely separate per-job CSV still aggregates, minus its header.""" + monkeypatch.chdir(tmp_path) + Path("perf.csv").write_text("model,performance,metric,status\nold,1.0,tok/s,SUCCESS\n") + job_csv = tmp_path / "job" / "perf.csv" + job_csv.parent.mkdir() + job_csv.write_text("model,performance,metric,status\nnew,2.0,tok/s,SUCCESS\n") + deployment.slurm_config["results_dir"] = str(job_csv.parent) + + deployment._collect_slurm_multi_results( + "12345", {"perf_files": [], "logs": [], "successful_runs": [], "failed_runs": []}, None + ) + + text = Path("perf.csv").read_text() + assert "old,1.0" in text and "new,2.0" in text + assert text.count("model,performance") == 1