feat(rlix): sender-side NCCL broadcast + topology support (rlix#42)

examples/rlix/run_miles_dual.py

@@ -140,15 +140,40 @@ def _overlap_pools_from_env(num_gpus_per_node: int) -> (
                 )
         if len(set(mapping)) != len(mapping):
             raise ValueError(f"{label}={mapping} has duplicate GPU ids")
-    if not set(p1_train).issubset(set(p1_infer)):
-        raise ValueError(
-            f"p1_train={p1_train} not ⊆ p1_infer={p1_infer} "
-            f"(per-pipeline partial-overlap invariant)"
-        )
-    if not set(p2_train).issubset(set(p2_infer)):
+    # Per-pipeline topology families (rlix#42):
+    #   subset   (train ⊆ infer) — classic M11 partial overlap, fully tested;
+    #   disjoint (train ∩ infer == ∅) — dedicated train card(s); no engine
+    #            shares the sender's GPU, so all-broadcast NCCL transport
+    #            is possible. Newly admitted; logged as experimental.
+    # A PARTIAL intersection (crossing but neither subset nor disjoint)
+    # stays rejected: the scheduler's donor-shrink/grant accounting for a
+    # train pool that is half-inside / half-outside the infer pool has no
+    # test coverage yet — fail fast with the reason rather than wedge a
+    # run mid-training.
+    for label_t, train, label_i, infer in (
+        ("p1_train", p1_train, "p1_infer", p1_infer),
+        ("p2_train", p2_train, "p2_infer", p2_infer),
+    ):
+        train_set, infer_set = set(train), set(infer)
+        if train_set.issubset(infer_set):
+            continue
+        if not (train_set & infer_set):
+            import logging as _logging
+
+            _logging.getLogger("run_miles_dual").warning(
+                "%s=%s is fully DISJOINT from %s=%s — dedicated-train "
+                "topology (rlix#42): no colocate engines; every engine is "
+                "NCCL-broadcast-eligible. Newer than the overlap contract; "
+                "watch the first run.",
+                label_t, train, label_i, infer,
+            )
+            continue
         raise ValueError(
-            f"p2_train={p2_train} not ⊆ p2_infer={p2_infer} "
-            f"(per-pipeline partial-overlap invariant)"
+            f"{label_t}={train} partially intersects {label_i}={infer}: "
+            "each pipeline's train pool must be either a subset of its "
+            "infer pool (overlap/time-sharing) or fully disjoint from it "
+            "(dedicated train cards). Mixed shapes are unsupported — the "
+            "scheduler's shrink/grant accounting for them is unverified."
         )
     return (p1_train, p1_infer), (p2_train, p2_infer)
 
@@ -226,10 +251,16 @@ def _build_pipeline(
 
     train_size = len(train_mapping)
     infer_size = len(infer_mapping)
-    if not set(train_mapping).issubset(set(infer_mapping)):
+    # Same two-family rule as _overlap_pools_from_env (rlix#42): subset
+    # (overlap/time-sharing) or fully-disjoint (dedicated train cards)
+    # are valid; partial intersections are rejected there before this
+    # point, so only re-assert the invariant pair here.
+    _train_set, _infer_set = set(train_mapping), set(infer_mapping)
+    if not _train_set.issubset(_infer_set) and (_train_set & _infer_set):
         raise ValueError(
-            f"mp{pipeline_index}: train_mapping={train_mapping} not ⊆ "
-            f"infer_mapping={infer_mapping} (per-pipeline partial-overlap)"
+            f"mp{pipeline_index}: train_mapping={train_mapping} partially "
+            f"intersects infer_mapping={infer_mapping}; must be a subset "
+            "(overlap) or fully disjoint (dedicated train cards)"
         )
     args = _per_pipeline_args(
         base_args,

examples/rlix/run_miles_rlix.py

@@ -37,7 +37,62 @@ def _build_cluster_device_mappings(args) -> dict[str, list[int]]:
     convention) so train can be a strict subset of infer (partial
     overlap topology). No new device_mapping CLI args are introduced
     (Layer 1 forbidden).
+
+    rlix#42 explicit-mapping override: ``MILES_SINGLE_TRAIN_GPUS`` /
+    ``MILES_SINGLE_INFER_GPUS`` (comma lists of physical GPU ids) replace
+    the range derivation, admitting the dedicated-train (fully-disjoint)
+    topology — the minimal all-NCCL-broadcast shape (e.g. train "0",
+    infer "1,2"). Same two-family invariant as the dual driver: subset
+    (overlap) or fully disjoint; partial intersections rejected.
     """
+    import os
+
+    train_env = os.environ.get("MILES_SINGLE_TRAIN_GPUS", "").strip()
+    infer_env = os.environ.get("MILES_SINGLE_INFER_GPUS", "").strip()
+    if train_env or infer_env:
+        if not (train_env and infer_env):
+            raise ValueError(
+                "MILES_SINGLE_TRAIN_GPUS and MILES_SINGLE_INFER_GPUS must be "
+                "set together (comma lists of physical GPU ids)"
+            )
+        train = [int(g) for g in train_env.split(",") if g.strip() != ""]
+        infer = [int(g) for g in infer_env.split(",") if g.strip() != ""]
+        for label, mapping in (("train", train), ("infer", infer)):
+            if len(set(mapping)) != len(mapping):
+                raise ValueError(f"MILES_SINGLE_{label.upper()}_GPUS has duplicates: {mapping}")
+        train_set, infer_set = set(train), set(infer)
+        if not train_set.issubset(infer_set) and (train_set & infer_set):
+            raise ValueError(
+                f"train={train} partially intersects infer={infer}: must be a "
+                "subset (overlap) or fully disjoint (dedicated train cards)"
+            )
+        # The mapping lengths MUST match the CLI counts that still size
+        # the actual actors (RayTrainGroup uses actor_num_nodes ×
+        # actor_num_gpus_per_node; Phase B uses rollout_num_gpus) — a
+        # divergence would fail late in scheduler/placement work or
+        # silently allocate an unintended topology (codex impl-r9).
+        expected_train = int(args.actor_num_nodes) * int(args.actor_num_gpus_per_node)
+        expected_infer = int(args.rollout_num_gpus)
+        if len(train) != expected_train or len(infer) != expected_infer:
+            raise ValueError(
+                f"MILES_SINGLE_TRAIN_GPUS={train} / MILES_SINGLE_INFER_GPUS={infer} "
+                f"lengths must match the CLI-derived worker counts: expected "
+                f"len(train)=={expected_train} (actor_num_nodes × "
+                f"actor_num_gpus_per_node) and len(infer)=={expected_infer} "
+                f"(rollout_num_gpus); got {len(train)} / {len(infer)}"
+            )
+        if not (train_set & infer_set):
+            import logging as _logging
+
+            _logging.getLogger("run_miles_rlix").warning(
+                "train=%s is fully DISJOINT from infer=%s — dedicated-train "
+                "topology (rlix#42): no colocate engines; every engine is "
+                "NCCL-broadcast-eligible.",
+                train,
+                infer,
+            )
+        return {"actor_train": train, "actor_infer": infer}
+
     actor_count = int(args.actor_num_nodes) * int(args.actor_num_gpus_per_node)
     rollout_count = int(args.rollout_num_gpus)
     return {