diff --git a/miles/utils/arguments.py b/miles/utils/arguments.py index 70a30eebbce..8cbe83242a2 100644 --- a/miles/utils/arguments.py +++ b/miles/utils/arguments.py @@ -99,6 +99,7 @@ def reset_arg(parser, name, **kwargs): _FT_CHOICES = ["rollout", "train"] +_DEFAULT_FT_API_SERVER_PORT = 18080 def get_miles_extra_args_provider(add_custom_arguments=None): @@ -984,14 +985,18 @@ def add_fault_tolerance_arguments(parser): parser.add_argument( "--api-server-port", type=int, - default=0, - help="Port for HTTP api server. 0 = disabled.", + default=None, + help=f"Port for HTTP api server. 0 = disabled. Left unset it is " + f"{_DEFAULT_FT_API_SERVER_PORT} under --use-fault-tolerance and 0 otherwise, " + f"because the mini fault-tolerance controller drives cells over this port.", ) parser.add_argument( "--mini-ft-controller-enable", - action="store_true", - default=False, - help="Enable the mini fault-tolerance controller that auto-heals Fatal cells.", + action=argparse.BooleanOptionalAction, + default=None, + help="Enable the mini fault-tolerance controller that auto-heals Fatal cells. " + "Left unset it follows --ft-components and --api-server-port, which is what makes " + "--use-fault-tolerance heal on its own.", ) parser.add_argument( "--mini-ft-controller-poll-interval", @@ -2888,6 +2893,18 @@ def _validate_rematerialize_param_from_master_weight(args): args.check_rematerialize_param_from_master_weight = True +def _resolve_api_server_port(args: argparse.Namespace) -> int: + if (port := args.api_server_port) is not None: + return port + return _DEFAULT_FT_API_SERVER_PORT if args.ft_components else 0 + + +def _resolve_mini_ft_controller_enable(args: argparse.Namespace) -> bool: + if (enable := args.mini_ft_controller_enable) is not None: + return enable + return bool(args.ft_components) and args.api_server_port != 0 + + def miles_validate_args(args): validate_dashboard_args(args) @@ -2899,9 +2916,6 @@ def miles_validate_args(args): ) args.eval_datasets = _resolve_eval_datasets(args) - if args.mini_ft_controller_enable and args.api_server_port == 0: - raise ValueError("--mini-ft-controller-enable requires --api-server-port to be set (non-zero)") - if "train" in args.ft_components: args.indep_dp = True args.delay_split_train_data_by_dp = True @@ -3595,6 +3609,12 @@ def miles_validate_args(args): _maybe_apply_dumper_overrides(args) + args.api_server_port = _resolve_api_server_port(args) + args.mini_ft_controller_enable = _resolve_mini_ft_controller_enable(args) + + if args.mini_ft_controller_enable and args.api_server_port == 0: + raise ValueError("--mini-ft-controller-enable requires --api-server-port to be set (non-zero)") + def validate_skip_actor_forward_only(args) -> None: option = "--skip-actor-forward-only" @@ -3689,6 +3709,7 @@ def _maybe_apply_dumper_overrides(args) -> None: if args.use_fault_tolerance: logger.info("Dumper mode: disabling --use-fault-tolerance to suppress fault tolerance heartbeats") args.use_fault_tolerance = False + args.ft_components = [] logger.info("Dumper mode: all heartbeat mechanisms disabled") args.router_disable_health_check = True diff --git a/tests/fast/utils/api_server/test_server.py b/tests/fast/utils/api_server/test_server.py index 8324fb26e76..43623030b69 100644 --- a/tests/fast/utils/api_server/test_server.py +++ b/tests/fast/utils/api_server/test_server.py @@ -5,6 +5,8 @@ import httpx import pytest +from tests.fast.ray.rollout.conftest import make_args as make_rollout_args + from miles.ray.rollout.server_cell import compute_pending_rollout_cell_status from miles.utils.ft_utils.api_server import server from miles.utils.ft_utils.api_server.registry import _CellRegistry @@ -231,12 +233,12 @@ def _start( monkeypatch.setattr(server, "_start_api_server_raw", lambda registry, port: registries.append(registry)) server.start_api_server( - args=SimpleNamespace(), + args=make_rollout_args(), actor_model=make_mock_group([]), inference_controller=MockInferenceController( {cell_id: compute_pending_rollout_cell_status() for cell_id in cell_ids} ), - port=0, + port=18080, ft_components=ft_components, ) @@ -271,7 +273,6 @@ async def test_both_handlers_coexist_under_mixed_ft(self, monkeypatch: pytest.Mo assert [handler.cell_type for handler in registry._handlers] == ["actor", "rollout"] - class TestDynamicCells: @pytest.mark.asyncio async def test_a_cell_that_appears_after_startup_is_served( diff --git a/tests/fast/utils/test_arguments.py b/tests/fast/utils/test_arguments.py index 0654cd83b15..ad43bf09fe0 100644 --- a/tests/fast/utils/test_arguments.py +++ b/tests/fast/utils/test_arguments.py @@ -8,10 +8,10 @@ from miles.backends.sglang_utils.arguments import add_sglang_arguments, collect_eval_sglang_overrides from miles.backends.sglang_utils.arguments import validate_args as validate_sglang_args -from miles.router.config import MilesRouterConfig, compute_miles_router_config from miles.utils.arguments import ( _maybe_apply_dumper_overrides, _resolve_ft_components, + _resolve_mini_ft_controller_enable, _resolve_rollout_functions, _validate_rematerialize_param_from_master_weight, get_miles_extra_args_provider, @@ -81,6 +81,7 @@ def _make_args( *, dumper_enable: bool = False, use_fault_tolerance: bool = False, + ft_components: list[str] | None = None, router_disable_health_check: bool = False, rollout_health_check_interval: float = 30.0, miles_router_health_check_failure_threshold: int = 3, @@ -96,6 +97,8 @@ def _make_args( return SimpleNamespace( dumper_enable=dumper_enable, use_fault_tolerance=use_fault_tolerance, + ft_components=ft_components if ft_components is not None else [], + mini_ft_controller_enable=None, router_disable_health_check=router_disable_health_check, rollout_health_check_interval=rollout_health_check_interval, miles_router_health_check_failure_threshold=miles_router_health_check_failure_threshold, @@ -135,20 +138,23 @@ def test_disables_fault_tolerance_and_sglang_router_heartbeats(self) -> None: assert args.use_fault_tolerance is False assert args.router_disable_health_check is True - def test_leaves_miles_router_heartbeat_enabled(self) -> None: - """Dumper mode does not suppress MilesRouter probing: its health check interval is unchanged.""" - args = self._make_args( - dumper_enable=True, - use_fault_tolerance=True, - rollout_health_check_interval=30.0, - ) + def test_no_healing_loop_survives_dumper_mode(self) -> None: + """It is resolved from ft_components, which dumper mode clears, so resolving it first + would leave the loop polling a registry with nothing in it for the whole run.""" + args = self._make_args(dumper_enable=True, use_fault_tolerance=True, ft_components=["rollout"]) + _maybe_apply_dumper_overrides(args) - config: MilesRouterConfig = compute_miles_router_config(args, host="10.0.0.1", port=1234) + assert _resolve_mini_ft_controller_enable(args) is False - assert args.rollout_health_check_interval == 30.0 - assert config.health_check_interval == 30.0 - assert config.health_check_failure_threshold == 3 + def test_the_selected_ft_components_go_with_the_flag(self) -> None: + """ft_components is resolved from the flag long before this runs, so clearing the flag + alone would leave every component selected and its probes still firing.""" + args = self._make_args(dumper_enable=True, use_fault_tolerance=True, ft_components=["rollout", "train"]) + + _maybe_apply_dumper_overrides(args) + + assert args.ft_components == [] def test_forces_single_rollout(self) -> None: args = self._make_args(dumper_enable=True, num_rollout=100) diff --git a/tests/fast/utils/test_mini_ft_controller.py b/tests/fast/utils/test_mini_ft_controller.py index 3f3f3fb6fef..aea7d8a6439 100644 --- a/tests/fast/utils/test_mini_ft_controller.py +++ b/tests/fast/utils/test_mini_ft_controller.py @@ -10,6 +10,11 @@ import httpx import pytest +from miles.utils.arguments import ( + _DEFAULT_FT_API_SERVER_PORT, + _resolve_api_server_port, + _resolve_mini_ft_controller_enable, +) from miles.utils.ft_utils import mini_ft_controller from miles.utils.ft_utils.api_server.models import Cell, CellCondition, CellMetadata, CellSpec, CellStatus, TriState from miles.utils.ft_utils.mini_ft_controller import ( @@ -649,26 +654,40 @@ async def test_resume_sends_correct_patch(self) -> None: assert body == {"spec": {"suspend": False}} -class TestArgumentValidation: - def test_requires_api_server_port(self) -> None: - """mini_ft_controller_enable=True + api_server_port=0 → error.""" - from miles.utils.arguments import miles_validate_args - +class TestFtControllerDefaults: + @staticmethod + def _resolve(**overrides) -> tuple[int, bool]: args = argparse.Namespace( - mini_ft_controller_enable=True, - api_server_port=0, - use_fault_tolerance=False, - ft_components=None, - eval_datasets=None, - eval_data=None, - eval_config=None, - eval_prompt_data=None, - use_miles_dashboard=False, - run_uuid=None, + **{ + "api_server_port": None, + "mini_ft_controller_enable": None, + "ft_components": [], + **overrides, + } ) + args.api_server_port = _resolve_api_server_port(args) + return args.api_server_port, _resolve_mini_ft_controller_enable(args) + + def test_asking_for_fault_tolerance_opens_the_port_and_starts_the_healing_loop(self) -> None: + """The health checkers only publish a status; without both of these a run watches an + engine die and leaves it routed.""" + assert self._resolve(ft_components=["rollout"]) == (_DEFAULT_FT_API_SERVER_PORT, True) + + def test_a_run_without_fault_tolerance_opens_no_port_and_starts_no_loop(self) -> None: + """There is nothing to heal, so the port would be surface area for nobody.""" + assert self._resolve(ft_components=[]) == (0, False) + + def test_an_explicitly_disabled_port_also_disables_the_healing_loop(self) -> None: + """The loop drives cells over that port, so leaving it on would fail every poll.""" + assert self._resolve(ft_components=["rollout"], api_server_port=0) == (0, False) + + def test_an_explicit_port_is_kept_as_given(self) -> None: + """A run that pins the port has an external controller expecting to find it there.""" + assert self._resolve(ft_components=["rollout"], api_server_port=9999) == (9999, True) - with pytest.raises(ValueError, match="--mini-ft-controller-enable requires --api-server-port"): - miles_validate_args(args) + def test_an_explicit_healing_choice_wins_over_the_default(self) -> None: + """The flag is how a run opts out of healing while keeping the health reporting.""" + assert self._resolve(ft_components=["rollout"], mini_ft_controller_enable=False)[1] is False class _FakeRunner: