feat(planner): run a live retune sweep through DGDR v1beta2 Requests

components/src/dynamo/common/configuration/groups/kv_router_args.py

@@ -24,6 +24,7 @@
 from dynamo.common.configuration.arg_group import ArgGroup
 from dynamo.common.configuration.config_base import ConfigBase
 from dynamo.common.configuration.utils import (
+    ArgEnv,
     Deprecated,
     add_argument,
     add_negatable_bool_argument,
@@ -297,9 +298,17 @@ def kv_router_kwargs(self) -> dict:
 
 
 class KvRouterArgGroup(ArgGroup):
-    """CLI arguments for the shared KvRouterConfig parameters."""
+    """CLI arguments for the shared KvRouterConfig parameters.
+
+    Args:
+        env: Another process's env (see ``ArgEnv``).
+    """
+
+    def __init__(self, *, env: Optional[ArgEnv] = None) -> None:
+        self.env = env
 
     def add_arguments(self, parser) -> None:
+        environ = os.environ if self.env is None else self.env.environ
         g = parser.add_argument_group("KV Router Options")
 
         add_negatable_bool_argument(
@@ -320,6 +329,7 @@ def add_arguments(self, parser) -> None:
                 "and prefill-token load tracking, and disables remote/shared cache indexers. "
                 "The builtin policy does not request cache inputs in this mode."
             ),
+            env=self.env,
         )
         add_argument(
             g,
@@ -334,6 +344,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="overlap_score_credit",
+            env=self.env,
         )
         add_argument(
             g,
@@ -349,21 +360,23 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="overlap_score_credit_decay",
+            env=self.env,
         )
         add_argument(
             g,
             flag_name="--router-kv-overlap-score-weight",
             obsolete_flag="--kv-overlap-score-weight",
             env_var=(
                 "DYN_ROUTER_KV_OVERLAP_SCORE_WEIGHT"
-                if "DYN_ROUTER_KV_OVERLAP_SCORE_WEIGHT" in os.environ
+                if "DYN_ROUTER_KV_OVERLAP_SCORE_WEIGHT" in environ
                 else "DYN_OVERLAP_SCORE_WEIGHT"
             ),
             dest="overlap_score_weight",
             arg_type=float,
             deprecated=_policy_parameter("prefill_load_scale"),
             default=None,
             help=argparse.SUPPRESS,
+            env=self.env,
         )
         add_argument(
             g,
@@ -377,6 +390,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="prefill_load_scale",
+            env=self.env,
         )
         add_argument(
             g,
@@ -392,6 +406,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="decode_active_request_weight",
+            env=self.env,
         )
         add_argument(
             g,
@@ -406,6 +421,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="host_cache_hit_weight",
+            env=self.env,
         )
         add_argument(
             g,
@@ -419,6 +435,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=float,
             dest="disk_cache_hit_weight",
+            env=self.env,
         )
         add_argument(
             g,
@@ -431,6 +448,7 @@ def add_arguments(self, parser) -> None:
                 "Higher values promote more randomness, and 0 falls back to deterministic."
             ),
             arg_type=float,
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -444,6 +462,7 @@ def add_arguments(self, parser) -> None:
             ),
             dest="use_kv_events",
             obsolete_flag="--kv-events",
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -454,6 +473,7 @@ def add_arguments(self, parser) -> None:
                 "KV Router: Enable best-effort active-sequence synchronization through "
                 "the Runtime event plane."
             ),
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -466,6 +486,7 @@ def add_arguments(self, parser) -> None:
                 "By default, active blocks are tracked for load balancing."
             ),
             obsolete_flag="--track-active-blocks",
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -480,6 +501,7 @@ def add_arguments(self, parser) -> None:
                 "suffix; shared prompt blocks retain full weight."
             ),
             obsolete_flag="--track-output-blocks",
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -493,6 +515,7 @@ def add_arguments(self, parser) -> None:
                 "(when KV cache reuse is not expected)."
             ),
             obsolete_flag="--assume-kv-reuse",
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -505,6 +528,7 @@ def add_arguments(self, parser) -> None:
                 "Use --no-router-track-prefill-tokens to ignore prompt tokens in router "
                 "prefill-token load, queue pressure, and active_prefill_tokens metrics."
             ),
+            env=self.env,
         )
         add_argument(
             g,
@@ -513,20 +537,23 @@ def add_arguments(self, parser) -> None:
             default="public-xxh3-v1",
             choices=["public-xxh3-v1", "keyed-xxh3-v1"],
             help="KV Router: Hash function for router-derived active-sequence identities.",
+            env=self.env,
         )
         add_argument(
             g,
             flag_name="--router-tracking-key-file",
             env_var="DYN_ROUTER_TRACKING_KEY_FILE",
             default=None,
             help="KV Router: File containing the 32-byte provider tracking key.",
+            env=self.env,
         )
         add_argument(
             g,
             flag_name="--router-tracking-key-id",
             env_var="DYN_ROUTER_TRACKING_KEY_ID",
             default=None,
             help="KV Router: Provider-managed tracking-key epoch identifier.",
+            env=self.env,
         )
         add_argument(
             g,
@@ -540,6 +567,7 @@ def add_arguments(self, parser) -> None:
                 "'none' keeps static prompt load accounting. "
                 "'ais' decays the oldest active prefill request using AISimulate-predicted duration."
             ),
+            env=self.env,
         )
         add_argument(
             g,
@@ -551,6 +579,7 @@ def add_arguments(self, parser) -> None:
                 "Only used when --no-router-kv-events is set."
             ),
             arg_type=float,
+            env=self.env,
         )
         add_argument(
             g,
@@ -564,6 +593,7 @@ def add_arguments(self, parser) -> None:
                 "the existing time-based behavior; 'lru' uses each worker rank's "
                 "advertised KV capacity. Side indexers remain TTL-only."
             ),
+            env=self.env,
         )
         add_argument(
             g,
@@ -588,6 +618,7 @@ def add_arguments(self, parser) -> None:
                 "explicitly for predictable semantics, or use a smaller threshold."
             ),
             arg_type=nullable_float,
+            env=self.env,
         )
         add_argument(
             g,
@@ -602,6 +633,7 @@ def add_arguments(self, parser) -> None:
                 "router_queue_threshold is set."
             ),
             arg_type=str,
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -616,6 +648,7 @@ def add_arguments(self, parser) -> None:
                 "for each request."
             ),
             dest="conditional_disagg_enabled",
+            env=self.env,
         )
         add_argument(
             g,
@@ -631,6 +664,7 @@ def add_arguments(self, parser) -> None:
             arg_type=_conditional_disagg_config_arg,
             dest="conditional_disagg_config",
             metavar="JSON",
+            env=self.env,
         )
         add_argument(
             g,
@@ -647,6 +681,7 @@ def add_arguments(self, parser) -> None:
                 "Use 'default' for Dynamo's built-in worker selector."
             ),
             arg_type=str,
+            env=self.env,
         )
         add_argument(
             g,
@@ -663,6 +698,7 @@ def add_arguments(self, parser) -> None:
                 "Use 'default' for Dynamo's built-in worker selector."
             ),
             arg_type=str,
+            env=self.env,
         )
         add_argument(
             g,
@@ -675,6 +711,7 @@ def add_arguments(self, parser) -> None:
                 "approximate routing when --no-router-kv-events is set."
             ),
             arg_type=int,
+            env=self.env,
         )
         add_argument(
             g,
@@ -692,6 +729,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=str,
             choices=["fcfs", "wspt"],
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -703,6 +741,7 @@ def add_arguments(self, parser) -> None:
                 "component via the request plane instead of maintaining a local radix tree."
             ),
             dest="use_remote_indexer",
+            env=self.env,
         )
         add_negatable_bool_argument(
             g,
@@ -715,6 +754,7 @@ def add_arguments(self, parser) -> None:
                 "index for routing decisions yet."
             ),
             dest="enable_session_prefix_index",
+            env=self.env,
         )
         add_argument(
             g,
@@ -730,6 +770,7 @@ def add_arguments(self, parser) -> None:
                 "The default policy uses 0.5 when shared cache is enabled."
             ),
             arg_type=float,
+            env=self.env,
         )
         add_argument(
             g,
@@ -744,6 +785,7 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=str,
             choices=["none", "hicache"],
+            env=self.env,
         )
         add_argument(
             g,
@@ -756,4 +798,5 @@ def add_arguments(self, parser) -> None:
                 "Independent of --router-ttl-secs, which covers pure approximate mode."
             ),
             arg_type=float,
+            env=self.env,
         )