feat(planner): search the scaling policy in retune, not user constraints

components/src/dynamo/planner/config/planner_config.py

@@ -2451,12 +2451,12 @@ def _validate_retune(self, retune: RetuneConfig) -> None:
         planner_domain = expand_planner_domain(self)
         router_domain = expand_router_domain(retune)
 
-        if self.enable_throughput_scaling and retune.history_seconds is not None:
+        if retune.history_seconds is not None:
             required = required_history_seconds(self, planner_domain)
             if retune.history_seconds < required:
                 raise ValueError(
                     f"retune.history_seconds must be >= {required:g}s "
-                    "(throughput_adjustment_interval_seconds x "
+                    "(the largest throughput_adjustment_interval_seconds x "
                     "max(prophet_window_size, kalman_min_points) over the "
                     "current predictor and the load_predictor candidates) to "
                     "warm throughput predictors"
@@ -2466,14 +2466,16 @@ def _validate_retune(self, retune: RetuneConfig) -> None:
             raise ValueError(
                 "retune requires a non-empty router or planner search domain"
             )
-        size = grid_size(router_domain, planner_domain.grid)
+        size = grid_size(router_domain) * len(planner_domain.points)
         if size > retune.max_candidates:
+            description = describe_grid(
+                router_domain, planner_domain.grid, len(planner_domain.points)
+            )
             raise ValueError(
                 f"retune grid has {size} candidates, exceeding "
                 f"retune.max_candidates={retune.max_candidates} "
-                f"({describe_grid(router_domain, planner_domain.grid)}); an "
-                "omitted AISimulate knob searches its default domain, so pin "
-                "knobs with a scalar to shrink the grid"
+                f"({description}); an omitted AISimulate knob searches its "
+                "default domain, so pin knobs with a scalar to shrink the grid"
             )
         if retune.executor == "dgdr":
             self._validate_live_retune_budget(retune, router_domain)
@@ -2648,36 +2650,33 @@ def _validate_retune_planner_grid(
         self,
         grid: Mapping[str, Sequence[Any]],
         public_names: Mapping[str, str],
+        *,
+        load_scaling: Optional[bool] = None,
+        throughput_scaling: Optional[bool] = None,
     ) -> dict[str, list[Any]]:
         """Reject planner grid fields that are inert in this configuration.
 
         ``grid`` is keyed by ``PlannerConfig`` field; ``public_names`` gives
-        the ``retune.planner`` name to report. Each value must pass the
-        field's strict validation, as a runtime reconfiguration would.
-        Returns the grid with validated values (``80`` becomes ``80.0`` for a
-        float field), so candidates compare equal to the live config by key.
+        the ``retune.planner`` name to report. ``load_scaling`` and
+        ``throughput_scaling`` say whether some grid point scales that way
+        (default: this config does). Each value must pass the field's strict
+        validation, as a runtime reconfiguration would. Returns the grid with
+        validated values (``80`` becomes ``80.0`` for a float field), so
+        candidates compare equal to the live config by key.
         """
 
-        active_endpoints = (
-            {"min_endpoint"}
-            if self.mode == "agg"
-            else {"prefill_min_endpoint", "decode_min_endpoint"}
-        )
+        if load_scaling is None:
+            load_scaling = self.enable_load_scaling
+        if throughput_scaling is None:
+            throughput_scaling = self.enable_throughput_scaling
         validated: dict[str, list[Any]] = {}
         for key, values in grid.items():
             name = f"retune.planner.{public_names.get(key, key)}"
-            if key in ("min_endpoint", "prefill_min_endpoint", "decode_min_endpoint"):
-                relevant = key in active_endpoints
-                requirement = (
-                    "mode='agg' (use prefill_min_workers/decode_min_workers in disagg)"
-                    if key == "min_endpoint"
-                    else "mode='disagg' (use min_workers in agg)"
-                )
-            elif key == "max_throughput_scaling_replicas":
-                relevant = self.enable_throughput_scaling
+            if key == "max_throughput_scaling_replicas":
+                relevant = throughput_scaling
                 requirement = "enable_throughput_scaling"
             elif key == "load_scaling_down_sensitivity":
-                relevant = self.enable_load_scaling
+                relevant = load_scaling
                 requirement = "enable_load_scaling"
             elif key in (
                 "prefill_scale_up_queue_tokens",
@@ -2689,7 +2688,8 @@ def _validate_retune_planner_grid(
                 relevant = self.optimization_target == "load"
                 requirement = "optimization_target='load'"
             else:
-                # min_gpu_budget / max_gpu_budget apply to any scaling mode.
+                # The scaling-policy fields: expand_planner_domain keeps only
+                # the policies that apply.
                 relevant = True
                 requirement = ""
             if not relevant:
@@ -2736,6 +2736,11 @@ def from_config_arg(cls, config_arg: str) -> "PlannerConfig":
 
         return cls.model_validate(data)
 
+    @property
+    def fpm_max_age_seconds(self) -> float:
+        """Freshness window of live FPM samples: three load intervals."""
+        return 3.0 * self.load_adjustment_interval_seconds
+
     @classmethod
     def _load_from_file(cls, path: Path) -> "PlannerConfig":
         suffix = path.suffix.lower()

components/src/dynamo/planner/core/base.py

@@ -607,10 +607,24 @@ def apply_runtime_candidate(
         self._config_generation += 1
         self.prometheus_metrics.sla_target_ttft_ms.set(self.config.ttft_ms)
         self.prometheus_metrics.sla_target_itl_ms.set(self.config.itl_ms)
+        self._sync_fpm_max_age()
         logger.info(
             "Activated planner configuration revision %d", self._config_generation
         )
 
+    def _sync_fpm_max_age(self) -> None:
+        """Keep the live FPM freshness window at three load intervals after a
+        runtime change of ``load_adjustment_interval_seconds``; the
+        subscribers keep their samples."""
+        provider = getattr(self.environment, "fpm_provider", None)
+        setter = getattr(provider, "set_max_fpm_age_seconds", None)
+        if not callable(setter):
+            return
+        age = self.config.fpm_max_age_seconds
+        if getattr(provider, "max_fpm_age_seconds", None) != age:
+            setter(age)
+            logger.info("FPM freshness window is now %.1fs", age)
+
     @asynccontextmanager
     async def _bounded_config_lock(self) -> AsyncIterator[None]:
         """Acquire the tick decision lock or ask the caller to retry."""
@@ -1314,14 +1328,16 @@ async def _run_one_tick(
     async def run(self) -> None:
         engine = self._ensure_engine()
         next_tick = engine.initial_tick(time.time())
-        poll_interval = self.config.load_adjustment_interval_seconds / 10
-        gpu_shape_retry_interval = min(
-            max(poll_interval, _GPU_SHAPE_RETRY_MIN_SECONDS),
-            _GPU_SHAPE_RETRY_MAX_SECONDS,
-        )
 
         try:
             while True:
+                # Read every iteration: a runtime reconfiguration may change
+                # the load interval.
+                poll_interval = self.config.load_adjustment_interval_seconds / 10
+                gpu_shape_retry_interval = min(
+                    max(poll_interval, _GPU_SHAPE_RETRY_MIN_SECONDS),
+                    _GPU_SHAPE_RETRY_MAX_SECONDS,
+                )
                 now = time.time()
                 if now < next_tick.at_s:
                     await asyncio.sleep(min(next_tick.at_s - now, poll_interval))

components/src/dynamo/planner/core/planner_factory.py

@@ -86,7 +86,7 @@ def construct_environment(
             backend=config.backend,
             model_name=config.model_name,
             runtime=runtime,
-            max_fpm_age_seconds=3.0 * config.load_adjustment_interval_seconds,
+            max_fpm_age_seconds=config.fpm_max_age_seconds,
         )
         connector = construct_connector(
             config,
@@ -132,7 +132,7 @@ def construct_environment(
                 backend=config.backend,
                 model_name=config.model_name,
                 runtime=runtime,
-                max_fpm_age_seconds=3.0 * config.load_adjustment_interval_seconds,
+                max_fpm_age_seconds=config.fpm_max_age_seconds,
                 state_source=environment,
                 namespace_source=namespace_binding,
             )

components/src/dynamo/planner/core/state_machine.py

@@ -209,6 +209,15 @@ def begin_tick(self) -> None:
         """Reset per-tick diagnostics before builtin plugins run."""
         self._reset_diag()
 
+    def reset_throughput_lower_bounds(self) -> None:
+        """Forget the last throughput targets that floor load scaling.
+
+        For a planner whose throughput scaling turns off at runtime: load
+        scaling would otherwise stay floored at a target nothing updates.
+        """
+        self._throughput_lower_bound_p = 1
+        self._throughput_lower_bound_d = 1
+
     def observe_worker_counts(self, counts: WorkerCounts) -> None:
         self._update_inventory(counts)
 

components/src/dynamo/planner/environment/metrics_provider/runtime_provider.py

@@ -62,6 +62,17 @@ def bind_sources(
         self.state_source = state_source
         self.namespace_source = namespace_source
 
+    def set_max_fpm_age_seconds(self, max_fpm_age_seconds: float) -> None:
+        """Change the FPM freshness window of both subscribers in place.
+
+        Tracked samples are kept; reads and the subscribers' clean-up use the
+        new age from now on. Subscribers created later start with it.
+        """
+        for subscriber in (self._prefill_fpm_sub, self._decode_fpm_sub):
+            if subscriber is not None:
+                subscriber.set_max_age(max_age_seconds=max_fpm_age_seconds)
+        self.max_fpm_age_seconds = max_fpm_age_seconds
+
     async def async_init(self, namespace: Optional[str] = None) -> None:
         del namespace
         await self._rebind_subscribers()

components/src/dynamo/planner/offline/replay_adapter.py

@@ -79,8 +79,8 @@
 from dynamo.planner.plugins.clock import VirtualClock
 from dynamo.planner.plugins.orchestrator.engine_adapter import OrchestratorEngineAdapter
 from dynamo.planner.reconfiguration import (
-    PREDICTOR_FIELDS,
     candidate_config,
+    predictor_rebuild_required,
     prepare_predictor,
     runtime_candidate_errors,
 )
@@ -494,10 +494,9 @@ def _apply_retune_decisions(self, decisions: Sequence[RetuneDecision]) -> None:
                     f"planner configuration: {'; '.join(errors)}"
                 )
             predictor = None
-            if any(
-                getattr(candidate, field) != getattr(self._config, field)
-                for field in PREDICTOR_FIELDS
-            ):
+            if predictor_rebuild_required(self._config, candidate):
+                # The sweep bucketed the warmup at the candidate's throughput
+                # interval, as the swapped-in predictor steps.
                 predictor = prepare_predictor(
                     candidate, capabilities, decision.predictor_warmup
                 )