Forbid positional arguments when constructing ray actors

miles/dashboard/backend.py

@@ -62,7 +62,9 @@ def init_dashboard(args, *, primary: bool = True, router_addr: str | None = None
                     node_id=ray.get_runtime_context().get_node_id(), soft=False
                 ),
             )
-            .remote(config, prometheus_handle_factory=_prometheus_factory if config.forward_prometheus else None)
+            .remote(
+                config=config, prometheus_handle_factory=_prometheus_factory if config.forward_prometheus else None
+            )
         )
         ray.get(_handle.ping.remote())
         _handle.start.remote()

miles/dashboard/collector.py

@@ -98,7 +98,7 @@ def _default_spawn_sampler(node_id: str, node_ip: str, interval: float):
             scheduling_strategy=NodeAffinitySchedulingStrategy(node_id=node_id, soft=False),
         )
         .remote(
-            _SelfGpuPush(ray.get_runtime_context().current_actor),
+            push=_SelfGpuPush(ray.get_runtime_context().current_actor),
             node=node_ip,
             interval=interval,
             push_processes=_SelfGpuProcessPush(ray.get_runtime_context().current_actor),
@@ -143,8 +143,8 @@ class DashboardCollector:
 
     def __init__(
         self,
-        config: CollectorConfig,
         *,
+        config: CollectorConfig,
         prometheus_handle_factory=None,  # () -> handle with .update.remote(dict), or None
         scraper_http_get=None,  # test hook, forwarded to SglangScraper
     ):

miles/dashboard/gpu_sampler.py

@@ -126,8 +126,8 @@ class GpuSampler:
 
     def __init__(
         self,
-        push: Callable[[str, list[GpuSample]], None],
         *,
+        push: Callable[[str, list[GpuSample]], None],
         node: str,
         interval: float = 1.0,
         nvml=None,

miles/ray/multi_lora/controller.py

@@ -59,7 +59,7 @@ def _load_subclass(path: str | None, base_cls):
 
 @ray.remote(num_cpus=0)
 class MultiLoRAController:
-    def __init__(self, args, router_url: str, host: str = "0.0.0.0") -> None:
+    def __init__(self, *, args, router_url: str, host: str = "0.0.0.0") -> None:
         backend_cls = _load_subclass(getattr(args, "multi_lora_backend_path", None), MultiLoRABackend)
         server_cls = _load_subclass(getattr(args, "multi_lora_http_server_path", None), MultiLoRAHTTPServer)
         self.backend = backend_cls(args, router_url)
@@ -120,4 +120,4 @@ def create_multilora_controller(args, router_url: str, host: str = "0.0.0.0"):
         name=CONTROLLER_NAME,
         namespace=CONTROLLER_NAMESPACE,
         **compute_ray_pin_head_options(),
-    ).remote(args, router_url, host)
+    ).remote(args=args, router_url=router_url, host=host)

miles/ray/placement_group.py

@@ -208,7 +208,7 @@ async def create_rollout_components(args, pg) -> RolloutComponents:
 
     rollout_executor = RolloutExecutor.options(
         num_cpus=1, num_gpus=0, **(compute_ray_pin_head_options() if args.pin_rollout_manager_to_head else {})
-    ).remote(args)
+    ).remote(args=args)
 
     # calculate num_rollout from num_epoch
     num_rollout_per_epoch = None

miles/ray/rollout/rollout_executor.py

@@ -47,7 +47,7 @@
 class RolloutExecutor:
     """The class to run rollout and convert rollout data to training data."""
 
-    def __init__(self, args):
+    def __init__(self, *, args):
         event_logger_checkpoint.restore(args)
         configure_logger(args, source=RolloutExecutorProcessIdentity())
 

miles/ray/train/actor_factory.py

@@ -93,11 +93,11 @@ def allocate_gpus_for_actor(
             rank_dir = os.path.join(args.offload_train_disk_dir, f"cell{cell_index}_rank{rank}")
             options["runtime_env"] = {"env_vars": {**env_vars, "TMS_DISK_BACKUP_DIR": rank_dir}}
         actor = TrainRayActor.options(**options).remote(
-            args,
-            world_size,
-            rank,
-            master_addr,
-            master_port,
+            args=args,
+            world_size=world_size,
+            rank=rank,
+            master_addr=master_addr,
+            master_port=master_port,
             indep_dp_store_addr=indep_dp_store_addr,
             role=role,
             cell_index=cell_index,

miles/ray/train_actor.py

@@ -39,6 +39,7 @@ def get_local_gpu_id():
 class TrainRayActor:
     def __init__(
         self,
+        *,
         args,
         world_size: int,
         rank: int,

miles/utils/http_utils.py

@@ -345,7 +345,7 @@ def _init_ray_distributed_post(args):
     # Define the async actor
     @ray.remote
     class _HttpPosterActor:
-        def __init__(self, concurrency: int):
+        def __init__(self, *, concurrency: int):
             # Lazy creation to this actor's event loop
             self._client = httpx.AsyncClient(
                 limits=httpx.Limits(max_connections=max(1, concurrency)),
@@ -371,7 +371,7 @@ async def do_post(self, url, payload, max_retries=60, action="post", headers=Non
                 max_concurrency=per_actor_conc,
                 # Use tiny CPU to schedule
                 num_cpus=0.001,
-            ).remote(per_actor_conc)
+            ).remote(concurrency=per_actor_conc)
             created.append(actor)
 
     _post_actors = created

miles/utils/tracking_utils/prometheus_utils.py

@@ -40,7 +40,7 @@ def init_prometheus(args, start_server: bool = False):
                     soft=False,
                 ),
             )
-            .remote(args)
+            .remote(args=args)
         )
         ray.get(_collector_handle.ping.remote())
         logger.info("Prometheus collector actor created")
@@ -74,7 +74,7 @@ class _PrometheusCollector:
     Ray handles the RPC transparently.
     """
 
-    def __init__(self, args):
+    def __init__(self, *, args):
         from prometheus_client import Gauge, start_http_server
 
         self._Gauge = Gauge

tests/fast-gpu/test_gpu_sampler_hw.py

@@ -17,7 +17,7 @@ def __call__(self, node, batch):
 
 
 def test_auto_detection_picks_the_backend_matching_the_hardware():
-    sampler = GpuSampler(PushSpy(), node="ci")
+    sampler = GpuSampler(push=PushSpy(), node="ci")
     assert sampler.available, "no GPU telemetry backend initialized on a GPU runner"
     expected = "AMD SMI" if torch.version.hip else "NVML"
     assert sampler._provider.name == expected
@@ -26,7 +26,7 @@ def test_auto_detection_picks_the_backend_matching_the_hardware():
 def test_every_device_reports_telemetry_and_processes():
     push = PushSpy()
     push_processes = PushSpy()
-    sampler = GpuSampler(push, node="ci", push_processes=push_processes)
+    sampler = GpuSampler(push=push, node="ci", push_processes=push_processes)
     assert sampler.available
 
     uuids = sampler.gpu_uuids()

tests/fast/dashboard/test_collector.py

@@ -1,7 +1,9 @@
 import logging
 import time
 from pathlib import Path
+from types import SimpleNamespace
 
+import pytest
 from tests.fast.dashboard.dummy_telemetry import BASE_TS, dump_dummy_telemetry
 
 from miles.dashboard.collector import CollectorConfig, DashboardCollector
@@ -26,7 +28,7 @@ def make_collector(tmp_path, **kwargs) -> DashboardCollector:
     config = kwargs.pop("config", None) or CollectorConfig(
         dashboard_dir=str(tmp_path / "dashboard"), run_name="collector-test", start_ts=1.0
     )
-    return DashboardCollector(config, **kwargs)
+    return DashboardCollector(config=config, **kwargs)
 
 
 def test_collector_satisfies_dummy_telemetry_contract(tmp_path):
@@ -114,7 +116,7 @@ def test_flush_thread_persists_periodically(tmp_path):
     config = CollectorConfig(
         dashboard_dir=str(tmp_path / "dashboard"), run_name="r", start_ts=0.0, flush_interval_seconds=0.05
     )
-    collector = DashboardCollector(config)
+    collector = DashboardCollector(config=config)
     collector.start()
     collector.push_metrics(MetricsRecord(ts=1.0, step_key="rollout/step", step=0, metrics={"a": 1}))
     time.sleep(0.2)
@@ -208,7 +210,7 @@ class FakeHandle:
     config = CollectorConfig(
         dashboard_dir=str(tmp_path / "dashboard"), run_name="r", start_ts=0.0, forward_prometheus=True
     )
-    collector = DashboardCollector(config, prometheus_handle_factory=lambda: FakeHandle())
+    collector = DashboardCollector(config=config, prometheus_handle_factory=lambda: FakeHandle())
     collector.push_gpu_samples(
         "10.0.0.1", [GpuSample(ts=1.0, node="10.0.0.1", gpu=0, util=87, mem_mb=1000, power_w=600)]
     )
@@ -313,3 +315,71 @@ def test_external_engines_synthesized_from_scrapes(tmp_path):
         "http://10.1.0.5:15000": "regular",
         "http://10.1.0.6:15000": "external",
     }
+
+
+class TestConstructorContract:
+    def test_a_positional_config_is_rejected(self, tmp_path):
+        """The collector is built as a ray actor, so config must be bound by keyword only."""
+        config = CollectorConfig(dashboard_dir=str(tmp_path / "dashboard"), run_name="r", start_ts=1.0)
+
+        with pytest.raises(TypeError):
+            DashboardCollector(config)
+
+
+class _FakeSamplerStart:
+    def remote(self):
+        return True
+
+
+class _FakeSamplerHandle:
+    def __init__(self, sampler):
+        self.sampler = sampler
+        self.start = _FakeSamplerStart()
+
+
+class _FakeRayActorClass:
+    def __init__(self, cls):
+        self._cls = cls
+        self.options_kwargs = None
+        self.positional_args = None
+        self.keyword_args = None
+
+    def options(self, **kwargs):
+        self.options_kwargs = kwargs
+        return self
+
+    def remote(self, *args, **kwargs):
+        self.positional_args = args
+        self.keyword_args = kwargs
+        return _FakeSamplerHandle(self._cls(*args, **kwargs))
+
+
+class TestDefaultSpawnSampler:
+    def test_builds_the_gpu_sampler_through_keyword_arguments_only(self, monkeypatch):
+        """The sampler actor is constructed with keywords its keyword-only signature accepts."""
+        import ray
+
+        from miles.dashboard import collector as collector_mod
+        from miles.dashboard.gpu_sampler import GpuSampler
+
+        actor_classes: list[_FakeRayActorClass] = []
+
+        def fake_ray_remote(cls):
+            actor_class = _FakeRayActorClass(cls)
+            actor_classes.append(actor_class)
+            return actor_class
+
+        monkeypatch.setattr(ray, "remote", fake_ray_remote)
+        monkeypatch.setattr(ray, "get", lambda value: value)
+        monkeypatch.setattr(ray, "kill", lambda handle: None)
+        monkeypatch.setattr(ray, "get_runtime_context", lambda: SimpleNamespace(current_actor="collector-handle"))
+
+        handle = collector_mod._default_spawn_sampler("a" * 56, "10.0.0.7", 0.25)
+
+        [actor_class] = actor_classes
+        assert actor_class.positional_args == ()
+        assert set(actor_class.keyword_args) == {"push", "node", "interval", "push_processes"}
+        sampler = handle.sampler
+        assert isinstance(sampler, GpuSampler)
+        assert (sampler.node, sampler.interval) == ("10.0.0.7", 0.25)
+        assert isinstance(sampler._push, collector_mod._SelfGpuPush)