Crash rollout engines from the soak harness instead of the controller

miles/ray/rollout/inference_controller.py

@@ -77,8 +77,6 @@ async def init(self) -> None:
     @with_lock
     async def prepare_rollout(self, rollout_id):
         await self._health_monitoring_resume()
-        if self.args.ci_test and self._rollout_ft_enabled and rollout_id >= 2:
-            await self._try_ci_fault_injection()
         dashboard_hooks.register_engines(self.servers)
 
     @with_lock
@@ -288,23 +286,6 @@ async def _health_monitoring_pause(self) -> None:
     async def _health_monitoring_resume(self) -> None:
         self._health_checker_activeness.bump_active(True)
 
-    @property
-    @requires_lock
-    def _rollout_ft_enabled(self) -> bool:
-        return self.args.use_fault_tolerance and "rollout" in self.args.ft_components
-
-    @property
-    @requires_lock
-    def _server(self) -> RolloutServer | None:
-        """Default server (first model).  For backward compatibility."""
-        if not self.servers:
-            return None
-        return next(iter(self.servers.values()))
-
-    @requires_lock
-    async def _try_ci_fault_injection(self):
-        raise NotImplementedError("rollout fault injection is being rebuilt with rollout fault tolerance")
-
 
 @dataclass(frozen=True)
 class UpdatableEngines:

miles/ray/rollout/server_cell.py

@@ -95,6 +95,7 @@ def cell_status(self) -> CellStatus:
                     conditions=[
                         CellCondition.allocated(TriState.TRUE),
                         CellCondition.from_health_checker_status(self._health_checker.status),
+                        CellCondition.serving(TriState.TRUE if self.is_serving else TriState.FALSE),
                     ],
                 )
 
@@ -123,6 +124,10 @@ def is_pending_weights_or_serving(self) -> bool:
     def is_pending_weights(self) -> bool:
         return isinstance(self._state, StatePendingWeights)
 
+    @property
+    def is_serving(self) -> bool:
+        return isinstance(self._state, StateServing)
+
     @property
     def addr_info(self) -> CellAddrInfo:
         assert isinstance(self._state, (StateInitializing, StatePendingWeights, StateServing))

miles/utils/ft_utils/api_server/handles.py

@@ -132,8 +132,8 @@ async def suspend(self, cell_id: str) -> None:
         self._inference_controller.notify_cell_suspended(cell_id)
 
     async def resume(self, cell_id: str) -> None:
-        await self._worker_manager.start_cells.remote([cell_id])
         self._inference_controller.notify_cell_resumed(cell_id)
+        await self._worker_manager.start_cells.remote([cell_id])
 
     async def inject_fault(self, cell_id: str, *, mode: FailureMode, sub_index: int) -> None:
         await self._worker_manager.inject_fault.remote(cell_id, mode=mode.value, worker_in_cell_index=sub_index)

miles/utils/ft_utils/api_server/models.py

@@ -23,7 +23,7 @@ class _OkResponse(StrictBaseModel):
 
 
 class CellCondition(StrictBaseModel):
-    type: Literal["Allocated", "Healthy"]
+    type: Literal["Allocated", "Healthy", "Serving"]
     status: TriState
     reason: str | None = None
     message: str | None = None
@@ -37,6 +37,10 @@ def allocated(cls, status: TriState) -> CellCondition:
     def healthy(cls, status: TriState, *, reason: str | None = None) -> CellCondition:
         return cls(type="Healthy", status=status, reason=reason)
 
+    @classmethod
+    def serving(cls, status: TriState) -> CellCondition:
+        return cls(type="Serving", status=status)
+
     @classmethod
     def from_health_checker_status(cls, status: TriState) -> CellCondition:
         match status:

tests/e2e/ft/README.md

@@ -3,7 +3,7 @@
 ## Layout
 
 - Scenario logic lives in `conftest_ft/scenario_<name>.py`.
-- CI runs it via thin per-mode entry files `test_trainer_ft_<scenario>_<mode>.py`, each registered with `register_cuda_ci(est_time=..., suite="stage-c-8-gpu-h200", labels=["ft-short"])` (comparison scenarios) or `labels=["ft-long"]` (soak scenarios).
+- CI runs it via thin per-mode entry files `test_<component>_ft_<scenario>_<mode>.py` (`<component>` is `trainer` or `rollout`, after the components the mode enables ft on), each registered with `register_cuda_ci(est_time=..., suite="stage-c-8-gpu-h200", labels=["ft-short"])` (comparison scenarios) or `labels=["ft-long"]` (soak scenarios).
 - The CUDA CI runner executes each entry as bare `python3 <file>` (exit code = pass/fail); the entry just calls the scenario's `run_ci(mode)`.
 
 | Scenario (`conftest_ft/scenario_*.py`) | Type | What it verifies |
@@ -32,6 +32,7 @@
 - All scenarios use `--rollout-batch-size 32 --n-samples-per-prompt 8 --global-batch-size 256` (256 samples/rollout), which divides evenly across both 2 and 4 cells. Uneven sample distribution across replicas is **not** exercised.
 - 1-node modes use the 5-layer MoE (`Qwen3-30B-A3B-5layer`), except `dp2_cp2_real_rollout_dense` and `colocate_dp2_cp2_rollout_ft` (dense `Qwen3-0.6B` — see `scenario_with_failure` for why).
 - Authorized CI skips (no entry file): `6node_dp4_cp2_tp2_pp2_ep2_etp2` (multi-node), `with_failure × dp4_cp2`.
+- `colocate_dp2_cp2_rollout_ft` has one entry, `test_rollout_ft_random_colocate_dp2_cp2.py` (`scenario_ft_random`); it is the only mode that crashes engines rather than trainer cells.
 
 ## Running
 
@@ -267,16 +268,12 @@ Architecture (external fault injection, not inside training loop):
      --crash-probability is set high enough that the soak reliably clears this floor. Faults are
      random, so neither an exact sequence nor the end-state membership is asserted — the
      witness only proves repeated faults were injected and healing actually ran.
-  8. Rollout healing witness: every accepted rollout injection is paired with a Running -> Pending ->
-     Running of the cell it targeted, taken from that cell's phase history at or after the injection
-     and consumed by at most one injection, proving its engine was replaced and came back serving.
+  8. Rollout recovery witness: every accepted rollout injection must be paired, per cell and in
+     order, with one completed Serving -> (Suspended|Pending) -> Serving cycle; an injection still
+     unpaired when training ends fails the soak. The terminal state is Serving, not Running:
+     phase Running also covers a replacement that got weights but never re-entered the router.
      Suspended is not required in between — it lasts only --mini-ft-controller-resume-delay (10s),
-     which a 2s poll can miss. Pending alone does not prove the replacement was gated: a disaggregated
-     relaunch also passes through it. An unpaired injection fails the witness unless the cell is still
-     not back in Running when training ends (its recovery simply had no time to finish); >= 2 paired
-     recoveries are required, so a single heal cannot vouch for many injections. The injector takes one
-     last cell snapshot after stopping its poll thread, so a recovery that completed between the final
-     poll and the end of training is still counted.
+     which a 2s poll can miss.
 
 CLI options: --seed (default 42), --num-steps (default 30), --crash-probability (default 0.5)
 ```

tests/e2e/ft/conftest_ft/fault_injection.py

@@ -1,11 +1,13 @@
 # NOTE: You MUST read tests/e2e/ft/README.md as source-of-truth and documentations
 
+import dataclasses
 import enum
 import logging
 import random
 import threading
 import time
 from collections.abc import Callable
+from typing import Literal
 
 import requests
 
@@ -50,66 +52,121 @@ def genuinely_alive(self, cells: list[dict]) -> list[dict]:
         return [c for c in cells if cell_is_alive(c) and c["metadata"]["name"] not in self._state_of_cell_name]
 
 
+class ObservedCellState(enum.Enum):
+    SUSPENDED = "Suspended"  # torn down, holding no gpu
+    PENDING = "Pending"  # allocated but gated: no engine serving yet
+    RUNNING_NOT_SERVING = "RunningNotServing"  # engine is up but not registered in the router
+    SERVING = "Serving"  # registered in the router, i.e. actually able to answer requests
+
+
+_RELAUNCH_STATES: tuple[ObservedCellState, ...] = (ObservedCellState.SUSPENDED, ObservedCellState.PENDING)
+
+
+def compute_observed_cell_state(cell: dict) -> ObservedCellState:
+    phase = cell["status"]["phase"]
+    if phase == "Suspended":
+        return ObservedCellState.SUSPENDED
+    if phase == "Pending":
+        return ObservedCellState.PENDING
+    serving = any(cond["type"] == "Serving" and cond["status"] == "True" for cond in cell["status"]["conditions"])
+    return ObservedCellState.SERVING if serving else ObservedCellState.RUNNING_NOT_SERVING
+
+
 @dataclasses.dataclass(frozen=True)
-class InjectionOutcome:
-    cell_name: str
-    injection_index: int
-    recovered: bool
-    still_down: bool
+class _CellEvent:
+    kind: Literal["injected", "observed"]
+    state: ObservedCellState | None = None
+
 
+@dataclasses.dataclass
+class _CellInfo:
+    cell_type: str | None = None
+    events: list[_CellEvent] = dataclasses.field(default_factory=list)
 
-def _find_healing_end(phases: list[str], start: int) -> int | None:
-    remaining = ["Running", "Pending", "Running"]
-    for index in range(start, len(phases)):
-        if phases[index] == remaining[0]:
-            remaining.pop(0)
-            if not remaining:
-                return index
-    return None
 
+class RecoveryWitness:
+    """Pairs every accepted injection with one completed relaunch-and-serve cycle of the same cell."""
 
-class PhaseHistory:
     def __init__(self) -> None:
-        self.phases_of_cell_name: dict[str, list[str]] = {}
-        self._cell_type_of_name: dict[str, str] = {}
-        self._injection_marks_of_cell_name: dict[str, list[int]] = {}
+        self._info_of_cell_name: dict[str, _CellInfo] = {}
+
+    def note_injected(self, cell_name: str) -> None:
+        self._info(cell_name).events.append(_CellEvent(kind="injected"))
 
     def observe(self, cells: list[dict]) -> None:
         for cell in cells:
-            name = cell["metadata"]["name"]
-            self._cell_type_of_name[name] = _cell_type_of(cell)
-            phases = self.phases_of_cell_name.setdefault(name, [])
-            phase = cell["status"]["phase"]
-            if not phases or phases[-1] != phase:
-                phases.append(phase)
+            info = self._info(cell["metadata"]["name"])
+            info.cell_type = _cell_type_of(cell)
+            info.events.append(_CellEvent(kind="observed", state=compute_observed_cell_state(cell)))
+
+    @property
+    def states_of_cell_name(self) -> dict[str, list[ObservedCellState]]:
+        return {
+            name: states
+            for name, info in self._info_of_cell_name.items()
+            if (states := _compute_distinct_states(info.events))
+        }
+
+    def num_injections(self, *, cell_type: str | None = None) -> int:
+        return sum(
+            sum(1 for event in info.events if event.kind == "injected")
+            for info in self._matching_infos(cell_type=cell_type)
+        )
 
-    def note_injected(self, cell_name: str) -> None:
-        phases = self.phases_of_cell_name.setdefault(cell_name, [])
-        marks = self._injection_marks_of_cell_name.setdefault(cell_name, [])
-        marks.append(max(len(phases) - 1, 0))
-
-    def injection_outcomes(self, *, cell_type: str | None = None) -> list[InjectionOutcome]:
-        outcomes: list[InjectionOutcome] = []
-        for name, marks in sorted(self._injection_marks_of_cell_name.items()):
-            if cell_type is not None and self._cell_type_of_name.get(name) != cell_type:
-                continue
-
-            phases = self.phases_of_cell_name[name]
-            next_search_start = 0
-            for injection_index, mark in enumerate(marks):
-                healing_end = _find_healing_end(phases, max(mark, next_search_start))
-                if healing_end is not None:
-                    next_search_start = healing_end
-                outcomes.append(
-                    InjectionOutcome(
-                        cell_name=name,
-                        injection_index=injection_index,
-                        recovered=healing_end is not None,
-                        still_down=healing_end is None and bool(phases) and phases[-1] != "Running",
-                    )
-                )
-
-        return outcomes
+    def num_completed_recoveries(self, *, cell_type: str | None = None) -> int:
+        return sum(
+            _compute_recovery_tally(info.events).num_completed for info in self._matching_infos(cell_type=cell_type)
+        )
+
+    def cells_with_unfinished_recovery(self, *, cell_type: str | None = None) -> dict[str, int]:
+        return {
+            name: tally.num_unfinished
+            for name, info in self._info_of_cell_name.items()
+            if (cell_type is None or info.cell_type == cell_type)
+            and (tally := _compute_recovery_tally(info.events)).num_unfinished
+        }
+
+    def _info(self, cell_name: str) -> _CellInfo:
+        return self._info_of_cell_name.setdefault(cell_name, _CellInfo())
+
+    def _matching_infos(self, *, cell_type: str | None) -> list[_CellInfo]:
+        return [info for info in self._info_of_cell_name.values() if cell_type is None or info.cell_type == cell_type]
+
+
+@dataclasses.dataclass(frozen=True)
+class _RecoveryTally:
+    num_completed: int
+    num_unfinished: int
+
+
+class _RecoveryStage(enum.Enum):
+    AWAITING_RELAUNCH = enum.auto()
+    AWAITING_SERVING = enum.auto()
+
+
+def _compute_recovery_tally(events: list[_CellEvent]) -> _RecoveryTally:
+    pending: list[_RecoveryStage] = []
+    num_completed = 0
+    for event in events:
+        if event.kind == "injected":
+            pending.append(_RecoveryStage.AWAITING_RELAUNCH)
+            continue
+        if not pending:
+            continue
+        if pending[0] is _RecoveryStage.AWAITING_RELAUNCH and event.state in _RELAUNCH_STATES:
+            pending[0] = _RecoveryStage.AWAITING_SERVING
+        elif pending[0] is _RecoveryStage.AWAITING_SERVING and event.state is ObservedCellState.SERVING:
+            pending.pop(0)
+            num_completed += 1
+    return _RecoveryTally(num_completed=num_completed, num_unfinished=len(pending))
+
+
+def _compute_distinct_states(events: list[_CellEvent]) -> list[ObservedCellState]:
+    states: list[ObservedCellState] = []
+    for event in events:
+        if event.kind == "observed" and event.state is not None and (not states or states[-1] != event.state):
+            states.append(event.state)
+    return states
 
 
 def _compute_next_injection_time(rng: random.Random, mean_interval_seconds: float) -> float:
@@ -124,7 +181,7 @@ def run_fault_injection_loop(
     stop_event: threading.Event,
     on_successful_injection: Callable[[], None],
     cell_type: str | None,
-    phase_history: PhaseHistory,
+    recovery_witness: RecoveryWitness,
     poll_interval_seconds: float = POLL_INTERVAL_SECONDS,
 ) -> None:
     rng = random.Random(seed)
@@ -142,7 +199,7 @@ def run_fault_injection_loop(
         # Track recovery on every poll so a crash->detect->heal cycle that completes between two
         # sparse injections is seen, not missed (which would exclude the cell from the live set forever).
         gate.observe({c["metadata"]["name"]: c for c in cells})
-        phase_history.observe(cells)
+        recovery_witness.observe(cells)
 
         if time.monotonic() < next_injection_time:
             continue
@@ -171,7 +228,7 @@ def run_fault_injection_loop(
             )
             resp.raise_for_status()
             gate.note_injected(cell_name)
-            phase_history.note_injected(cell_name)
+            recovery_witness.note_injected(cell_name)
             on_successful_injection()
             next_injection_time = _compute_next_injection_time(rng, mean_interval_seconds)
         except Exception:
@@ -199,7 +256,7 @@ def _matches_cell_type(cell: dict, cell_type: str | None) -> bool:
 class FaultInjectorHandle:
     def __init__(self, *, base_url: str, seed: int, mean_interval_seconds: float, cell_type: str | None) -> None:
         self.num_successful_injections: int = 0
-        self.phase_history = PhaseHistory()
+        self.recovery_witness = RecoveryWitness()
         self._base_url = base_url
         self._cell_type = cell_type
         self._stop_event = threading.Event()
@@ -212,7 +269,7 @@ def __init__(self, *, base_url: str, seed: int, mean_interval_seconds: float, ce
                 "stop_event": self._stop_event,
                 "on_successful_injection": self._on_successful_injection,
                 "cell_type": cell_type,
-                "phase_history": self.phase_history,
+                "recovery_witness": self.recovery_witness,
             },
             daemon=True,
             name="ft-random-fault-injector",
@@ -230,7 +287,7 @@ def _observe_final_snapshot(self) -> None:
         cells = list_cells(base_url=self._base_url, cell_type=self._cell_type)
         if cells is None:
             return
-        self.phase_history.observe(cells)
+        self.recovery_witness.observe(cells)
 
     def _on_successful_injection(self) -> None:
         self.num_successful_injections += 1