honor model-card declarations on the slurm_multi path

docs/launchers.md

@@ -647,6 +647,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 <model>.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": "<supply-your-image>"` 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

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:
     """

src/madengine/deployment/slurm.py

@@ -25,6 +25,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
@@ -63,6 +65,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)
 
@@ -73,6 +82,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"))
@@ -302,11 +317,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(
@@ -359,6 +373,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."""
@@ -607,9 +661,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
@@ -1741,10 +1797,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
                 )
 
 
@@ -2068,12 +2123,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.
@@ -2082,15 +2191,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/<user>/<jobid>/perf.csv
         #   * Large EP / KV cache:  <workspace>/slurm_output/perf_csv/*<jobid>*.csv
         # Plus the legacy <cwd>/perf.csv path some flows still use.
-        # Priority: results_dir config > shared_inference NFS > slurm_output/perf_csv
-        # > <cwd>/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 > <cwd>/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:
@@ -2139,16 +2252,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 <cwd>/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]"