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

[zhenyulincs]

miles half of rlops/rlix#42 (companion rlix PR: rlops/rlix#43 — full plan, evidence trail, and codex review history live there in plans/miles-nccl-broadcast-plan.md).

What

Sender-side NCCL for the F4 run_sync_session broadcast path (previously a deliberate NotImplementedError guard with dead receiver-side code), plus the driver topology support the rlix transport modes need. Boundary: every change is in RLix-port-added code — zero pristine upstream lines touched (UpdateWeightFromDistributed, arguments.py, rlix_validation.py, sglang_engine.py untouched).

Changes

megatron_utils/actor.py — _dispatch_nccl_broadcast / _run_sender_broadcast (mirrors the proven standalone UpdateWeightFromDistributed ordering):

1. dispatch receiver setup_collective_group refs async (no blocking get before the sender joins — rendezvous deadlock otherwise);
2. sender joins as rank 0 with a bounded rendezvous timeout that doubles as the process-group's per-collective bound (NCCL watchdog; fail-fast if TORCH_NCCL_ASYNC_ERROR_HANDLING/legacy alias is explicitly 0);
3. per bucket: metadata refs async → memory-preflighted CPU→GPU staging (mem_get_info + 1 GiB env-tunable margin, degrades to tensor-by-tensor — batching-only change) → dist.broadcast in metadata order under a cumulative transport deadline → receiver acks get the remaining window;
4. sender-owned finally teardown on every exit path: cancel nested refs, receiver destroy fan-out (400-tolerant), sender destroy_process_group. Budget hierarchy keeps worst-case unwind at 0.9× the service session deadline.

ray/rollout.py — RolloutManager.register_router_if_active: atomic active-only router re-admission for the rlix sync-under-load bracket. State check + idempotent /add_worker inside one serialized manager call (manager is max_concurrency-1 ⇒ cannot interleave with shrink_engines); concurrently-shrunk engines are skipped, not raised on (deliberately avoids _resolve_engine_indices, whose liveness contract raises for exactly those).

Drivers (user-directed topology support): per-pipeline invariant is now two-family — train ⊆ infer (overlap/time-sharing, the tested M11 contract) or fully-disjoint (dedicated train cards ⇒ every engine NCCL-eligible); partial intersections stay rejected with an actionable error. run_miles_rlix.py adds MILES_SINGLE_TRAIN_GPUS/MILES_SINGLE_INFER_GPUS explicit mappings (length-validated against the CLI-derived worker counts) — the minimal all-NCCL topology on 4 GPUs.

Test plan

• tests/test_nccl_broadcast_dispatch.py (16): dispatch ordering, receiver-failure abort, join-timeout budgets, stalled-transport cumulative deadline, preflight degradation, NCCL-watchdog env guard
• tests/test_register_router_if_active.py (4): contract on the real manager class — active-only, skip-not-raise, cheap empty probe, failure propagation
• tests/test_single_mapping_override.py (9): matching/undersized/oversized/duplicate/partial-intersection/subset/disjoint/half-set/legacy
• Existing tests/test_miles_pipeline.py unchanged and green
• E2E (4×A100, with feat(miles): NCCL broadcast weight-update transport (#42) rlix#43): 20-rollout dual overlap run — 45 NCCL group creations, 48 NCCL receives + 46 cpu-serialize legs (mixed every sync), zero OOM, zero flush timeouts, EXIT 0; NVLink counters show byte-exact GiB transfers on broadcast syncs and zero on the cpu path

Notes for reviewers

• Blame-classification: git diff touches only RLix-port sections (per-file rationale in the rlix PR's plan Appendix B).
• Known limitation (documented in the rlix plan): tms hook must be torch when broadcast + offload_train are combined — preload leaves stale cached allocator pointers at sender staging time.