From 98cdf0df7bb2f79e60763a1584b62d795d26e766 Mon Sep 17 00:00:00 2001
From: fzyzcjy <ch271828n@outlook.com>
Date: Tue, 4 Aug 2026 20:34:47 +0800
Subject: [PATCH] Route all trainer worker calls through the worker handle
 instead of Ray handles

---
 miles/dashboard/hooks.py                      |  17 +-
 miles/ray/rollout/inference_controller.py     |   2 +-
 miles/ray/train/cell.py                       |  57 +---
 miles/ray/train/cell_monitor.py               |   2 +-
 miles/ray/train/cell_state.py                 |   5 +-
 miles/utils/workers/ray_worker_manager.py     |   3 +-
 miles/utils/workers/worker_info.py            |  13 +-
 tests/fast/dashboard/test_hooks.py            | 222 ++++++++++++-
 tests/fast/dashboard/test_trajectory_sink.py  |  32 ++
 .../ray/rollout/test_inference_controller.py  | 293 +++++++++++++++++-
 .../rollout/test_inference_controller_tick.py |  21 ++
 tests/fast/ray/train/conftest.py              |   4 +
 tests/fast/ray/train/fake_worker_manager.py   |   2 +-
 tests/fast/ray/train/test_cell.py             | 158 ++--------
 tests/fast/ray/train/test_cell_master_addr.py |   6 +-
 tests/fast/ray/train/test_cell_monitor.py     |  16 +-
 tests/fast/ray/train/test_cell_restart.py     |   8 +-
 tests/fast/ray/train/test_group.py            |  93 +++---
 .../ray/train/test_group_failure_reporting.py |   8 +-
 .../train/test_group_reconcile_adapters.py    |   6 +-
 .../ray/train/test_train_external_data.py     |   4 +-
 .../fast/ray/train/test_train_return_value.py |   6 +-
 .../real_ray/test_ray_worker_manager.py       |  26 +-
 .../utils/workers/test_ray_worker_manager.py  | 107 ++++++-
 .../workers/test_ray_worker_manager_serve.py  |  16 +-
 tests/fast/utils/workers/test_worker_info.py  |  41 +++
 .../utils/workers/worker_provider/test_ray.py |  11 +-
 27 files changed, 878 insertions(+), 301 deletions(-)
 create mode 100644 tests/fast/utils/workers/test_worker_info.py

diff --git a/miles/dashboard/hooks.py b/miles/dashboard/hooks.py
index 089e8a9b57a..19c3625f645 100644
--- a/miles/dashboard/hooks.py
+++ b/miles/dashboard/hooks.py
@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import threading
 import time
@@ -321,7 +322,7 @@ def register_router(args) -> None:
         _warner.warn("dashboard router registration failed; engine metrics will be missing")
 
 
-def register_engines(servers) -> None:
+async def register_engines(servers) -> None:
     """Called at the top of every InferenceController.prepare_rollout(): pushes an engine
     topology snapshot whenever the set of engine actors changed (startup,
     fault-tolerance recovery). Steady state costs one worker-manager round trip
@@ -342,7 +343,7 @@ def register_engines(servers) -> None:
         )
         if fingerprint == _engines_fingerprint:
             return
-        engines = _compute_engine_infos(cells, worker_infos_per_cell)
+        engines = await _compute_engine_infos(cells, worker_infos_per_cell)
         handle.update_topology.remote(TopologySnapshot(ts=time.time(), engines=engines))
         _engines_fingerprint = fingerprint
     except Exception:
@@ -389,13 +390,11 @@ def _collect_worker_infos(cells) -> list[list]:
     return _ray_get(futures)
 
 
-def _compute_engine_infos(cells, worker_infos_per_cell) -> list[EngineInfo]:
-    uuid_refs = [
-        info.actor_handle._get_gpu_uuids.remote(info.gpu_ids)
-        for worker_infos in worker_infos_per_cell
-        for info in worker_infos
-    ]
-    probed_uuids = iter(_ray_get(uuid_refs))
+async def _compute_engine_infos(cells, worker_infos_per_cell) -> list[EngineInfo]:
+    flat_infos = [info for worker_infos in worker_infos_per_cell for info in worker_infos]
+    probed_uuids = iter(
+        await asyncio.gather(*[info.handle._get_gpu_uuids(gpu_ids=info.gpu_ids) for info in flat_infos])
+    )
 
     engines = []
     for engine_rank, (cell, worker_infos) in enumerate(zip(cells, worker_infos_per_cell, strict=True)):
diff --git a/miles/ray/rollout/inference_controller.py b/miles/ray/rollout/inference_controller.py
index 19f5141d08d..8c75f42f7b5 100644
--- a/miles/ray/rollout/inference_controller.py
+++ b/miles/ray/rollout/inference_controller.py
@@ -78,7 +78,7 @@ async def init(self) -> None:
     @with_lock
     async def prepare_rollout(self, rollout_id):
         await self._health_monitoring_resume()
-        dashboard_hooks.register_engines(self.servers)
+        await dashboard_hooks.register_engines(self.servers)
 
     @with_lock
     async def prepare_eval(self):
diff --git a/miles/ray/train/cell.py b/miles/ray/train/cell.py
index 51343b07472..84962067cfb 100644
--- a/miles/ray/train/cell.py
+++ b/miles/ray/train/cell.py
@@ -2,8 +2,6 @@
 import logging
 import time
 
-import ray
-
 from miles.ray.specs.train import MASTER_PORT_NAME
 from miles.ray.train.cell_monitor import compute_cell_status
 from miles.ray.train.cell_state import (
@@ -18,12 +16,14 @@
 from miles.utils.ft_utils.indep_dp import IndepDPInfo
 from miles.utils.retry_utils import NonRetryableError
 from miles.utils.tracking_utils.structured_log import log_structured
+from miles.utils.workers.worker_handle import BaseWorkerHandle, WorkerUnreachableError
 from miles.utils.workers.worker_provider.ray import RayWorkerProvider
 from miles.utils.workers.worker_spec import HostAndPort
 
 logger = logging.getLogger(__name__)
 
 KILL_RPC_TIMEOUT_S = 10.0
+CONFIRM_DEAD_TIMEOUT_S = 120.0
 
 
 class RayTrainCell:
@@ -54,9 +54,7 @@ def __init__(
         self._master_addr: HostAndPort = worker_infos[0].self_addrs[MASTER_PORT_NAME]
 
         # NOTE: do *NOT* directly modify `self._state`, but instead use `self._change_state`
-        self._state: CellState = StateAllocatedUninitialized(
-            actor_handles=[info.actor_handle for info in worker_infos]
-        )
+        self._state: CellState = StateAllocatedUninitialized(worker_handles=[info.handle for info in worker_infos])
 
     # ------------------------ API ------------------------
 
@@ -94,7 +92,7 @@ async def train(
         attempt: int,
         external_data: list | None = None,
     ) -> list:
-        if external_data is not None and len(external_data) != len(self._get_actor_handles()):
+        if external_data is not None and len(external_data) != len(self._get_worker_handles()):
             raise NonRetryableError("external_data must contain one payload per train worker")
 
         return await self._execute_raw(
@@ -141,14 +139,14 @@ async def prepare_indep_dp_mode_healing(
     # ------------------------ state transition ------------------------
 
     async def _kill_workers_and_confirm_dead(self) -> None:
-        handles = self._get_actor_handles() if self.is_allocated else []
+        handles = self._get_worker_handles() if self.is_allocated else []
 
         log_structured(
             logger.info, tag="ft", op="confirm_dead", phase="start", cell=self.cell_id, n_actors=len(handles)
         )
         start = time.monotonic()
         await asyncio.gather(*[_kill_worker(handle) for handle in handles])
-        await asyncio.gather(*[_confirm_actor_dead(handle) for handle in handles])
+        await asyncio.gather(*[handle.wait_dead(timeout=CONFIRM_DEAD_TIMEOUT_S) for handle in handles])
         log_structured(
             logger.info,
             tag="ft",
@@ -162,14 +160,14 @@ def _mark_as_alive(self, indep_dp_info: IndepDPInfo) -> None:
         self._change_state(
             "_mark_as_alive",
             StateAllocatedUninitialized,
-            StateAllocatedAlive(actor_handles=self._state.actor_handles, indep_dp_info=indep_dp_info),
+            StateAllocatedAlive(worker_handles=self._state.worker_handles, indep_dp_info=indep_dp_info),
         )
 
     def _update_indep_dp_info(self, indep_dp_info: IndepDPInfo) -> None:
         self._change_state(
             "_update_indep_dp_info",
             StateAllocatedAlive,
-            StateAllocatedAlive(actor_handles=self._state.actor_handles, indep_dp_info=indep_dp_info),
+            StateAllocatedAlive(worker_handles=self._state.worker_handles, indep_dp_info=indep_dp_info),
         )
 
     def _mark_as_errored(self) -> None:
@@ -180,7 +178,7 @@ def _mark_as_errored(self) -> None:
         self._change_state(
             "_mark_as_errored",
             (StateAllocatedUninitialized, StateAllocatedAlive, StateAllocatedErrored),
-            StateAllocatedErrored(actor_handles=self._state.actor_handles, indep_dp_info=indep_dp_info),
+            StateAllocatedErrored(worker_handles=self._state.worker_handles, indep_dp_info=indep_dp_info),
         )
 
     def _change_state(
@@ -225,14 +223,14 @@ async def _execute_raw(
         compute_kwargs,
         kill_on_failure: bool = True,
     ) -> list:
-        handles = self._get_actor_handles()
+        handles = self._get_worker_handles()
         log_structured(
             logger.info, tag="ft", op="execute", phase="start", cell=self.cell_id, fn=fn_name, n_actors=len(handles)
         )
         start = time.monotonic()
         try:
             result = await asyncio.gather(
-                *[getattr(actor, fn_name).remote(**compute_kwargs(i)) for i, actor in enumerate(handles)]
+                *[getattr(handle, fn_name)(**compute_kwargs(i)) for i, handle in enumerate(handles)]
             )
             log_structured(
                 logger.info,
@@ -291,40 +289,17 @@ def indep_dp_info(self) -> IndepDPInfo | None:
         assert isinstance(self._state, (StateAllocatedAlive, StateAllocatedErrored))
         return self._state.indep_dp_info
 
-    def _get_actor_handles(self) -> list[ray.actor.ActorHandle]:
+    def _get_worker_handles(self) -> list[BaseWorkerHandle]:
         assert isinstance(
             self._state, StateAllocatedBase
         ), f"Cell {self.cell_id} is not allocated (state={type(self._state).__name__})"
-        return self._state.actor_handles
+        return self._state.worker_handles
 
 
-async def _kill_worker(handle: ray.actor.ActorHandle) -> None:
+async def _kill_worker(handle: BaseWorkerHandle) -> None:
     try:
-        await asyncio.wait_for(handle.kill_self.remote(), timeout=KILL_RPC_TIMEOUT_S)
-    except (ray.exceptions.RayActorError, ray.exceptions.RayTaskError):
+        await asyncio.wait_for(handle.kill_self(), timeout=KILL_RPC_TIMEOUT_S)
+    except WorkerUnreachableError:
         return
     except (TimeoutError, asyncio.TimeoutError):
         logger.warning("Timed out asking a worker to kill itself; falling back to the death confirmation probe")
-
-
-async def _confirm_actor_dead(handle: ray.actor.ActorHandle) -> None:
-    CONFIRM_DEAD_TIMEOUT_S = 120.0
-    CONFIRM_DEAD_PROBE_INTERVAL_S = 1.0
-
-    async def _probe() -> None:
-        await handle.__ray_ready__.remote()
-
-    deadline = time.monotonic() + CONFIRM_DEAD_TIMEOUT_S
-    while True:
-        try:
-            await asyncio.wait_for(_probe(), timeout=CONFIRM_DEAD_PROBE_INTERVAL_S)
-        except (ray.exceptions.RayActorError, ray.exceptions.RayTaskError):
-            return
-        except (TimeoutError, asyncio.TimeoutError):
-            pass
-
-        if time.monotonic() >= deadline:
-            logger.error("Timed out after %.0fs confirming actor death; proceeding anyway", CONFIRM_DEAD_TIMEOUT_S)
-            return
-
-        await asyncio.sleep(CONFIRM_DEAD_PROBE_INTERVAL_S)
diff --git a/miles/ray/train/cell_monitor.py b/miles/ray/train/cell_monitor.py
index 90fde7082c0..6f6211ad211 100644
--- a/miles/ray/train/cell_monitor.py
+++ b/miles/ray/train/cell_monitor.py
@@ -24,7 +24,7 @@ async def _check() -> None:
         # Cell health is liveness, not training progress: the heartbeat RPC runs on
         # a dedicated concurrency group and returns even while the training thread is
         # blocked in a (legitimately waiting) cross-cell collective. A returned result
-        # proves the process is alive; an RayActorError or RPC timeout proves it is not.
+        # proves the process is alive; a WorkerUnreachableError or RPC timeout proves it is not.
         if not cell.is_alive:
             return
 
diff --git a/miles/ray/train/cell_state.py b/miles/ray/train/cell_state.py
index dcb6fa23012..f1a06ca9114 100644
--- a/miles/ray/train/cell_state.py
+++ b/miles/ray/train/cell_state.py
@@ -1,8 +1,7 @@
-import ray
-
 from pydantic import BaseModel, ConfigDict
 
 from miles.utils.ft_utils.indep_dp import IndepDPInfo
+from miles.utils.workers.worker_handle import BaseWorkerHandle
 
 
 class StateBase(BaseModel):
@@ -10,7 +9,7 @@ class StateBase(BaseModel):
 
 
 class StateAllocatedBase(StateBase):
-    actor_handles: list[ray.actor.ActorHandle]
+    worker_handles: list[BaseWorkerHandle]
 
 
 class StateAllocatedUninitialized(StateAllocatedBase):
diff --git a/miles/utils/workers/ray_worker_manager.py b/miles/utils/workers/ray_worker_manager.py
index 83244f5cadc..1ea8a61d68c 100644
--- a/miles/utils/workers/ray_worker_manager.py
+++ b/miles/utils/workers/ray_worker_manager.py
@@ -15,6 +15,7 @@
 from miles.utils.workers.addr_allocator import PortAllocator
 from miles.utils.workers.command_actor import CommandActor
 from miles.utils.workers.naming import compute_cell_id, compute_worker_name
+from miles.utils.workers.ray_worker_handle import RayWorkerHandle
 from miles.utils.workers.worker_info import WorkerInfo
 from miles.utils.workers.worker_provider.base import CellInfo
 from miles.utils.workers.worker_spec import (
@@ -98,7 +99,7 @@ def get_worker_infos(self, cell_id: str) -> list[WorkerInfo]:
                 generation=actor.generation,
                 self_addrs=actor.self_addrs,
                 gpu_ids=actor.gpu_ids,
-                actor_handle=actor.actor_handle,
+                handle=RayWorkerHandle(actor.actor_handle),
             )
             for actor in (cell.actors if cell.actors is not None else [])
         ]
diff --git a/miles/utils/workers/worker_info.py b/miles/utils/workers/worker_info.py
index f5db336f9f7..8ffc1348518 100644
--- a/miles/utils/workers/worker_info.py
+++ b/miles/utils/workers/worker_info.py
@@ -1,16 +1,17 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-
-import ray.actor
+from pydantic import ConfigDict
 
+from miles.utils.pydantic_utils import StrictBaseModel
+from miles.utils.workers.worker_handle import BaseWorkerHandle
 from miles.utils.workers.worker_spec import NamedHostAndPorts
 
 
-@dataclass(kw_only=True)
-class WorkerInfo:
+class WorkerInfo(StrictBaseModel):
+    model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
+
     name: str
     generation: int
     self_addrs: NamedHostAndPorts
     gpu_ids: list[int]
-    actor_handle: ray.actor.ActorHandle
+    handle: BaseWorkerHandle
diff --git a/tests/fast/dashboard/test_hooks.py b/tests/fast/dashboard/test_hooks.py
index da2ada92562..1a5353f696f 100644
--- a/tests/fast/dashboard/test_hooks.py
+++ b/tests/fast/dashboard/test_hooks.py
@@ -1,3 +1,4 @@
+import asyncio
 import logging
 
 import numpy as np
@@ -5,10 +6,11 @@
 
 from miles.dashboard import backend, hooks
 from miles.dashboard.hooks import BATCH_MAX_EVENTS, BATCH_MAX_SECONDS, _Identity
-from miles.dashboard.store import Role
+from miles.dashboard.store import EngineInfo, Role
 from miles.ray.rollout.server_cell import ServerCellMetadata
 from miles.utils.timer import Timer
 from miles.utils.workers.ray_worker_manager import RayWorkerManager
+from miles.utils.workers.worker_handle import BaseWorkerHandle
 from miles.utils.workers.worker_info import WorkerInfo
 from miles.utils.workers.worker_spec import HostAndPort
 
@@ -146,9 +148,65 @@ def remote(self, *args, **kwargs):
         return self._value_fn(*args, **kwargs)  # hooks._ray_get is patched to the identity function
 
 
-class FakeEngineHandle:
+class FakeWorkerHandle(BaseWorkerHandle):
+    async def _get_gpu_uuids(self, *, gpu_ids):
+        return [None] * len(gpu_ids)
+
+    async def wait_ready(self, *, timeout):
+        return None
+
+    async def wait_dead(self, *, timeout):
+        return None
+
+
+class UuidWorkerHandle(BaseWorkerHandle):
+    def __init__(self, uuid_by_gpu_id: dict[int, str]):
+        self._uuid_by_gpu_id = uuid_by_gpu_id
+
+    async def _get_gpu_uuids(self, *, gpu_ids):
+        return [self._uuid_by_gpu_id[gpu_id] for gpu_id in gpu_ids]
+
+    async def wait_ready(self, *, timeout):
+        return None
+
+    async def wait_dead(self, *, timeout):
+        return None
+
+
+class GatedWorkerHandle(BaseWorkerHandle):
+    def __init__(self, uuid, *, signal=None, wait_for=None):
+        self._uuid = uuid
+        self._signal = signal
+        self._wait_for = wait_for
+
+    async def _get_gpu_uuids(self, *, gpu_ids):
+        if self._signal is not None:
+            self._signal.set()
+        if self._wait_for is not None:
+            await self._wait_for.wait()
+        return [self._uuid for _ in gpu_ids]
+
+    async def wait_ready(self, *, timeout):
+        return None
+
+    async def wait_dead(self, *, timeout):
+        return None
+
+
+class FlakyWorkerHandle(BaseWorkerHandle):
     def __init__(self):
-        self._get_gpu_uuids = _FakeProbe(lambda gpu_ids: [None] * len(gpu_ids))
+        self.fail = True
+
+    async def _get_gpu_uuids(self, *, gpu_ids):
+        if self.fail:
+            raise RuntimeError("worker unreachable")
+        return [None] * len(gpu_ids)
+
+    async def wait_ready(self, *, timeout):
+        return None
+
+    async def wait_dead(self, *, timeout):
+        return None
 
 
 class FakeManagerHandle:
@@ -161,10 +219,10 @@ def __init__(self, infos_by_cell):
 class FakeCell:
     """Duck-typed ServerCell: the hooks read only the driver-side routing facts."""
 
-    def __init__(self, url, cell_index=0, alive=True):
+    def __init__(self, url, cell_index=0, alive=True, worker_type="regular"):
         self.meta = ServerCellMetadata(
             model_id="default",
-            worker_type="regular",
+            worker_type=worker_type,
             cell_id=f"inference-engine-0-0-{cell_index}",
             num_gpus_per_engine=1,
             gpu_offset=cell_index,
@@ -178,13 +236,13 @@ def __init__(self, url, cell_index=0, alive=True):
         self.is_pending_weights_or_serving = alive
 
 
-def _worker_info(name, node, gpus, generation=1):
+def _worker_info(name, node, gpus, generation=1, handle=None):
     return WorkerInfo(
         name=name,
         generation=generation,
         self_addrs={"primary": HostAndPort(host=node, port=30001)},
         gpu_ids=gpus,
-        actor_handle=FakeEngineHandle(),
+        handle=handle if handle is not None else FakeWorkerHandle(),
     )
 
 
@@ -193,7 +251,7 @@ def _servers(cells):
     return {"default": server}
 
 
-def test_register_engines_groups_multinode_and_dedups(monkeypatch):
+async def test_register_engines_groups_multinode_and_dedups(monkeypatch):
     """Worker-manager infos become one EngineInfo per cell; repush only on worker change."""
     handle = FakeHandle()
     monkeypatch.setattr(backend, "_handle", handle)
@@ -207,7 +265,7 @@ def test_register_engines_groups_multinode_and_dedups(monkeypatch):
     monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(lambda: FakeManagerHandle(infos_by_cell)))
     servers = _servers([FakeCell("http://a:1", cell_index=0), FakeCell("http://b:1", cell_index=1)])
 
-    hooks.register_engines(servers)
+    await hooks.register_engines(servers)
     [(args, _)] = handle.update_topology.calls
     [snapshot] = args
     assert [e.addr for e in snapshot.engines] == ["http://a:1", "http://b:1"]
@@ -215,22 +273,28 @@ def test_register_engines_groups_multinode_and_dedups(monkeypatch):
     assert multinode.gpus == [["node-a", 0], ["node-a", 1], ["node-b", 0], ["node-b", 1]]
     assert len(multinode.gpu_uuids) == 4
 
-    hooks.register_engines(servers)  # steady state: fingerprint unchanged
+    await hooks.register_engines(servers)  # steady state: fingerprint unchanged
     assert len(handle.update_topology.calls) == 1
 
     infos_by_cell["inference-engine-0-0-1"] = [_worker_info("inference-engine-0-1-0", "node-a", [2, 3], generation=2)]
-    hooks.register_engines(servers)  # recovery: same worker, new generation
+    await hooks.register_engines(servers)  # recovery: same worker, new generation
     assert len(handle.update_topology.calls) == 2
 
+    # Counting the repush says it fired, not what it carried: the fingerprint watches the
+    # worker while the addr comes from the cell, so a repush can still publish stale engines.
+    ([republished], _) = handle.update_topology.calls[1]
+    assert [e.addr for e in republished.engines] == ["http://a:1", "http://b:1"]
+    assert republished.engines[1].gpus == [["node-a", 2], ["node-a", 3]]
 
-def test_register_engines_skips_dead_cells(monkeypatch):
+
+async def test_register_engines_skips_dead_cells(monkeypatch):
     """Cells that are not alive are left out of the snapshot and never queried."""
     handle = FakeHandle()
     monkeypatch.setattr(backend, "_handle", handle)
     infos_by_cell = {"inference-engine-0-0-0": [_worker_info("inference-engine-0-0-0", "n", [0])]}
     monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(lambda: FakeManagerHandle(infos_by_cell)))
 
-    hooks.register_engines(
+    await hooks.register_engines(
         _servers([FakeCell("http://a:1", cell_index=0), FakeCell("http://b:1", cell_index=1, alive=False)])
     )
 
@@ -238,7 +302,7 @@ def test_register_engines_skips_dead_cells(monkeypatch):
     assert [e.addr for e in args[0].engines] == ["http://a:1"]
 
 
-def test_register_engines_survives_missing_worker_manager(monkeypatch, caplog):
+async def test_register_engines_survives_missing_worker_manager(monkeypatch, caplog):
     """Engines not yet owned by the worker manager degrade to a warning, not a crash."""
     handle = FakeHandle()
     monkeypatch.setattr(backend, "_handle", handle)
@@ -249,17 +313,130 @@ def _no_manager():
     monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(_no_manager))
     hooks._warner.reset_window_for_test()
     with caplog.at_level(logging.WARNING):
-        hooks.register_engines(_servers([FakeCell("http://a:1")]))
+        await hooks.register_engines(_servers([FakeCell("http://a:1")]))
 
     assert handle.update_topology.calls == []
     assert any("engine registration failed" in r.message for r in caplog.records)
 
 
-def test_register_engines_without_collector_is_noop():
-    hooks.register_engines(_servers([FakeCell("http://a:1")]))
+async def test_register_engines_without_collector_is_noop():
+    await hooks.register_engines(_servers([FakeCell("http://a:1")]))
     assert hooks._engines_fingerprint is None
 
 
+async def test_register_engines_publishes_topology_from_a_running_event_loop(monkeypatch, caplog):
+    """Driven from inside a running asyncio loop (as prepare_rollout does), the topology is published without warning."""
+    handle = FakeHandle()
+    monkeypatch.setattr(backend, "_handle", handle)
+    infos_by_cell = {"inference-engine-0-0-0": [_worker_info("inference-engine-0-0-0", "node-a", [0, 1])]}
+    monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(lambda: FakeManagerHandle(infos_by_cell)))
+    hooks._warner.reset_window_for_test()
+
+    assert asyncio.get_running_loop().is_running()
+    with caplog.at_level(logging.WARNING):
+        await hooks.register_engines(_servers([FakeCell("http://a:1")]))
+
+    assert [e.addr for e in handle.update_topology.calls[0][0][0].engines] == ["http://a:1"]
+    assert hooks._engines_fingerprint is not None
+    assert not [r for r in caplog.records if "engine registration failed" in r.message]
+
+
+async def test_compute_engine_infos_projects_cells_and_workers_exactly():
+    """Every cell becomes one EngineInfo whose gpu pairs and probed uuids stay aligned worker by worker."""
+    cells = [
+        FakeCell("http://a:1", cell_index=0, worker_type="decode"),
+        FakeCell("http://b:1", cell_index=1, worker_type="prefill"),
+    ]
+    worker_infos_per_cell = [
+        [
+            _worker_info("engine-0-0", "[2001:db8::7]", [4, 5], handle=UuidWorkerHandle({4: "GPU-a", 5: "GPU-b"})),
+            _worker_info("engine-0-1", "node-b", [0], handle=UuidWorkerHandle({0: "GPU-c"})),
+        ],
+        [_worker_info("engine-1-0", "node-c", [3], handle=UuidWorkerHandle({3: "GPU-d"}))],
+    ]
+
+    engines = await hooks._compute_engine_infos(cells, worker_infos_per_cell)
+
+    assert engines == [
+        EngineInfo(
+            addr="http://a:1",
+            worker_type="decode",
+            engine_rank=0,
+            gpus=[["2001:db8::7", 4], ["2001:db8::7", 5], ["node-b", 0]],
+            gpu_uuids=["GPU-a", "GPU-b", "GPU-c"],
+        ),
+        EngineInfo(
+            addr="http://b:1",
+            worker_type="prefill",
+            engine_rank=1,
+            gpus=[["node-c", 3]],
+            gpu_uuids=["GPU-d"],
+        ),
+    ]
+
+
+async def test_compute_engine_infos_probes_workers_concurrently():
+    """The first worker's probe only finishes once the second one started, so serial probing would hang."""
+    second_started = asyncio.Event()
+    worker_infos = [
+        _worker_info("engine-0-0", "node-a", [0], handle=GatedWorkerHandle("GPU-a", wait_for=second_started)),
+        _worker_info("engine-0-1", "node-b", [1], handle=GatedWorkerHandle("GPU-b", signal=second_started)),
+    ]
+
+    engines = await asyncio.wait_for(hooks._compute_engine_infos([FakeCell("http://a:1")], [worker_infos]), timeout=5)
+
+    assert engines[0].gpu_uuids == ["GPU-a", "GPU-b"]
+
+
+async def test_register_engines_retries_after_gpu_uuid_probe_failure(monkeypatch, caplog):
+    """A failed uuid probe publishes nothing and leaves the fingerprint unset, so the next call republishes."""
+    handle = FakeHandle()
+    monkeypatch.setattr(backend, "_handle", handle)
+    probe = FlakyWorkerHandle()
+    infos_by_cell = {"inference-engine-0-0-0": [_worker_info("inference-engine-0-0-0", "node-a", [0], handle=probe)]}
+    monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(lambda: FakeManagerHandle(infos_by_cell)))
+    hooks._warner.reset_window_for_test()
+    servers = _servers([FakeCell("http://a:1")])
+
+    with caplog.at_level(logging.WARNING):
+        await hooks.register_engines(servers)
+
+    assert handle.update_topology.calls == []
+    assert hooks._engines_fingerprint is None
+    assert any("engine registration failed" in r.message for r in caplog.records)
+
+    probe.fail = False
+    await hooks.register_engines(servers)
+
+    [(args, _)] = handle.update_topology.calls
+    assert [e.addr for e in args[0].engines] == ["http://a:1"]
+
+
+async def test_register_engines_republishes_on_cell_liveness_transitions(monkeypatch):
+    """A cell that drops out and comes back must be removed from and then restored to the published topology."""
+    handle = FakeHandle()
+    monkeypatch.setattr(backend, "_handle", handle)
+    infos_by_cell = {
+        "inference-engine-0-0-0": [_worker_info("inference-engine-0-0-0", "node-a", [0])],
+        "inference-engine-0-0-1": [_worker_info("inference-engine-0-1-0", "node-b", [1])],
+    }
+    monkeypatch.setattr(RayWorkerManager, "get_handle", staticmethod(lambda: FakeManagerHandle(infos_by_cell)))
+    first, second = FakeCell("http://a:1", cell_index=0), FakeCell("http://b:1", cell_index=1)
+    servers = _servers([first, second])
+
+    await hooks.register_engines(servers)
+    second.is_pending_weights_or_serving = False
+    await hooks.register_engines(servers)
+    second.is_pending_weights_or_serving = True
+    await hooks.register_engines(servers)
+
+    assert [[e.addr for e in args[0].engines] for (args, _) in handle.update_topology.calls] == [
+        ["http://a:1", "http://b:1"],
+        ["http://a:1"],
+        ["http://a:1", "http://b:1"],
+    ]
+
+
 # ------------------------------ dashboard_log -------------------------------
 
 
@@ -320,6 +497,14 @@ def test_register_router_before_router_start_is_a_wiring_bug(monkeypatch):
         hooks.register_router(_router_args(ip=None))
 
 
+def test_register_router_without_resolvable_collector_is_noop(monkeypatch):
+    """An unreachable collector must end the hook before it asserts on the router address."""
+    monkeypatch.setattr(backend, "_handle", None)
+    monkeypatch.setattr(backend, "resolve_collector", lambda: None)
+
+    hooks.register_router(_router_args(ip=None))
+
+
 def test_register_router_without_dashboard_is_noop(monkeypatch):
     """With the dashboard off the hook returns before resolve_collector, which would block."""
     monkeypatch.setattr(backend, "resolve_collector", _never_resolve)
@@ -358,6 +543,9 @@ def test_report_data_buffer_swallows_push_failures(monkeypatch, caplog):
     handle = FakeHandle()
     handle.push_data_buffer = FakeRemoteMethod(fail=True)
     monkeypatch.setattr(backend, "_handle", handle)
+    # The module-level warner is rate limited, so an earlier warning in this
+    # process would otherwise swallow the one this test is looking for.
+    monkeypatch.setattr(hooks._warner, "_last_warn", float("-inf"))
     with caplog.at_level(logging.WARNING):
         hooks.report_data_buffer(7)  # must not raise
     assert any("data-buffer report failed" in r.message for r in caplog.records)
diff --git a/tests/fast/dashboard/test_trajectory_sink.py b/tests/fast/dashboard/test_trajectory_sink.py
index 5e259cf624d..e309ec0a504 100644
--- a/tests/fast/dashboard/test_trajectory_sink.py
+++ b/tests/fast/dashboard/test_trajectory_sink.py
@@ -68,6 +68,38 @@ def test_attempt_and_gen_events_carry_identity_and_version():
     assert _pushed(handle)[-1].detail == Sample.Status.PENDING.value
 
 
+def test_event_without_weight_version_spans_uses_empty_version():
+    """A sample whose generation reported no weight version is still emitted, with an empty version."""
+    handle = FakeHandle()
+    sink = TrajectorySink(handle)
+
+    sink.attempt_start(_sample(versions=()))
+    sink.flush()
+
+    [event] = _pushed(handle)
+    assert (event.kind, event.sample_index, event.weight_version) == (TrajectoryEventKind.ATTEMPT_START, 7, "")
+
+
+def test_event_uses_last_span_from_multi_span_generation_call():
+    """The reported version is the last span of the last call, not that call's first span."""
+    handle = FakeHandle()
+    sample = Sample(
+        index=7,
+        group_index=2,
+        weight_versions=[
+            WeightVersionsPerCall(spans=[WeightVersionSpan("3", 0, 1)]),
+            WeightVersionsPerCall(spans=[WeightVersionSpan("4", 1, 2), WeightVersionSpan("5", 2, 3)]),
+        ],
+    )
+    sink = TrajectorySink(handle)
+
+    sink.gen_start(sample)
+    sink.flush()
+
+    [event] = _pushed(handle)
+    assert event.weight_version == "5"
+
+
 def test_spans_use_explicit_timestamps_and_turns():
     handle = FakeHandle()
     sink = TrajectorySink(handle)
diff --git a/tests/fast/ray/rollout/test_inference_controller.py b/tests/fast/ray/rollout/test_inference_controller.py
index fcdaea98a63..a2f278016be 100644
--- a/tests/fast/ray/rollout/test_inference_controller.py
+++ b/tests/fast/ray/rollout/test_inference_controller.py
@@ -5,10 +5,16 @@
 from typing import Any
 
 import pytest
+from sglang.srt.constants import GPU_MEMORY_TYPE_CUDA_GRAPH, GPU_MEMORY_TYPE_KV_CACHE, GPU_MEMORY_TYPE_WEIGHTS
 from tests.fast.ray.rollout.conftest import make_args
 
+from miles.dashboard import hooks as dashboard_hooks
 from miles.ray.rollout import inference_controller as inference_controller_module
-from miles.ray.rollout.inference_controller import InferenceController, _compute_server_cell_meta_from_info
+from miles.ray.rollout.inference_controller import (
+    InferenceController,
+    UpdatableEngines,
+    _compute_server_cell_meta_from_info,
+)
 from miles.ray.rollout.rollout_server import RolloutServer
 from miles.ray.rollout.server_cell import ServerCellMetadata
 from miles.ray.specs.inference import compute_engine_pool_ids, compute_router_pool_id, specs_inference_engine
@@ -60,7 +66,14 @@ def _make_cell_meta(info: CellInfo) -> ServerCellMetadata:
 
 
 class _RecordingServer:
-    def __init__(self, server_cells: dict | None = None, *, model_name: str = "model", update_weights: bool = False):
+    def __init__(
+        self,
+        server_cells: dict | None = None,
+        *,
+        model_name: str = "model",
+        update_weights: bool = False,
+        cells_gate: asyncio.Event | None = None,
+    ):
         self.server_cells = server_cells or {}
         self.update_weights = update_weights
         self.model_name = model_name
@@ -68,12 +81,28 @@ def __init__(self, server_cells: dict | None = None, *, model_name: str = "model
         self.api_clients: list = []
         self.engine_gpu_counts: list[int] = []
         self.engine_gpu_offsets: list[int] = []
+        self.offload_tags: list = []
+        self.onload_tags: list = []
+        self.check_weights_kwargs: list[dict] = []
+        self.waited_expected_num_cells = 0
+        self.dispose_count = 0
+        self._cells_gate = cells_gate
 
     async def offload(self, tags=None):
         self.calls.append(("offload",))
+        self.offload_tags.append(tags)
+
+    async def onload(self, tags=None):
+        self.onload_tags.append(tags)
+
+    async def dispose(self):
+        self.dispose_count += 1
 
     async def check_weights(self, action, allow_quant_error=False, selector="all", skip_list=None):
         self.calls.append(("check_weights", action))
+        self.check_weights_kwargs.append(
+            dict(action=action, allow_quant_error=allow_quant_error, selector=selector, skip_list=skip_list)
+        )
         return [self.model_name]
 
     async def add_cell(self, cell_meta: ServerCellMetadata):
@@ -85,7 +114,35 @@ async def remove_cell(self, cell_id: str):
         del self.server_cells[cell_id]
 
     async def wait_expected_num_cells(self) -> None:
-        return None
+        if self._cells_gate is not None:
+            await self._cells_gate.wait()
+        self.waited_expected_num_cells += 1
+
+
+class _FakeUpdatableCell:
+    def __init__(self, workers_hash: str):
+        self.meta = SimpleNamespace(workers_hash=workers_hash)
+        self.marked_ready = 0
+        self.is_pending_weights = True
+        self.is_pending_weights_or_serving = True
+
+    async def mark_weights_ready(self) -> None:
+        self.marked_ready += 1
+
+
+class _TickingCell:
+    def __init__(self, cell_id: str = "engine-0"):
+        self.meta = SimpleNamespace(cell_id=cell_id)
+        self.tick_count = 0
+
+    async def tick(self) -> None:
+        self.tick_count += 1
+
+
+class _RecordingEvalFleet:
+    def __init__(self, args: Namespace, *, srv):
+        self.args = args
+        self.srv = srv
 
 
 def _make_controller(servers: dict) -> InferenceController:
@@ -129,6 +186,22 @@ async def test_preparing_a_rollout_resumes_probing(self):
 
         assert controller._health_checker_activeness.get().active
 
+    @pytest.mark.asyncio
+    async def test_preparing_a_rollout_awaits_the_dashboard_engine_registration(self, monkeypatch):
+        """The dashboard hook is a coroutine, so prepare_rollout must await it instead of leaving it unscheduled."""
+        awaited: list[dict] = []
+
+        async def _record(servers: dict) -> None:
+            awaited.append(servers)
+
+        monkeypatch.setattr(dashboard_hooks, "register_engines", _record)
+        servers = {"default": _RecordingServer()}
+        controller = _make_controller(servers)
+
+        await controller.prepare_rollout(rollout_id=0)
+
+        assert awaited == [servers]
+
     @pytest.mark.asyncio
     async def test_preparing_an_eval_resumes_probing(self):
         """Eval drives the same engines as a rollout does."""
@@ -231,6 +304,7 @@ def __init__(self, cell_infos: list[CellInfo]) -> None:
         self._cell_infos = cell_infos
         self._pools: list[str] = []
         self.watched_pool_ids: list[str] | None = None
+        self.stop_watch_calls = 0
 
     def created_with(self, pool_ids: list[str]) -> "_FakeWorkerProvider":
         self._pools = list(pool_ids)
@@ -244,7 +318,7 @@ async def watch_cells(self, reconcile: ReconcileFn) -> StopWatchFn:
                 await reconcile(info.cell_id, info)
 
         async def _stop_watch() -> None:
-            return None
+            self.stop_watch_calls += 1
 
         return _stop_watch
 
@@ -476,3 +550,214 @@ async def test_the_weight_checker_is_a_noop_without_an_updatable_model(self):
 
         assert await self._controller(ref).check_weights(action="compare") == []
         assert ref.calls == []
+
+    @pytest.mark.asyncio
+    async def test_check_weights_forwards_all_selection_arguments(self):
+        """Losing the selector or the skip list here would compare tensors the caller asked to leave alone."""
+        actor = _RecordingServer(model_name="actor", update_weights=True)
+
+        await self._controller(actor).check_weights(
+            action="compare", allow_quant_error=True, selector="first", skip_list=["lm_head"]
+        )
+
+        assert actor.check_weights_kwargs == [
+            dict(action="compare", allow_quant_error=True, selector="first", skip_list=["lm_head"])
+        ]
+
+
+class TestMemoryLifecycleFanOut:
+    @pytest.mark.asyncio
+    async def test_memory_lifecycle_entrypoints_fan_out_with_exact_tags(self):
+        """Every server must be told exactly which memory pools to release and to reclaim."""
+        first, second = _RecordingServer(model_name="a"), _RecordingServer(model_name="b")
+        controller = _make_controller({"a": first, "b": second})
+
+        await controller.offload(tags=[GPU_MEMORY_TYPE_KV_CACHE])
+        await controller.onload(tags=[GPU_MEMORY_TYPE_CUDA_GRAPH])
+        await controller.onload_weights()
+        await controller.onload_kv()
+
+        for srv in (first, second):
+            assert srv.offload_tags == [[GPU_MEMORY_TYPE_KV_CACHE]]
+            assert srv.onload_tags == [
+                [GPU_MEMORY_TYPE_CUDA_GRAPH],
+                [GPU_MEMORY_TYPE_WEIGHTS],
+                [GPU_MEMORY_TYPE_KV_CACHE, GPU_MEMORY_TYPE_CUDA_GRAPH],
+            ]
+
+
+class TestUpdatableEnginesPayload:
+    @pytest.mark.asyncio
+    async def test_start_update_weights_returns_clients_gpu_layout_and_generation_snapshot(self):
+        """The trainer indexes these four lists in parallel, so swapping or dropping one misplaces every shard."""
+        srv = _RecordingServer(
+            {"engine-0": _FakeUpdatableCell("hash-a"), "engine-1": _FakeUpdatableCell("hash-b")},
+            model_name="actor",
+            update_weights=True,
+        )
+        srv.api_clients = ["client-0", "client-1"]
+        srv.engine_gpu_counts = [2, 4]
+        srv.engine_gpu_offsets = [0, 2]
+        controller = _make_controller({"actor": srv, "ref": _RecordingServer(model_name="ref")})
+
+        updatable = await controller.start_update_weights()
+        await controller.end_update_weights(snapshot_cell_id_to_hashes=updatable.snapshot_cell_id_to_hashes)
+
+        assert updatable == UpdatableEngines(
+            rollout_engines=["client-0", "client-1"],
+            engine_gpu_counts=[2, 4],
+            engine_gpu_offsets=[0, 2],
+            snapshot_cell_id_to_hashes={"engine-0": "hash-a", "engine-1": "hash-b"},
+        )
+
+    @pytest.mark.asyncio
+    async def test_end_update_weights_skips_a_cell_from_a_different_worker_generation(self):
+        """A cell relaunched during the update runs new processes that never received these weights."""
+        relaunched, untouched = _FakeUpdatableCell("hash-new"), _FakeUpdatableCell("hash-b")
+        srv = _RecordingServer(
+            {"engine-0": relaunched, "engine-1": untouched}, model_name="actor", update_weights=True
+        )
+        controller = _make_controller({"actor": srv})
+
+        await controller.start_update_weights()
+        await controller.end_update_weights(snapshot_cell_id_to_hashes={"engine-0": "hash-old", "engine-1": "hash-b"})
+
+        assert (relaunched.marked_ready, untouched.marked_ready) == (0, 1)
+
+
+class TestInitLifecycle:
+    @pytest.mark.asyncio
+    async def test_debug_train_only_init_has_no_rollout_side_effects(self, monkeypatch: pytest.MonkeyPatch):
+        """A train-only debug run owns no engines, so init must not reach any rollout machinery."""
+
+        async def _no_servers(args: Namespace, **kwargs: Any) -> dict:
+            raise AssertionError("debug_train_only must not create rollout servers")
+
+        async def _no_session_server(args: Namespace) -> None:
+            raise AssertionError("debug_train_only must not wait for the session server")
+
+        monkeypatch.setattr(inference_controller_module, "create_rollout_servers", _no_servers)
+        monkeypatch.setattr(inference_controller_module, "wait_session_server_ready", _no_session_server)
+        monkeypatch.setattr(
+            inference_controller_module,
+            "RayWorkerProvider",
+            SimpleNamespace(create=lambda **kwargs: pytest.fail("debug_train_only must not watch cells")),
+        )
+        monkeypatch.setattr(
+            dashboard_hooks, "register_router", lambda args: pytest.fail("debug_train_only has no router")
+        )
+        controller = InferenceController(make_args(debug_train_only=True))
+
+        await controller.init()
+
+        assert controller.servers == {}
+        assert controller.eval_fleet is None
+        assert controller._watcher_disposers == []
+        assert controller._ticker is None
+
+    @pytest.mark.asyncio
+    async def test_init_passes_its_exact_context_lock_to_the_server_factory(self, monkeypatch: pytest.MonkeyPatch):
+        """A server built on a second lock would let engine work run inside the controller's own window."""
+        received: dict[str, Any] = {}
+
+        async def _fake_create_rollout_servers(args: Namespace, **kwargs: Any) -> dict[str, _RecordingServer]:
+            received.update(kwargs)
+            return {"default": _RecordingServer()}
+
+        monkeypatch.setattr(inference_controller_module, "create_rollout_servers", _fake_create_rollout_servers)
+        monkeypatch.setattr(
+            inference_controller_module,
+            "RayWorkerProvider",
+            SimpleNamespace(create=lambda *, pool_ids: _FakeWorkerProvider([]).created_with(pool_ids)),
+        )
+        controller = InferenceController(make_args())
+
+        await controller.init()
+        await controller.dispose()
+
+        assert received["context_lock"] is controller.context_lock
+
+    @pytest.mark.asyncio
+    async def test_init_creates_the_eval_fleet_from_the_eval_server(self, monkeypatch: pytest.MonkeyPatch):
+        """The eval fleet drives the dedicated eval engines, so it must be handed that server and no other."""
+        monkeypatch.setattr(inference_controller_module, "EvalFleet", _RecordingEvalFleet)
+        default, eval_srv = _RecordingServer(model_name="default"), _RecordingServer(model_name="eval")
+        _patch_init(monkeypatch, provider=_FakeWorkerProvider([]), servers={"default": default, "eval": eval_srv})
+        controller = InferenceController(make_args(eval_num_gpus=2))
+
+        await controller.init()
+        await controller.dispose()
+
+        assert isinstance(controller.eval_fleet, _RecordingEvalFleet)
+        assert controller.eval_fleet.srv is eval_srv
+        assert controller.eval_fleet.args is controller.args
+
+    @pytest.mark.asyncio
+    async def test_init_without_eval_gpus_creates_no_eval_fleet(self, monkeypatch: pytest.MonkeyPatch):
+        """A run without dedicated eval engines has no eval server to build a fleet from."""
+        monkeypatch.setattr(
+            inference_controller_module,
+            "EvalFleet",
+            lambda *args, **kwargs: pytest.fail("no eval fleet without eval gpus"),
+        )
+        _patch_init(monkeypatch, provider=_FakeWorkerProvider([]), servers={"default": _RecordingServer()})
+        controller = InferenceController(make_args(eval_num_gpus=0))
+
+        await controller.init()
+        await controller.dispose()
+
+        assert controller.eval_fleet is None
+
+    @pytest.mark.asyncio
+    async def test_init_registers_routing_and_waits_for_every_startup_gate(self, monkeypatch: pytest.MonkeyPatch):
+        """Returning before every server has its cells would start a rollout against engines that are not up."""
+        registered: list[Namespace] = []
+        waited_session: list[Namespace] = []
+
+        async def _wait_session_server_ready(args: Namespace) -> None:
+            waited_session.append(args)
+
+        monkeypatch.setattr(dashboard_hooks, "register_router", registered.append)
+        monkeypatch.setattr(inference_controller_module, "wait_session_server_ready", _wait_session_server_ready)
+        gate = asyncio.Event()
+        ready, blocked = _RecordingServer(), _RecordingServer(cells_gate=gate)
+        _patch_init(monkeypatch, provider=_FakeWorkerProvider([]), servers={"default": ready, "frozen": blocked})
+        args = make_args()
+        controller = InferenceController(args)
+
+        task = asyncio.create_task(controller.init())
+        for _ in range(20):
+            await asyncio.sleep(0)
+        assert not task.done()
+        assert ready.waited_expected_num_cells == 1
+        gate.set()
+        await asyncio.wait_for(task, timeout=5)
+        await controller.dispose()
+
+        assert registered == [args]
+        assert waited_session == [args]
+        assert blocked.waited_expected_num_cells == 1
+
+    @pytest.mark.asyncio
+    async def test_init_and_dispose_own_the_cell_watch_and_ticker_lifetimes(self, monkeypatch: pytest.MonkeyPatch):
+        """A watch or tick loop outliving the controller keeps dialing engines that nobody owns any more."""
+        monkeypatch.setattr(inference_controller_module, "TICK_INTERVAL_SECONDS", 0.01)
+        cell = _TickingCell()
+        provider = _FakeWorkerProvider([])
+        _patch_init(monkeypatch, provider=provider, servers={"default": _RecordingServer({"engine-0": cell})})
+        controller = InferenceController(make_args())
+
+        await controller.init()
+        await asyncio.sleep(0.05)
+        assert cell.tick_count > 0
+        assert controller._ticker._interval_seconds == inference_controller_module.TICK_INTERVAL_SECONDS
+        assert len(controller._watcher_disposers) == 1
+
+        await controller.dispose()
+        ticks_at_dispose = cell.tick_count
+        await asyncio.sleep(0.01)
+
+        assert provider.stop_watch_calls == 1
+        assert controller._watcher_disposers == []
+        assert controller._ticker is None
+        assert cell.tick_count == ticks_at_dispose
diff --git a/tests/fast/ray/rollout/test_inference_controller_tick.py b/tests/fast/ray/rollout/test_inference_controller_tick.py
index 7125318c95c..1eaca36c48c 100644
--- a/tests/fast/ray/rollout/test_inference_controller_tick.py
+++ b/tests/fast/ray/rollout/test_inference_controller_tick.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 
 from types import SimpleNamespace
 
@@ -98,6 +99,26 @@ async def test_a_cell_added_after_the_loop_started_is_picked_up(self):
 
         assert late.tick_count > 0
 
+    async def test_tick_cells_logs_each_failure_with_cell_id_and_traceback(self, caplog, monkeypatch):
+        """The log line is the only signal that a cell is stuck, and it has to name the cell and carry the cause."""
+        error = RuntimeError("cell exploded")
+        broken = _RecordingCell(error=error, cell_id="broken")
+        wedged = _RecordingCell(delay=60.0, cell_id="wedged")
+        controller = _make_controller({"default": _StubServer({"a": broken, "b": wedged})})
+        monkeypatch.setattr(inference_controller_module, "CELL_TICK_TIMEOUT_SECONDS", 0.01)
+
+        with caplog.at_level(logging.ERROR):
+            await controller._tick_cells()
+
+        failures = [record for record in caplog.records if record.message.startswith("Ticking cell ")]
+        assert [record.message for record in failures] == [
+            "Ticking cell broken failed",
+            "Ticking cell wedged failed",
+        ]
+        assert failures[0].exc_info[1] is error
+        assert isinstance(failures[1].exc_info[1], TimeoutError)
+        assert all(record.exc_info[2] is not None for record in failures)
+
     async def test_the_sweep_keeps_running_after_one_cell_raises(self):
         """One wedged engine must not stop every other cell from making progress."""
         broken, healthy = _RecordingCell(error=RuntimeError("cell exploded")), _RecordingCell()
diff --git a/tests/fast/ray/train/conftest.py b/tests/fast/ray/train/conftest.py
index 18cf5e890d5..139f301299c 100644
--- a/tests/fast/ray/train/conftest.py
+++ b/tests/fast/ray/train/conftest.py
@@ -47,6 +47,10 @@ async def _retry_without_sleeping(fn: Callable[[int], Awaitable[Any]], **kwargs:
     monkeypatch.setattr(group_module, "retry", _retry_without_sleeping)
 
 
+def get_raw_actor_handles(cell: RayTrainCell) -> list[ray.actor.ActorHandle]:
+    return [handle._actor_handle for handle in cell._get_worker_handles()]
+
+
 def make_indep_dp_info(
     *,
     cell_index: int = 0,
diff --git a/tests/fast/ray/train/fake_worker_manager.py b/tests/fast/ray/train/fake_worker_manager.py
index 94e16505458..b25217e9f25 100644
--- a/tests/fast/ray/train/fake_worker_manager.py
+++ b/tests/fast/ray/train/fake_worker_manager.py
@@ -67,7 +67,7 @@ def _get_worker_infos(self, cell_id: str) -> list[WorkerInfo]:
                 generation=1 + len(self.started_cell_ids),
                 self_addrs={MASTER_PORT_NAME: HostAndPort(host="10.0.0.1", port=20000)},
                 gpu_ids=[worker_index],
-                actor_handle=handle,
+                handle=RayWorkerHandle(handle),
             )
             for worker_index, handle in enumerate(self._handles[cell_id])
         ]
diff --git a/tests/fast/ray/train/test_cell.py b/tests/fast/ray/train/test_cell.py
index 999870bc2e6..6b345b858ab 100644
--- a/tests/fast/ray/train/test_cell.py
+++ b/tests/fast/ray/train/test_cell.py
@@ -1,13 +1,11 @@
 import asyncio
-import logging
-from types import SimpleNamespace
 
 import pytest
 import ray
 from tests.fast.ray.train import conftest as train_conftest
-from tests.fast.ray.train.conftest import make_alive_cell, make_cell, make_indep_dp_info
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_alive_cell, make_cell, make_indep_dp_info
 
-from miles.ray.train import cell as cell_module
+from miles.utils.workers.worker_handle import BaseWorkerHandle
 
 pytestmark = pytest.mark.asyncio
 
@@ -21,19 +19,20 @@ def test_starts_as_uninitialized_after_init(self):
         assert not cell.is_alive
         assert cell.is_uninitialized
 
-    def test_actor_handles_are_real_ray_actors(self):
+    def test_worker_handles_wrap_real_ray_actors(self):
         cell = make_cell(actor_count=3)
 
-        handles = cell._get_actor_handles()
+        handles = cell._get_worker_handles()
         assert len(handles) == 3
-        assert all(isinstance(h, ray.actor.ActorHandle) for h in handles)
+        assert all(isinstance(h, BaseWorkerHandle) for h in handles)
+        assert all(isinstance(h, ray.actor.ActorHandle) for h in get_raw_actor_handles(cell))
 
 
 class TestKillWorkers:
     async def test_killing_reaches_every_worker(self):
         """The dead workers must not linger in a cross-cell collective."""
         cell = make_alive_cell(0, alive_cell_indices=[0])
-        handles = cell._get_actor_handles()
+        handles = get_raw_actor_handles(cell)
 
         await cell._kill_workers_and_confirm_dead()
 
@@ -52,14 +51,14 @@ async def test_killing_does_not_involve_the_worker_manager(self):
     async def test_stop_kills_the_underlying_workers(self):
         """Stopping a cell really kills its workers, so every handle is confirmed dead and rejects new calls."""
         cell = make_cell(actor_count=2)
-        handles = cell._get_actor_handles()
+        wrapped_handles = cell._get_worker_handles()
 
         await cell._kill_workers_and_confirm_dead()
 
-        for handle in handles:
-            await asyncio.wait_for(cell_module._confirm_actor_dead(handle), timeout=30.0)
+        for wrapped in wrapped_handles:
+            await asyncio.wait_for(wrapped.wait_dead(timeout=30.0), timeout=35.0)
             with pytest.raises(ray.exceptions.RayActorError):
-                ray.get(handle.get_calls.remote())
+                ray.get(wrapped._actor_handle.get_calls.remote())
 
 
 class TestMarkAsAlive:
@@ -74,11 +73,11 @@ def test_transitions_uninitialized_to_alive(self):
 
     def test_preserves_actor_handles(self):
         cell = make_cell(actor_count=3)
-        handles_before = cell._get_actor_handles()
+        handles_before = get_raw_actor_handles(cell)
 
         cell._mark_as_alive(indep_dp_info=make_indep_dp_info())
 
-        assert cell._get_actor_handles() == handles_before
+        assert get_raw_actor_handles(cell) == handles_before
 
     def test_rejects_from_alive(self):
         cell = make_alive_cell(0, alive_cell_indices=[0])
@@ -98,11 +97,11 @@ def test_updates_stored_info(self):
 
     def test_preserves_actor_handles(self):
         cell = make_alive_cell(0, alive_cell_indices=[0])
-        handles = cell._get_actor_handles()
+        handles = get_raw_actor_handles(cell)
 
         cell._update_indep_dp_info(make_indep_dp_info(quorum_id=5))
 
-        assert cell._get_actor_handles() == handles
+        assert get_raw_actor_handles(cell) == handles
 
     def test_rejects_from_uninitialized(self):
         cell = make_cell()
@@ -147,7 +146,7 @@ async def test_kill_from_errored_reaches_the_workers(self):
         cell = make_alive_cell(0, alive_cell_indices=[0])
         cell._mark_as_errored()
         assert cell.is_errored
-        handles = cell._get_actor_handles()
+        handles = get_raw_actor_handles(cell)
 
         await cell._kill_workers_and_confirm_dead()
 
@@ -180,7 +179,7 @@ async def test_dispatches_init_and_marks_alive(self):
         assert cell.is_alive
         assert cell.indep_dp_info == info
 
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             calls = ray.get(handle.get_calls.remote())
             assert [name for name, _args, _kwargs in calls] == ["configure_master_addr_and_port", "init"]
             kwargs = calls[1][2]
@@ -192,14 +191,14 @@ class TestAsyncInitFailure:
     async def test_init_failure_leaves_cell_not_alive(self):
         """A failed remote init marks the cell errored and tears it down; it is never reported alive."""
         cell = make_cell(actor_count=1)
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             ray.get(handle.set_fail_methods.remote(["init"]))
 
         with pytest.raises(RuntimeError, match="Injected failure"):
             await cell.init(indep_dp_info=make_indep_dp_info())
 
         assert not cell.is_alive
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             with pytest.raises(ray.exceptions.RayActorError):
                 ray.get(handle.get_calls.remote())
 
@@ -214,7 +213,7 @@ async def test_reconfigure_and_update_info(self):
         assert cell.indep_dp_info == new_info
         assert cell.is_alive
 
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             calls = ray.get(handle.get_calls.remote())
             reconfig_calls = [c for c in calls if c[0] == "reconfigure_indep_dp"]
             assert len(reconfig_calls) == 1
@@ -226,7 +225,7 @@ async def test_sends_ckpt_to_correct_dst_ranks(self):
         new_info = make_indep_dp_info(alive_cell_indices=[0, 1, 2], quorum_id=2)
         await cell.prepare_indep_dp_mode_alive(indep_dp_info=new_info, send_ckpt_dst_ranks=[1, 2])
 
-        handle = cell._get_actor_handles()[0]
+        handle = get_raw_actor_handles(cell)[0]
         calls = ray.get(handle.get_calls.remote())
         send_calls = [c for c in calls if c[0] == "send_ckpt"]
         assert len(send_calls) == 2
@@ -244,7 +243,7 @@ async def test_healing_inits_and_marks_alive(self):
         assert cell.is_alive
         assert cell.indep_dp_info == info
 
-        handle = cell._get_actor_handles()[0]
+        handle = get_raw_actor_handles(cell)[0]
         calls = ray.get(handle.get_calls.remote())
         assert any(c[0] == "init" for c in calls)
 
@@ -324,116 +323,3 @@ async def test_full_kill_and_replacement_cycle(self):
         cell._mark_as_alive(indep_dp_info=info_v2)
         assert cell.is_alive
         assert cell.indep_dp_info.quorum_id == 2
-
-
-def _make_coro_factory(behavior):
-    async def _coro():
-        return behavior()
-
-    return _coro
-
-
-def _raise_factory(exc):
-    async def _coro():
-        raise exc
-
-    return _coro
-
-
-class _FakeReadyMethod:
-    def __init__(self, coro_factories):
-        self._coro_factories = list(coro_factories)
-        self.call_count = 0
-
-    def remote(self):
-        self.call_count += 1
-        index = min(self.call_count - 1, len(self._coro_factories) - 1)
-        return self._coro_factories[index]()
-
-
-def _make_fake_handle(coro_factories):
-    ready = _FakeReadyMethod(coro_factories)
-    handle = SimpleNamespace()
-    handle.__ray_ready__ = ready
-    return handle, ready
-
-
-def _ray_actor_error():
-    return ray.exceptions.RayActorError()
-
-
-def _ray_task_error():
-    return ray.exceptions.RayTaskError.__new__(ray.exceptions.RayTaskError)
-
-
-class TestConfirmActorDead:
-    async def test_returns_immediately_when_actor_error_on_first_probe(self):
-        """A dead actor whose first probe raises RayActorError is confirmed dead after one probe."""
-        handle, ready = _make_fake_handle([_raise_factory(_ray_actor_error())])
-
-        await cell_module._confirm_actor_dead(handle)
-
-        assert ready.call_count == 1
-
-    async def test_returns_immediately_when_task_error_on_first_probe(self):
-        """RayTaskError on the first probe is also treated as confirmed actor death."""
-        handle, ready = _make_fake_handle([_raise_factory(_ray_task_error())])
-
-        await cell_module._confirm_actor_dead(handle)
-
-        assert ready.call_count == 1
-
-    async def test_retries_after_timeout_then_confirms_death(self, monkeypatch):
-        """A probe timeout is tolerated; the loop retries and confirms death on the next probe."""
-        slept = []
-
-        async def _noop_sleep(seconds):
-            slept.append(seconds)
-
-        monkeypatch.setattr(cell_module.asyncio, "sleep", _noop_sleep)
-        monkeypatch.setattr(cell_module, "time", SimpleNamespace(monotonic=_make_monotonic([0.0, 1.0])))
-
-        handle, ready = _make_fake_handle(
-            [
-                _raise_factory(asyncio.TimeoutError()),
-                _raise_factory(_ray_actor_error()),
-            ]
-        )
-
-        await cell_module._confirm_actor_dead(handle)
-
-        assert ready.call_count == 2
-        assert slept == [1.0]
-
-    async def test_deadline_reached_returns_and_logs_error(self, monkeypatch, caplog):
-        """When the timeout deadline is exceeded after a hung probe, it returns and logs an ERROR."""
-
-        async def _noop_sleep(seconds):
-            return None
-
-        monkeypatch.setattr(cell_module.asyncio, "sleep", _noop_sleep)
-        monkeypatch.setattr(cell_module, "time", SimpleNamespace(monotonic=_make_monotonic([0.0, 200.0])))
-
-        handle, ready = _make_fake_handle([_raise_factory(asyncio.TimeoutError())])
-
-        with caplog.at_level(logging.ERROR, logger="miles.ray.train.cell"):
-            await cell_module._confirm_actor_dead(handle)
-
-        assert ready.call_count == 1
-        error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
-        assert len(error_records) == 1
-        assert "Timed out after 120s confirming actor death" in error_records[0].getMessage()
-
-
-def _make_monotonic(values):
-    seq = list(values)
-    state = {"i": 0}
-
-    def _monotonic():
-        i = state["i"]
-        if i < len(seq):
-            state["i"] = i + 1
-            return seq[i]
-        return seq[-1]
-
-    return _monotonic
diff --git a/tests/fast/ray/train/test_cell_master_addr.py b/tests/fast/ray/train/test_cell_master_addr.py
index 74aa6339aa1..b9318607b56 100644
--- a/tests/fast/ray/train/test_cell_master_addr.py
+++ b/tests/fast/ray/train/test_cell_master_addr.py
@@ -1,6 +1,6 @@
 import pytest
 import ray
-from tests.fast.ray.train.conftest import make_cell, make_indep_dp_info
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_cell, make_indep_dp_info
 
 pytestmark = pytest.mark.asyncio
 
@@ -8,7 +8,7 @@
 def _calls_of(cell, method: str) -> list:
     return [
         [call for call in ray.get(handle.get_calls.remote()) if call[0] == method]
-        for handle in cell._get_actor_handles()
+        for handle in get_raw_actor_handles(cell)
     ]
 
 
@@ -28,5 +28,5 @@ async def test_the_master_address_is_configured_before_the_process_group_is_buil
 
         await cell.init(indep_dp_info=make_indep_dp_info(quorum_id=0))
 
-        methods = [call[0] for call in ray.get(cell._get_actor_handles()[0].get_calls.remote())]
+        methods = [call[0] for call in ray.get(get_raw_actor_handles(cell)[0].get_calls.remote())]
         assert methods.index("configure_master_addr_and_port") < methods.index("init")
diff --git a/tests/fast/ray/train/test_cell_monitor.py b/tests/fast/ray/train/test_cell_monitor.py
index c0d9ea8e54e..b44da1998f2 100644
--- a/tests/fast/ray/train/test_cell_monitor.py
+++ b/tests/fast/ray/train/test_cell_monitor.py
@@ -1,17 +1,17 @@
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
-import ray
 
 from miles.ray.train.cell_monitor import compute_cell_status, create_trainer_cell_health_checker
 from miles.ray.train.cell_state import StateAllocatedAlive, StateAllocatedErrored, StateAllocatedUninitialized
 from miles.utils.ft_utils.api_server.models import TriState
 from miles.utils.ft_utils.health_checker import ActiveAndEpoch, SimpleHealthCheckerConfig
 from miles.utils.ft_utils.indep_dp import IndepDPInfo
+from miles.utils.workers.worker_handle import BaseWorkerHandle, WorkerUnreachableError
 
 
-def _make_actor_handle_mock() -> MagicMock:
-    return MagicMock(spec=ray.actor.ActorHandle)
+def _make_worker_handle_mock() -> MagicMock:
+    return MagicMock(spec=BaseWorkerHandle)
 
 
 def _make_indep_dp_info() -> IndepDPInfo:
@@ -26,7 +26,7 @@ def _make_indep_dp_info() -> IndepDPInfo:
 
 
 def _make_alive_state() -> StateAllocatedAlive:
-    return StateAllocatedAlive(actor_handles=[_make_actor_handle_mock()], indep_dp_info=_make_indep_dp_info())
+    return StateAllocatedAlive(worker_handles=[_make_worker_handle_mock()], indep_dp_info=_make_indep_dp_info())
 
 
 def _find_condition(status, type_: str):
@@ -64,7 +64,7 @@ def test_health_unknown_reports_healthy_unknown_not_translated_to_true(self):
 class TestComputeCellStatusOtherStates:
     @pytest.mark.parametrize("health_status", [TriState.TRUE, TriState.FALSE, TriState.UNKNOWN])
     def test_uninitialized_ignores_health_checker(self, health_status: TriState):
-        state = StateAllocatedUninitialized(actor_handles=[_make_actor_handle_mock()])
+        state = StateAllocatedUninitialized(worker_handles=[_make_worker_handle_mock()])
 
         result = compute_cell_status(state, health_status)
 
@@ -74,7 +74,7 @@ def test_uninitialized_ignores_health_checker(self, health_status: TriState):
 
     @pytest.mark.parametrize("health_status", [TriState.TRUE, TriState.FALSE, TriState.UNKNOWN])
     def test_errored_always_reports_unhealthy(self, health_status: TriState):
-        state = StateAllocatedErrored(actor_handles=[_make_actor_handle_mock()], indep_dp_info=_make_indep_dp_info())
+        state = StateAllocatedErrored(worker_handles=[_make_worker_handle_mock()], indep_dp_info=_make_indep_dp_info())
 
         result = compute_cell_status(state, health_status)
 
@@ -115,7 +115,7 @@ async def test_rpc_returns_means_healthy_regardless_of_progress(self):
     async def test_rpc_error_propagates_as_unhealthy(self):
         """A dead actor makes the heartbeat RPC raise; _check propagates it so the
         checker reports unhealthy."""
-        execute = AsyncMock(side_effect=ray.exceptions.RayActorError())
+        execute = AsyncMock(side_effect=WorkerUnreachableError("worker gone"))
         cell = _make_cell_mock(is_alive=True, execute=execute)
 
         checker = create_trainer_cell_health_checker(
@@ -124,7 +124,7 @@ async def test_rpc_error_propagates_as_unhealthy(self):
             get_activeness=lambda: ActiveAndEpoch(active=True, epoch=0),
         )
 
-        with pytest.raises(ray.exceptions.RayActorError):
+        with pytest.raises(WorkerUnreachableError):
             await checker._check_fn()
 
     @pytest.mark.asyncio
diff --git a/tests/fast/ray/train/test_cell_restart.py b/tests/fast/ray/train/test_cell_restart.py
index bc7629518d6..77882cf5b17 100644
--- a/tests/fast/ray/train/test_cell_restart.py
+++ b/tests/fast/ray/train/test_cell_restart.py
@@ -3,7 +3,7 @@
 import pytest
 import ray
 from tests.fast.ray.train import conftest as train_conftest
-from tests.fast.ray.train.conftest import make_cell
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_cell
 
 from miles.ray.train import cell as cell_module
 
@@ -29,7 +29,7 @@ class TestCellKillAndRestart:
     async def test_killing_a_failed_cell_reaches_the_workers_directly(self):
         """Waiting for an external controller would leave the other cells hanging in NCCL."""
         cell = make_cell(2)
-        handles = cell._get_actor_handles()
+        handles = get_raw_actor_handles(cell)
 
         await cell._kill_workers_and_confirm_dead()
 
@@ -41,13 +41,13 @@ async def test_killing_a_failed_cell_reaches_the_workers_directly(self):
     async def test_a_replacement_cell_picks_up_the_fresh_actor_handles(self):
         """Reusing the dead handles would make every later call fail."""
         cell = make_cell(0)
-        old_handles = cell._get_actor_handles()
+        old_handles = get_raw_actor_handles(cell)
         await cell._kill_workers_and_confirm_dead()
 
         train_conftest.fake_worker_manager._stop_cells([cell.cell_id])
         replacement = make_cell(0)
 
-        assert replacement._get_actor_handles() != old_handles
+        assert get_raw_actor_handles(replacement) != old_handles
 
     async def test_killing_twice_is_harmless(self):
         """Healing may tear down an already dead cell, which must not raise."""
diff --git a/tests/fast/ray/train/test_group.py b/tests/fast/ray/train/test_group.py
index f53b768a2b3..5367588645b 100644
--- a/tests/fast/ray/train/test_group.py
+++ b/tests/fast/ray/train/test_group.py
@@ -5,6 +5,7 @@
 import pytest
 import ray
 from tests.fast.ray.train import conftest as train_conftest
+from tests.fast.ray.train.conftest import get_raw_actor_handles
 
 from miles.backends.megatron_utils.ft.types import TrainStepOutcome, TrainStepOutput
 from miles.ray.train.group import RayTrainGroup
@@ -64,38 +65,41 @@ def _make_controller(
     group = RayTrainGroup(
         args=_make_mock_args(indep_dp=True, gpus_per_cell=actor_count_per_cell, num_cells=num_cells),
         role="actor",
+        with_ref=False,
         inference_controller=inference_controller,
         rollout_executor=rollout_executor,
     )
     for cell_index in range(num_cells):
-        cell = group._create_cell(f"{group._pool_id}-{cell_index}", cell_index=cell_index, workers_hash="pseudo-hash-1")
+        cell = group._create_cell(
+            f"{group._pool_id}-{cell_index}", cell_index=cell_index, workers_hash="pseudo-hash-1"
+        )
         group._cells_by_id[cell.cell_id] = cell
     return group
 
 
 async def _stop_cell(group: RayTrainGroup, cell_index: int) -> None:
     """Suspension stops the cell in the manager; reconcile then drops it from the bookkeeping."""
-    cell_id = f"{group._pool}-{cell_index}"
+    cell_id = f"{group._pool_id}-{cell_index}"
     train_conftest.fake_worker_manager._stop_cells([cell_id])
     await group._reconcile(cell_id, None)
 
 
 def _cell(group: RayTrainGroup, cell_index: int) -> object:
-    return group._cells_by_id[f"{group._pool}-{cell_index}"]
+    return group._cells_by_id[f"{group._pool_id}-{cell_index}"]
 
 
 def _start_cell(group: RayTrainGroup, cell_index: int) -> None:
     """The manager relaunches the cell, so reconcile hands the controller a fresh object."""
-    cell_id = f"{group._pool}-{cell_index}"
+    cell_id = f"{group._pool_id}-{cell_index}"
     group._cells_by_id[cell_id] = group._create_cell(cell_id, cell_index=cell_index, workers_hash="pseudo-hash-2")
 
 
 def _was_stopped(group: RayTrainGroup, cell_index: int) -> bool:
-    return [f"{group._pool}-{cell_index}"] in train_conftest.fake_worker_manager.stopped_cell_ids
+    return [f"{group._pool_id}-{cell_index}"] in train_conftest.fake_worker_manager.stopped_cell_ids
 
 
 def _was_killed(group: RayTrainGroup, cell_index: int) -> bool:
-    for handle in _cell(group, cell_index)._get_actor_handles():
+    for handle in get_raw_actor_handles(_cell(group, cell_index)):
         try:
             ray.get(handle.get_calls.remote())
             return False
@@ -133,7 +137,7 @@ def test_cells_are_allocated_after_init(self):
     def test_each_cell_has_own_actors(self):
         group = _make_controller(num_cells=3, actor_count_per_cell=2)
 
-        handles_per_cell = [cell._get_actor_handles() for cell in group._cells]
+        handles_per_cell = [get_raw_actor_handles(cell) for cell in group._cells]
         assert all(len(h) == 2 for h in handles_per_cell)
 
         all_handles = [h for handles in handles_per_cell for h in handles]
@@ -209,12 +213,12 @@ async def test_picks_first_alive_cell(self):
 
         await group._execute_first_alive("save_model", rollout_id=42)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             calls = ray.get(handle.get_calls.remote())
             assert any(c[0] == "save_model" for c in calls)
 
         for cell in group._cells[1:]:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 assert not any(c[0] == "save_model" for c in calls)
 
@@ -224,7 +228,7 @@ async def test_skips_errored_picks_next(self):
 
         await group._execute_first_alive("update_weights")
 
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             calls = ray.get(handle.get_calls.remote())
             assert any(c[0] == "update_weights" for c in calls)
 
@@ -255,7 +259,7 @@ async def test_skips_errored_cells(self):
 
         await group._execute_all_alive_and_catch("train")
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             calls = ray.get(handle.get_calls.remote())
             assert any(c[0] == "train" for c in calls)
 
@@ -296,7 +300,7 @@ async def test_reconfigure_triggers_on_alive_change(self):
 
         # Step 6: Actors received reconfigure_indep_dp
         for cell in [_cell(group, 0), _cell(group, 2)]:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 assert any(c[0] == "reconfigure_indep_dp" for c in calls)
 
@@ -329,12 +333,12 @@ async def test_pending_cell_gets_healed(self):
             assert cell.indep_dp_info.alive_size == 3
 
         # Step 5: Healed cell's actors received init
-        for handle in _cell(group, 2)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 2)):
             calls = ray.get(handle.get_calls.remote())
             assert any(c[0] == "init" for c in calls)
 
         # Step 6: Source cell sent ckpt to healed cell's alive_rank
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             calls = ray.get(handle.get_calls.remote())
             send_calls = [c for c in calls if c[0] == "send_ckpt"]
             assert len(send_calls) == 1
@@ -355,7 +359,7 @@ async def test_multiple_pending_cells_healed(self):
             assert cell.indep_dp_info.alive_cell_indices == [0, 1, 2]
 
         # Source (cell 0) sent ckpt to both healed cells
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             calls = ray.get(handle.get_calls.remote())
             send_calls = [c for c in calls if c[0] == "send_ckpt"]
             assert len(send_calls) == 2
@@ -371,7 +375,7 @@ async def test_healed_cell_receives_set_rollout_executor(self):
         await group._refresh_cells(rollout_id=0)
 
         assert _cell(group, 1).is_alive
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             calls = ray.get(handle.get_calls.remote())
             set_calls = [c for c in calls if c[0] == "set_rollout_executor"]
             assert set_calls, "healed cell never received set_rollout_executor"
@@ -469,7 +473,7 @@ async def test_repeated_refresh_without_change_does_not_reconfigure(self):
         # Clear init calls by noting current call count
         init_call_counts = {}
         for cell in group._cells:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 init_call_counts[id(handle)] = len(calls)
 
@@ -480,7 +484,7 @@ async def test_repeated_refresh_without_change_does_not_reconfigure(self):
 
         # No new calls dispatched
         for cell in group._cells:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 assert len(calls) == init_call_counts[id(handle)]
 
@@ -530,7 +534,7 @@ async def test_train_refreshes_and_dispatches(self):
         await group.train(rollout_id=0, rollout_data_pack=_DUMMY_DATA_PACK)
 
         for cell in group._cells:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 assert any(c[0] == "train" for c in calls)
 
@@ -541,7 +545,7 @@ async def test_train_with_stopped_cell_only_dispatches_to_alive(self):
         await group.train(rollout_id=0, rollout_data_pack=_DUMMY_DATA_PACK)
 
         for cell in [_cell(group, 0), _cell(group, 2)]:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 assert any(c[0] == "train" for c in calls)
 
@@ -554,7 +558,7 @@ async def test_consecutive_train_no_reconfigure_overhead(self):
         # Note init call count
         init_counts = {}
         for cell in group._cells:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 init_counts[id(handle)] = len(ray.get(handle.get_calls.remote()))
 
         for step in range(3):
@@ -563,7 +567,7 @@ async def test_consecutive_train_no_reconfigure_overhead(self):
         assert group._indep_dp_quorum_id == 0
 
         for cell in group._cells:
-            for handle in cell._get_actor_handles():
+            for handle in get_raw_actor_handles(cell):
                 calls = ray.get(handle.get_calls.remote())
                 new_calls = calls[init_counts[id(handle)] :]
                 assert not any(c[0] == "reconfigure_indep_dp" for c in new_calls)
@@ -619,7 +623,7 @@ async def test_one_cell_failure_marks_errored_others_ok(self):
         group = await _make_alive_controller(num_cells=3)
 
         # Step 1: Make cell 1's actors fail on train
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             ray.get(handle.set_fail_methods.remote(["train"]))
 
         # Step 2: Broadcast train
@@ -632,7 +636,7 @@ async def test_one_cell_failure_marks_errored_others_ok(self):
 
         # Step 4: Other cells received train call
         for cell_idx in [0, 2]:
-            for handle in _cell(group, cell_idx)._get_actor_handles():
+            for handle in get_raw_actor_handles(_cell(group, cell_idx)):
                 calls = ray.get(handle.get_calls.remote())
                 assert any(c[0] == "train" for c in calls)
 
@@ -641,7 +645,7 @@ async def test_errored_cell_skipped_in_next_broadcast(self):
         group = await _make_alive_controller(num_cells=2)
 
         # Step 1: Make cell 0 fail
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["train"]))
 
         await group._execute_all_alive_and_catch("train", rollout_id=0, rollout_data_ref="data")
@@ -650,7 +654,7 @@ async def test_errored_cell_skipped_in_next_broadcast(self):
         # Step 2: Next broadcast only goes to cell 1
         await group._execute_all_alive_and_catch("train", rollout_id=1, rollout_data_ref="data")
 
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             calls = ray.get(handle.get_calls.remote())
             train_calls = [c for c in calls if c[0] == "train"]
             assert len(train_calls) == 2
@@ -662,7 +666,7 @@ async def test_first_cell_fails_retry_falls_back_to_next(self):
         group = await _make_alive_controller(num_cells=3)
 
         # Step 1: Make cell 0 fail on save_model
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["save_model"]))
 
         # Step 2: save_model uses retry(lambda _: self._execute_first_alive(...))
@@ -672,7 +676,7 @@ async def test_first_cell_fails_retry_falls_back_to_next(self):
         assert _was_killed(group, 0)
         assert _cell(group, 1).is_alive
 
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             calls = ray.get(handle.get_calls.remote())
             assert any(c[0] == "save_model" for c in calls)
 
@@ -680,7 +684,7 @@ async def test_single_execute_first_alive_raises_on_failure(self):
         """A single _execute_first_alive call raises (no retry) when the first cell fails."""
         group = await _make_alive_controller(num_cells=2)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["save_model"]))
 
         with pytest.raises(Exception):  # noqa: B017
@@ -692,7 +696,7 @@ async def test_losing_the_last_cell_keeps_the_worker_error_as_the_cause(self):
         """Without the cause the driver traceback says nothing about why the last cell died."""
         group = await _make_alive_controller(num_cells=1)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["save_model"]))
 
         with pytest.raises(NonRetryableError) as excinfo:
@@ -706,7 +710,7 @@ async def test_losing_the_last_cell_stays_retryable_while_a_cell_is_still_healin
         await _stop_cell(group, 1)
         _start_cell(group, 1)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["save_model"]))
 
         with pytest.raises(Exception) as excinfo:  # noqa: B017
@@ -719,7 +723,7 @@ async def test_terminal_failure_does_not_burn_another_backoff(self):
         """Retrying without a single live cell can never succeed, so it must fail fast."""
         group = await _make_alive_controller(num_cells=1)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_fail_methods.remote(["save_model"]))
 
         attempts = 0
@@ -754,10 +758,12 @@ async def test_healing_failure_marks_pending_cell_errored_keeps_alive(self):
         # Step 3: Refresh — healing init fails, cell auto-marks errored
         await group._refresh_cells(rollout_id=0)
 
-        # Step 4: Cell 2 errored, cells 0 and 1 still alive
+        # Step 4: Cell 2 errored, cells 0 and 1 still alive. _was_stopped would already be true
+        # from step 1, so it says nothing about the healing failure; the kill does.
         assert _cell(group, 0).is_alive
         assert _cell(group, 1).is_alive
-        assert _was_stopped(group, 2)
+        assert _cell(group, 2).is_errored
+        assert _was_killed(group, 2)
 
 
 class TestHeartbeatMonitor:
@@ -779,7 +785,7 @@ async def test_heartbeat_stale_timestamp_does_not_mark_errored(self):
 
         # Drive cell 1's last-active timestamp to the epoch (maximally stale); the
         # liveness check must ignore staleness while the heartbeat RPC keeps returning.
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             ray.get(handle.set_last_active_timestamp.remote(0.0))
 
         # Neither check raises (a returned heartbeat proves the process is alive) and
@@ -792,7 +798,7 @@ async def test_heartbeat_timeout_marks_errored(self):
         """When heartbeat call fails (actor unresponsive), cell is marked errored."""
         group = await _make_alive_controller(num_cells=2)
 
-        for handle in _cell(group, 0)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 0)):
             ray.get(handle.set_heartbeat_fail.remote(True))
 
         with pytest.raises(RuntimeError, match="Injected heartbeat failure"):
@@ -870,8 +876,9 @@ def test_retry_when_discarded_exists(self, results):
         ],
     )
     def test_raises_when_all_cells_errored(self, results):
-        """A freshly built group has no alive or pending cell, so an all-errored attempt is non-retryable."""
-        with pytest.raises(NonRetryableError, match="All cells failed"):
+        """Every cell failing this attempt still leaves an unstarted cell that can be healed into
+        the next one, so the controller must raise the retryable error rather than the fatal one."""
+        with pytest.raises(RuntimeError, match="All cells failed"):
             _make_controller(num_cells=1)._check_train_one_attempt(_alive_cells_for(results), results)
 
     def test_compute_attempt_outcomes_buckets_cells_by_index(self):
@@ -889,19 +896,19 @@ def test_a_payload_carrying_output_is_bucketed_by_its_outcome(self):
 
 async def _set_all_train_return(group: RayTrainGroup, value: TrainStepOutput) -> None:
     for cell in group._cells:
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             ray.get(handle.set_train_return_value.remote(value))
 
 
 async def _set_all_train_returns_per_attempt(group: RayTrainGroup, values: list[TrainStepOutput]) -> None:
     for cell in group._cells:
-        for handle in cell._get_actor_handles():
+        for handle in get_raw_actor_handles(cell):
             ray.get(handle.set_train_return_values_per_attempt.remote(values))
 
 
 def _count_train_calls(group: RayTrainGroup, cell_index: int) -> int:
     total = 0
-    for handle in _cell(group, cell_index)._get_actor_handles():
+    for handle in get_raw_actor_handles(_cell(group, cell_index)):
         calls = ray.get(handle.get_calls.remote())
         total += sum(1 for c in calls if c[0] == "train")
     return total
@@ -947,7 +954,7 @@ async def test_cell_errored_does_not_retry_when_others_normal(self):
         group = await _make_alive_controller(num_cells=3)
 
         # Step 1: Make cell 1 fail (exception)
-        for handle in _cell(group, 1)._get_actor_handles():
+        for handle in get_raw_actor_handles(_cell(group, 1)):
             ray.get(handle.set_fail_methods.remote(["train"]))
 
         # Step 2: Train completes without retry (cell 1 errored but others NORMAL)
@@ -1116,7 +1123,7 @@ def test_a_cell_removed_while_the_statuses_are_read_does_not_abort_the_read(self
         and iterating the live dict raises RuntimeError instead of answering the request."""
         controller = _make_controller(num_cells=3)
         victim = f"{controller._pool_id}-1"
-        real_cell = controller._cells_by_id[f"{controller._pool_id}-0"]
+        real_cell = _cell(controller, 0)
 
         class _EvictingCell:
             def cell_status(self_inner):
diff --git a/tests/fast/ray/train/test_group_failure_reporting.py b/tests/fast/ray/train/test_group_failure_reporting.py
index f27c30d1bc1..8faa7d4498e 100644
--- a/tests/fast/ray/train/test_group_failure_reporting.py
+++ b/tests/fast/ray/train/test_group_failure_reporting.py
@@ -2,7 +2,7 @@
 
 import pytest
 import ray
-from tests.fast.ray.train.conftest import make_alive_cell, make_cell
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_alive_cell, make_cell
 
 from miles.ray.train.group import TrainerController
 from miles.utils.ft_utils.health_checker import ActivenessTracker
@@ -31,7 +31,7 @@ def _make_controller(cells: list) -> RayTrainGroup:
 
 def _make_failing_controller(fn_name: str) -> RayTrainGroup:
     cell = make_alive_cell(0, alive_cell_indices=[0])
-    for handle in cell._get_actor_handles():
+    for handle in get_raw_actor_handles(cell):
         ray.get(handle.set_fail_methods.remote([fn_name]))
     return _make_controller([cell])
 
@@ -108,7 +108,7 @@ async def test_a_failed_attempt_is_fatal_once_no_cell_can_come_back(self):
     async def test_offload_tolerates_losing_the_last_alive_cell_while_a_cell_is_still_healing(self):
         """A healing cell keeps the group recoverable, so the lifecycle call must not raise at all."""
         alive_cell = make_alive_cell(0, alive_cell_indices=[0])
-        for handle in alive_cell._get_actor_handles():
+        for handle in get_raw_actor_handles(alive_cell):
             ray.get(handle.set_fail_methods.remote(["sleep"]))
         uninitialized_cell = make_cell(1)
         group = _make_controller([alive_cell, uninitialized_cell])
@@ -123,7 +123,7 @@ class TestMultipleCellsStillTolerateFailures:
     async def test_one_dead_cell_does_not_stop_the_lifecycle_call(self):
         """Fault tolerance depends on surviving cells carrying on without the dead one."""
         cells = [make_alive_cell(index, alive_cell_indices=[0, 1]) for index in range(2)]
-        ray.get(cells[0]._get_actor_handles()[0].set_fail_methods.remote(["sleep"]))
+        ray.get(get_raw_actor_handles(cells[0])[0].set_fail_methods.remote(["sleep"]))
         group = _make_controller(cells)
 
         await group.offload()
diff --git a/tests/fast/ray/train/test_group_reconcile_adapters.py b/tests/fast/ray/train/test_group_reconcile_adapters.py
index bc83af1dff9..921c514a1c5 100644
--- a/tests/fast/ray/train/test_group_reconcile_adapters.py
+++ b/tests/fast/ray/train/test_group_reconcile_adapters.py
@@ -2,7 +2,7 @@
 
 import pytest
 import ray
-from tests.fast.ray.train.conftest import make_alive_cell, make_cell
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_alive_cell, make_cell
 
 from miles.ray.train.group import TrainerController
 
@@ -19,7 +19,7 @@ def _make_controller(cells: list) -> RayTrainGroup:
 def _reconcile_calls_of(cell) -> list:
     return [
         [call for call in ray.get(handle.get_calls.remote()) if call[0] == "reconcile_adapters"]
-        for handle in cell._get_actor_handles()
+        for handle in get_raw_actor_handles(cell)
     ]
 
 
@@ -37,7 +37,7 @@ async def test_every_worker_of_every_cell_is_asked_to_reconcile(self):
     async def test_a_failing_worker_in_a_later_cell_propagates_instead_of_being_swallowed(self):
         """A stale adapter set would corrupt routing, so a failure in any cell must reach the caller."""
         cells = [make_cell(index) for index in range(2)]
-        ray.get(cells[1]._get_actor_handles()[0].set_fail_methods.remote(["reconcile_adapters"]))
+        ray.get(get_raw_actor_handles(cells[1])[0].set_fail_methods.remote(["reconcile_adapters"]))
         group = _make_controller(cells)
 
         with pytest.raises(Exception, match="Injected failure"):
diff --git a/tests/fast/ray/train/test_train_external_data.py b/tests/fast/ray/train/test_train_external_data.py
index 9535f950752..353f744120a 100644
--- a/tests/fast/ray/train/test_train_external_data.py
+++ b/tests/fast/ray/train/test_train_external_data.py
@@ -2,7 +2,7 @@
 
 import pytest
 import ray
-from tests.fast.ray.train.conftest import make_alive_cell
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_alive_cell
 
 from miles.ray.train import group as group_module
 from miles.ray.train.group import TrainerController
@@ -56,7 +56,7 @@ async def counting_train(**kwargs):
 def _train_calls_of(cell) -> list[tuple]:
     return [
         [call for call in ray.get(handle.get_calls.remote()) if call[0] == "train"]
-        for handle in cell._get_actor_handles()
+        for handle in get_raw_actor_handles(cell)
     ]
 
 
diff --git a/tests/fast/ray/train/test_train_return_value.py b/tests/fast/ray/train/test_train_return_value.py
index 04a33ff1924..10160342c2e 100644
--- a/tests/fast/ray/train/test_train_return_value.py
+++ b/tests/fast/ray/train/test_train_return_value.py
@@ -4,7 +4,7 @@
 
 import pytest
 import ray
-from tests.fast.ray.train.conftest import make_alive_cell
+from tests.fast.ray.train.conftest import get_raw_actor_handles, make_alive_cell
 
 from miles.backends.megatron_utils.ft.types import TrainStepOutcome, TrainStepOutput
 from miles.ray.train.group import TrainerController
@@ -28,14 +28,14 @@ def _make_controller(cells: list) -> RayTrainGroup:
 
 
 def _set_train_return_value(cell: Any, value: Any) -> None:
-    for handle in cell._get_actor_handles():
+    for handle in get_raw_actor_handles(cell):
         ray.get(handle.set_train_return_value.remote(value))
 
 
 def _count_train_calls(cell: Any) -> int:
     return sum(
         sum(1 for method, _args, _kwargs in ray.get(handle.get_calls.remote()) if method == "train")
-        for handle in cell._get_actor_handles()
+        for handle in get_raw_actor_handles(cell)
     )
 
 
diff --git a/tests/fast/utils/workers/real_ray/test_ray_worker_manager.py b/tests/fast/utils/workers/real_ray/test_ray_worker_manager.py
index 431065e440b..3ac93dc545d 100644
--- a/tests/fast/utils/workers/real_ray/test_ray_worker_manager.py
+++ b/tests/fast/utils/workers/real_ray/test_ray_worker_manager.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import time
 
 import pytest
@@ -13,6 +14,7 @@
 
 from miles.utils.http_utils import wait_tcp_ready
 from miles.utils.workers.ray_worker_manager import RayWorkerManager
+from miles.utils.workers.worker_handle import WorkerUnreachableError
 from miles.utils.workers.worker_provider.ray import RayWorkerProvider
 from miles.utils.workers.worker_spec import HostAndPort, PortInfo
 
@@ -316,7 +318,9 @@ def test_stopping_a_cell_ends_its_worker_processes_and_leaves_the_others_alone(
 
 
 class TestWorkerInfosOnRealRay:
-    def test_a_driver_can_describe_and_reach_the_workers_of_one_cell(self, manager_factory, worker_probe_factory):
+    async def test_a_driver_can_describe_and_reach_the_workers_of_one_cell(
+        self, manager_factory, worker_probe_factory
+    ):
         """Worker infos survive the trip to the driver, including usable actor handles."""
         probe = worker_probe_factory()
         manager_factory(
@@ -332,26 +336,32 @@ def test_a_driver_can_describe_and_reach_the_workers_of_one_cell(self, manager_f
         for worker_in_cell_index, info in enumerate(infos):
             recorded = records[f"1-{worker_in_cell_index}"]["context"]["self_addrs"]["primary"]
             assert {"host": info.self_addrs["primary"].host, "port": info.self_addrs["primary"].port} == recorded
-            node_ip = ray.get(info.actor_handle._get_node_ip.remote())
+            node_ip = await info.handle._get_node_ip()
             assert info.self_addrs["primary"].host.strip("[]") == node_ip
 
 
 class TestWorkerDeathOnRealRay:
-    def test_an_actor_dies_with_the_command_it_babysits(self, manager_factory, worker_probe_factory):
+    async def test_an_actor_dies_with_the_command_it_babysits(self, manager_factory, worker_probe_factory):
         """The actor's whole reason to exist is its subprocess, so its death must be visible to the driver."""
         probe = worker_probe_factory()
         manager_factory([make_command_spec("engine", num_workers_per_cell=2, launch_command=probe.launch_command)])
         probe.wait_for_records(2)
         infos = ray.get(RayWorkerManager.get_handle().get_worker_infos.remote("engine-0"))
 
-        ray.get(infos[0].actor_handle.kill_subprocess.remote())
+        # The babysit thread may reach os._exit before this reply is sent, so losing the reply is
+        # the death under test arriving early rather than a failure; the loop below still has to
+        # observe it, so nothing is taken on faith here.
+        try:
+            await infos[0].handle.kill_subprocess()
+        except WorkerUnreachableError:
+            pass
 
         deadline = time.monotonic() + 60
         while True:
             try:
-                ray.get(infos[0].actor_handle._get_node_ip.remote(), timeout=5)
-            except (ray.exceptions.RayActorError, ray.exceptions.GetTimeoutError):
+                await asyncio.wait_for(infos[0].handle._get_node_ip(), timeout=5)
+            except (WorkerUnreachableError, asyncio.TimeoutError):
                 break
             assert time.monotonic() < deadline, "the actor of a dead command is still alive"
-            time.sleep(0.5)
-        assert ray.get(infos[1].actor_handle._get_node_ip.remote(), timeout=30)
+            await asyncio.sleep(0.5)
+        assert await asyncio.wait_for(infos[1].handle._get_node_ip(), timeout=30)
diff --git a/tests/fast/utils/workers/test_ray_worker_manager.py b/tests/fast/utils/workers/test_ray_worker_manager.py
index 3b98aaa0f82..21bf7a95dae 100644
--- a/tests/fast/utils/workers/test_ray_worker_manager.py
+++ b/tests/fast/utils/workers/test_ray_worker_manager.py
@@ -211,6 +211,19 @@ async def test_static_ports_bypass_the_allocator(self, fake_ray_cluster: FakeRay
         assert manager.get_worker_addrs("router-0-0")["primary"].port == 7777
         assert len(fake_ray_cluster.calls_of("_get_free_port_block")) == 1
 
+    async def test_static_ports_marked_offset_by_cell_advance_per_cell(self, fake_ray_cluster: FakeRayCluster):
+        """Cells sharing one pinned base port would bind the same address twice on a node."""
+        spec = _make_spec(
+            "session-server",
+            num_cells=3,
+            port_infos=[PortInfo(name="primary", static_port=7000, allow_dynamic=False, offset_by_cell=True)],
+        )
+        manager = await _launch([spec])
+
+        assert [
+            manager.get_worker_addrs(f"session-server-{cell_index}-0")["primary"].port for cell_index in range(3)
+        ] == [7000, 7001, 7002]
+
     async def test_ports_are_tracked_per_node(self, fake_ray_cluster: FakeRayCluster):
         """Workers on different nodes may reuse the same port number."""
         fake_ray_cluster.use_node_ips("10.0.0.1", "10.0.0.2")
@@ -603,6 +616,28 @@ async def gated_alloc(self) -> None:
         assert sorted(entered) == [0, 1, 2]
         assert len({manager.get_worker_addrs(f"engine-{index}-0")["primary"].port for index in range(3)}) == 3
 
+    async def test_workers_within_one_cell_run_each_phase_concurrently(self, fake_ray_cluster: FakeRayCluster):
+        """The ranks of one engine configure together; serialising them adds a full setup per rank."""
+        from miles.utils.workers.ray_worker_manager import _CommandActorManager
+
+        original_alloc_ports = _CommandActorManager.alloc_ports
+        entered: list[int] = []
+        release = asyncio.Event()
+
+        async def gated_alloc(self) -> None:
+            entered.append(self.worker_in_cell_index)
+            if len(entered) == 3:
+                release.set()
+            await asyncio.wait_for(release.wait(), timeout=5)
+            await original_alloc_ports(self)
+
+        with pytest.MonkeyPatch.context() as patched:
+            patched.setattr(_CommandActorManager, "alloc_ports", gated_alloc)
+            manager = await asyncio.wait_for(_launch([_make_spec("engine", num_workers_per_cell=3)]), timeout=10)
+
+        assert sorted(entered) == [0, 1, 2]
+        assert len({manager.get_worker_addrs(f"engine-0-{index}")["primary"].port for index in range(3)}) == 3
+
 
 class TestGpuPlacement:
     async def test_gpu_slots_expand_over_cells_and_workers(self, fake_ray_cluster: FakeRayCluster):
@@ -810,6 +845,20 @@ async def poll_cells() -> None:
 
         assert polls_during_stop > 20
 
+    async def test_a_hung_shutdown_does_not_delay_killing_peer_workers(self, fake_ray_cluster: FakeRayCluster):
+        """One rank ignoring shutdown must not keep its peers holding gpus for the whole grace period."""
+        manager = await _launch([_make_spec("engine", num_workers_per_cell=2)])
+        fake_ray_cluster.handles[0].hanging_methods["shutdown"] = 3600
+
+        with patch.object(ray_worker_manager, "_SHUTDOWN_TIMEOUT", 0.5):
+            stop_task = asyncio.create_task(manager._pools["engine"].cells[0].stop())
+            await asyncio.sleep(0.05)
+            assert fake_ray_cluster.handles[1].killed
+            assert not stop_task.done()
+            await asyncio.wait_for(stop_task, timeout=5)
+
+        assert fake_ray_cluster.events.count(EVENT_KILL) == 2
+
     async def test_all_workers_are_killed_even_when_one_shutdown_hangs(self, fake_ray_cluster: FakeRayCluster):
         """One broken worker must not save its peers from teardown."""
         manager = await _launch([_make_spec("engine", num_workers_per_cell=3)])
@@ -840,7 +889,7 @@ async def test_describes_every_worker_of_only_the_requested_cell(self, fake_ray_
         assert [info.generation for info in infos] == [1, 1]
         assert [info.gpu_ids for info in infos] == [[4, 5], [6, 7]]
         assert [info.self_addrs for info in infos] == manager.get_addrs()["engine"][2:]
-        assert [info.actor_handle for info in infos] == fake_ray_cluster.handles[2:]
+        assert [info.handle._actor_handle for info in infos] == fake_ray_cluster.handles[2:]
 
     async def test_a_stopped_cell_reports_no_workers_instead_of_raising(self, fake_ray_cluster: FakeRayCluster):
         """A cell stopped between snapshot and round-trip must report an empty worker list, not crash."""
@@ -902,6 +951,21 @@ async def test_every_info_says_whether_its_cell_still_has_a_process(self, fake_r
 
         assert manager.get_cell_infos(pool_ids=["engine"])["engine-0"].alive
 
+    async def test_each_cell_meta_is_computed_from_its_own_cell_index(self, fake_ray_cluster: FakeRayCluster):
+        """Meta carries per-cell placement facts, so a shared index would mislabel every cell but one."""
+        spec = _make_spec("engine", num_cells=3).model_copy(
+            update={"meta": lambda ctx: {"gpu_offset": ctx.cell_index * 2}}
+        )
+        manager = await _launch([spec])
+
+        infos = manager.get_cell_infos(pool_ids=["engine"])
+
+        assert [infos[f"engine-{cell_index}"].meta for cell_index in range(3)] == [
+            {"gpu_offset": 0},
+            {"gpu_offset": 2},
+            {"gpu_offset": 4},
+        ]
+
 
 class TestStartAndStopCells:
     async def test_stopping_by_id_releases_only_the_named_cell(self, fake_ray_cluster: FakeRayCluster):
@@ -943,6 +1007,15 @@ async def test_a_restarted_cell_runs_its_command_again(self, fake_ray_cluster: F
 
         assert len(fake_ray_cluster.calls_of("run")) == 2
 
+    async def test_starting_a_running_cell_leaves_it_alone(self, fake_ray_cluster: FakeRayCluster):
+        """Relaunching a live cell would orphan its current actors, so a repeated resume is a no-op."""
+        manager = await _launch([_make_spec("engine")])
+        handles_before = list(fake_ray_cluster.handles)
+
+        await manager.start_cells(["engine-0"])
+
+        assert fake_ray_cluster.handles == handles_before
+
     async def test_stopping_an_already_stopped_cell_is_a_noop(self, fake_ray_cluster: FakeRayCluster):
         """Heal loops retry, so a redundant suspend must not blow up on missing actors."""
         manager = await _launch([_make_spec("engine")])
@@ -986,6 +1059,13 @@ async def test_an_unknown_cell_id_fails_loudly(self, fake_ray_cluster: FakeRayCl
         with pytest.raises(AssertionError):
             await manager.stop_cells(["engine-7"])
 
+    async def test_starting_an_unknown_cell_id_fails_loudly(self, fake_ray_cluster: FakeRayCluster):
+        """A typo'd cell id must not silently start nothing while the caller believes it resumed."""
+        manager = await _launch([_make_spec("engine")])
+
+        with pytest.raises(AssertionError):
+            await manager.start_cells(["engine-7"])
+
     async def test_a_cell_that_fails_while_allocating_ports_is_rolled_back_and_can_be_retried(
         self, fake_ray_cluster: FakeRayCluster
     ):
@@ -1142,6 +1222,31 @@ async def failing_alloc(self) -> None:
         assert manager.get_cell_infos(pool_ids=["engine"])["engine-0"].alive
         assert len(fake_ray_cluster.calls_of("run")) == 2
 
+    async def test_a_failed_start_leaves_no_late_sibling_actor_alive(self, fake_ray_cluster: FakeRayCluster):
+        """A sibling still launching when the request failed must not survive the rollback holding its gpus."""
+        from miles.utils.workers.ray_worker_manager import _CommandActorManager
+
+        original_launch_actor = _CommandActorManager.launch_actor
+        first_failed = asyncio.Event()
+
+        async def staggered_launch(self) -> None:
+            if self.parent.cell_index == 0:
+                first_failed.set()
+                raise RuntimeError("no capacity")
+            await first_failed.wait()
+            await asyncio.sleep(0.05)
+            await original_launch_actor(self)
+
+        manager = RayWorkerManager()
+        with pytest.raises(RuntimeError, match="no capacity"):
+            with pytest.MonkeyPatch.context() as patched:
+                patched.setattr(_CommandActorManager, "launch_actor", staggered_launch)
+                await manager.init([_make_spec("engine", num_cells=2)], {})
+        await asyncio.sleep(0.1)
+
+        assert all(handle.killed for handle in fake_ray_cluster.handles)
+        assert not any(info.alive for info in manager.get_cell_infos(pool_ids=["engine"]).values())
+
     async def test_a_failed_start_rolls_back_the_siblings_of_the_failing_cell(self, fake_ray_cluster: FakeRayCluster):
         """One request is one transaction, so no cell of it may survive half configured."""
         spec = _make_spec("engine", num_cells=2)
diff --git a/tests/fast/utils/workers/test_ray_worker_manager_serve.py b/tests/fast/utils/workers/test_ray_worker_manager_serve.py
index 2a0d28a10b6..a0b83a0e721 100644
--- a/tests/fast/utils/workers/test_ray_worker_manager_serve.py
+++ b/tests/fast/utils/workers/test_ray_worker_manager_serve.py
@@ -30,6 +30,7 @@ def _make_spec(
     num_gpu_slots_per_worker: int = 0,
     pg_name: str | None = None,
     env_var=None,
+    worker_class: str = _WORKER_CLASS_PATH,
 ) -> ServeWorkerSpec:
     return ServeWorkerSpec(
         name=name,
@@ -43,7 +44,7 @@ def _make_spec(
             num_gpu_slots_per_worker=num_gpu_slots_per_worker,
             pg_name=pg_name,
         ),
-        worker_class=_WORKER_CLASS_PATH,
+        worker_class=worker_class,
         ctor_kwargs=ctor_kwargs if ctor_kwargs is not None else (lambda _ctx: {}),
         concurrency_groups=concurrency_groups,
     )
@@ -116,6 +117,19 @@ async def test_env_vars_are_computed_per_worker(self, fake_ray_cluster: FakeRayC
         assert env_vars == [{"RANK_DIR": "/d/0"}, {"RANK_DIR": "/d/1"}]
 
 
+class TestServeWorkerClassFailures:
+    async def test_an_unloadable_worker_class_rolls_back_the_serve_cell(self, fake_ray_cluster: FakeRayCluster):
+        """A cell left alive around a class that cannot be imported would never be retried nor serve."""
+        spec = _make_spec(worker_class=f"{_WORKER_CLASS_PATH}Missing")
+        manager = RayWorkerManager()
+
+        with pytest.raises(Exception, match="DemoServeWorkerMissing"):
+            await manager.init([spec], {})
+
+        assert fake_ray_cluster.handles == []
+        assert not manager.get_cell_infos(pool_ids=["trainer"])["trainer-0"].alive
+
+
 class TestServeSchedulingOptions:
     async def test_concurrency_groups_reach_ray(self, fake_ray_cluster: FakeRayCluster):
         """The trainer heartbeat rpc must not queue behind a running train step."""
diff --git a/tests/fast/utils/workers/test_worker_info.py b/tests/fast/utils/workers/test_worker_info.py
new file mode 100644
index 00000000000..13198a712ca
--- /dev/null
+++ b/tests/fast/utils/workers/test_worker_info.py
@@ -0,0 +1,41 @@
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from miles.utils.workers.worker_handle import BaseWorkerHandle
+from miles.utils.workers.worker_info import WorkerInfo
+from miles.utils.workers.worker_spec import HostAndPort
+
+
+class _FakeWorkerHandle(BaseWorkerHandle):
+    async def wait_ready(self, *, timeout: float) -> None:
+        return None
+
+    async def wait_dead(self, *, timeout: float) -> None:
+        return None
+
+
+def _fields(**overrides: Any) -> dict[str, Any]:
+    return (
+        dict(
+            name="engine-0-0",
+            generation=1,
+            self_addrs={"primary": HostAndPort(host="10.0.0.1", port=30000)},
+            gpu_ids=[0],
+            handle=_FakeWorkerHandle(),
+        )
+        | overrides
+    )
+
+
+class TestWorkerInfoValidation:
+    def test_an_unknown_field_is_rejected(self):
+        """A misspelled field must fail loudly instead of silently dropping out of the description."""
+        with pytest.raises(ValidationError, match="gpu_id"):
+            WorkerInfo(**_fields(gpu_id=[0]))
+
+    def test_a_non_worker_handle_is_rejected(self):
+        """The handle is the only way to reach the worker, so a raw actor or stray object must not pass."""
+        with pytest.raises(ValidationError, match="BaseWorkerHandle"):
+            WorkerInfo(**_fields(handle=object()))
diff --git a/tests/fast/utils/workers/worker_provider/test_ray.py b/tests/fast/utils/workers/worker_provider/test_ray.py
index 06c3729a94a..3a740f680e9 100644
--- a/tests/fast/utils/workers/worker_provider/test_ray.py
+++ b/tests/fast/utils/workers/worker_provider/test_ray.py
@@ -8,6 +8,7 @@
 import pytest
 
 import miles.utils.workers.worker_provider.ray as ray_worker_provider_mod
+from miles.utils.workers.worker_handle import BaseWorkerHandle
 from miles.utils.workers.worker_info import WorkerInfo
 from miles.utils.workers.worker_provider.base import CellInfo
 from miles.utils.workers.worker_provider.ray import RayWorkerProvider
@@ -116,6 +117,14 @@ def get(refs: list[_FakeObjectRef]) -> list[list[WorkerInfo]]:
         return [ref.infos for ref in refs]
 
 
+class _FakeWorkerHandle(BaseWorkerHandle):
+    async def wait_ready(self, *, timeout: float) -> None:
+        return None
+
+    async def wait_dead(self, *, timeout: float) -> None:
+        return None
+
+
 def _worker_infos(cell_id: str, *, count: int) -> list[WorkerInfo]:
     return [
         WorkerInfo(
@@ -123,7 +132,7 @@ def _worker_infos(cell_id: str, *, count: int) -> list[WorkerInfo]:
             generation=1,
             self_addrs={"primary": HostAndPort(host="10.0.0.7", port=15000 + worker_index)},
             gpu_ids=[worker_index],
-            actor_handle=None,
+            handle=_FakeWorkerHandle(),
         )
         for worker_index in range(count)
     ]