fix(vllm): distinguish undeclared RL weight versions

.github/actions/setup-dynamo-operator/action.yml

@@ -597,11 +597,14 @@ runs:
         # propagation can lag in either direction. The READY column confirms
         # container readiness reported by each webhook-aware probe; the
         # admission smoke tests below are the authoritative end-to-end gate.
+        # Evicted and completed Pods can remain after their replacements are
+        # ready. Count only active Pods so those stale objects cannot block setup.
         end=$((SECONDS + 1800))
         while true; do
           POD_STATUSES=()
           mapfile -t POD_STATUSES < <(
-            kubectl get pods -n "${NAMESPACE}" --no-headers 2>/dev/null \
+            kubectl get pods -n "${NAMESPACE}" --no-headers \
+              --field-selector=status.phase!=Failed,status.phase!=Succeeded 2>/dev/null \
               | awk '
                   $1 ~ /^dynamo-platform-dynamo-operator-controller-manager-/ ||
                   $1 ~ /^grove-operator-/ {

components/src/dynamo/sglang/args.py

@@ -549,8 +549,13 @@ async def parse_args(args: list[str]) -> Config:
     if should_fetch_model(parsed_args, model_path):
         await fetch_model(model_path)
 
-    if is_snapshot_enabled():
+    snapshot_enabled = is_snapshot_enabled()
+    if snapshot_enabled:
         configure_snapshot_capture_env()
+        # SGLang reads these raw fields before late resolution, so snapshot mode
+        # must set them before ServerArgs creation.
+        parsed_args.enable_memory_saver = True
+        parsed_args.enable_forward_pass_metrics = False
 
     # TODO: sglang downloads the model in `from_cli_args`, which means we had to
     # fetch_model (download the model) here, in `parse_args`. `parse_args` should not
@@ -571,7 +576,11 @@ async def parse_args(args: list[str]) -> Config:
         parsed_args.enable_memory_saver = True
 
     fpm_source = _forward_pass_metrics_source(dynamo_config)
-    if fpm_source and not getattr(parsed_args, "enable_forward_pass_metrics", False):
+    if (
+        not snapshot_enabled
+        and fpm_source
+        and not getattr(parsed_args, "enable_forward_pass_metrics", False)
+    ):
         parsed_args.enable_forward_pass_metrics = True
         logging.info("Enabled forward_pass_metrics from %s", fpm_source)
 

components/src/dynamo/sglang/init_llm.py

@@ -65,18 +65,20 @@ async def init_decode(
     if snapshot_engine is not None:
         engine = snapshot_engine
         load_time = 0.0
-        if getattr(server_args, "enable_forward_pass_metrics", False):
-            raise RuntimeError(
-                "Snapshot ServerArgs must disable forward-pass metrics before "
-                "engine creation"
-            )
     else:
         set_forward_pass_metrics_worker_id(server_args, generate_endpoint)
         start_time = time.time()
         engine = sgl.Engine(server_args=server_args)
         load_time = time.time() - start_time
 
     server_args = config.use_resolved_server_args(engine.server_args)
+    if snapshot_engine is not None and getattr(
+        server_args, "enable_forward_pass_metrics", False
+    ):
+        raise RuntimeError(
+            "Snapshot ServerArgs must disable forward-pass metrics before "
+            "engine creation"
+        )
 
     if server_args.enable_trace:
         set_global_trace_level(dynamo_args.sglang_trace_level)
@@ -224,18 +226,20 @@ async def init_prefill(
     if snapshot_engine is not None:
         engine = snapshot_engine
         load_time = 0.0
-        if getattr(server_args, "enable_forward_pass_metrics", False):
-            raise RuntimeError(
-                "Snapshot ServerArgs must disable forward-pass metrics before "
-                "engine creation"
-            )
     else:
         set_forward_pass_metrics_worker_id(server_args, generate_endpoint)
         start_time = time.time()
         engine = sgl.Engine(server_args=server_args)
         load_time = time.time() - start_time
 
     server_args = config.use_resolved_server_args(engine.server_args)
+    if snapshot_engine is not None and getattr(
+        server_args, "enable_forward_pass_metrics", False
+    ):
+        raise RuntimeError(
+            "Snapshot ServerArgs must disable forward-pass metrics before "
+            "engine creation"
+        )
 
     if server_args.enable_trace:
         set_global_trace_level(dynamo_args.sglang_trace_level)

components/src/dynamo/sglang/tests/test_sglang_unit.py

@@ -3,12 +3,14 @@
 
 """Unit tests for SGLang backend components."""
 
+import asyncio
 import logging
 import os
 import re
 import sys
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 import torch
@@ -165,6 +167,54 @@ def test_resolved_server_args_uses_declarative_view(monkeypatch):
     assert raw_server_args.page_size is None
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("initializer", ["init_decode", "init_prefill"])
+@pytest.mark.parametrize("raw_metrics,resolved_metrics", [(True, False), (False, True)])
+async def test_snapshot_metrics_check_uses_resolved_configuration(
+    monkeypatch, initializer, raw_metrics, resolved_metrics
+):
+    import dynamo.sglang.init_llm as sglang_init_llm
+
+    raw_args = SimpleNamespace(
+        node_rank=0,
+        disaggregation_mode="null",
+        enable_forward_pass_metrics=raw_metrics,
+    )
+    resolved_args = SimpleNamespace(
+        node_rank=0,
+        enable_trace=False,
+        enable_forward_pass_metrics=resolved_metrics,
+    )
+    monkeypatch.setattr(sglang_compat, "sglang_resolved_view", lambda _: resolved_args)
+    config = sglang_args.Config(
+        raw_args,
+        SimpleNamespace(namespace="test", component="worker", endpoint="generate"),
+    )
+    engine = SimpleNamespace(server_args=raw_args)
+
+    class ReachedMetricsSetup(Exception):
+        pass
+
+    setup_metrics = AsyncMock(side_effect=ReachedMetricsSetup)
+    monkeypatch.setattr(sglang_init_llm, "setup_sgl_metrics", setup_metrics)
+    initialize = getattr(sglang_init_llm, initializer)
+
+    if resolved_metrics:
+        with pytest.raises(RuntimeError, match="must disable forward-pass metrics"):
+            await initialize(
+                Mock(), config, asyncio.Event(), [], snapshot_engine=engine
+            )
+        setup_metrics.assert_not_awaited()
+    else:
+        with pytest.raises(ReachedMetricsSetup):
+            await initialize(
+                Mock(), config, asyncio.Event(), [], snapshot_engine=engine
+            )
+        setup_metrics.assert_awaited_once()
+
+    assert config.server_args is resolved_args
+
+
 def test_compat_uses_current_sglang_model_config_accessor(monkeypatch):
     expected = SimpleNamespace(is_multimodal=True)
     server_args = SimpleNamespace()
@@ -492,6 +542,42 @@ def resolve(parsed_args):
     assert config.server_args.incremental_streaming_output is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("snapshot_enabled", "explicit_memory_saver", "expected"),
+    [(False, False, False), (True, False, True), (False, True, True)],
+)
+async def test_snapshot_memory_saver_is_enabled_before_resolution(
+    monkeypatch, tmp_path, snapshot_enabled, explicit_memory_saver, expected
+):
+    monkeypatch.delenv("DYN_GMS_USE_V1", raising=False)
+    monkeypatch.delenv(SNAPSHOT_CONTROL_DIR_ENV, raising=False)
+    if snapshot_enabled:
+        monkeypatch.setenv(SNAPSHOT_CONTROL_DIR_ENV, str(tmp_path))
+    monkeypatch.setattr(sglang_args, "configure_snapshot_capture_env", lambda: None)
+
+    def resolve(parsed_args):
+        # Engine chooses the subprocess allocator from the raw field; a late
+        # resolved-only override does not configure LD_PRELOAD for its children.
+        assert parsed_args.enable_memory_saver is expected
+        return SimpleNamespace(
+            enable_memory_saver=parsed_args.enable_memory_saver,
+            disaggregation_mode="null",
+            dllm_algorithm=None,
+            kv_events_config=None,
+            get_model_config=lambda: SimpleNamespace(is_multimodal=False),
+        )
+
+    monkeypatch.setattr(sglang_args.ServerArgs, "from_cli_args", resolve)
+    args = ["--model", str(tmp_path)]
+    if explicit_memory_saver:
+        args.append("--enable-memory-saver")
+
+    config = await parse_args(args)
+
+    assert config.server_args.enable_memory_saver is expected
+
+
 @pytest.mark.asyncio
 async def test_parse_args_applies_dynamo_defaults_before_resolution(
     monkeypatch, mock_sglang_cli

components/src/dynamo/vllm/handlers.py

@@ -151,6 +151,8 @@ def _discard_orphan_result(fut: "asyncio.Future[dict]") -> None:
         "weight_version",
     }
 )
+# An object sentinel cannot collide with a caller-supplied version.
+_WEIGHT_VERSION_UNDECLARED: Final = object()
 
 
 def build_prompt_tokens_details(
@@ -1185,7 +1187,7 @@ def __init__(
         # to prevent both bypassing the check before either inserts (atomicity).
         self._lora_capacity_guard = asyncio.Lock()
         self._paused: bool = False
-        self._weight_version: str = "initial"
+        self._weight_version: Any = _WEIGHT_VERSION_UNDECLARED
 
         embedding_loader = self.init_embedding_loader(config, encode_worker_client)
 
@@ -2000,16 +2002,41 @@ async def abort_request(self, body: dict) -> dict:
             logger.error(f"[RL] Failed to abort request {request_id}: {e}")
             return {"status": "error", "message": str(e)}
 
+    def _declare_weight_version(self, version: Any) -> None:
+        self._weight_version = version
+
     async def get_weight_version(self, body: dict) -> dict:
-        """Return the current weight version tag."""
+        """Report the worker's current declared weight-version state."""
+        if body is None:
+            body = {}
+        elif not isinstance(body, dict):
+            return {
+                "status": "error",
+                "message": "request body must be a JSON object",
+            }
+        version = self._weight_version
+        declared = version is not _WEIGHT_VERSION_UNDECLARED
+        return {
+            "status": "ok",
+            "version": version if declared else None,
+            "version_declared": declared,
+        }
+
+    async def set_weight_version(self, body: dict) -> dict:
+        """Validate and declare a weight version for subsequent requests."""
         if body is None:
             body = {}
         elif not isinstance(body, dict):
             return {
                 "status": "error",
                 "message": "request body must be a JSON object",
             }
-        return {"status": "ok", "version": getattr(self, "_weight_version", "initial")}
+        if "weight_version" not in body:
+            return {"status": "error", "message": "Missing 'weight_version' in body"}
+        version = body["weight_version"]
+        self._declare_weight_version(version)
+        logger.info("[RL] Weight version declared (version=%s)", version)
+        return {"status": "ok", "version": version}
 
     async def update_weights_from_disk(self, body: dict) -> dict:
         """Load weights from a shared filesystem checkpoint."""
@@ -2048,7 +2075,8 @@ async def update_weights_from_disk(self, body: dict) -> dict:
                 # weights is now stale and must not be reused. Invalidate it
                 # while still holding _pause_lock (generation is paused).
                 await self.engine_client.reset_prefix_cache()
-                self._weight_version = version
+                if "weight_version" in body:
+                    self._declare_weight_version(version)
                 logger.info(
                     f"[RL] Weights loaded from {path} (version={version}, rpc={rpc})"
                 )
@@ -2111,7 +2139,8 @@ async def update_weights_from_distributed(self, body: dict) -> dict:
                     # Weights changed: stale prefix/KV cache must be invalidated
                     # before resume so it is not reused under the new weights.
                     await self.engine_client.reset_prefix_cache()
-                self._weight_version = version
+                if "weight_version" in body:
+                    self._declare_weight_version(version)
                 logger.info(
                     f"[RL] Weights received via distributed "
                     f"(version={version}, rpc={rpc})"

components/src/dynamo/vllm/tests/test_vllm_backend_args.py

@@ -14,7 +14,6 @@
 
 import pytest
 
-from dynamo.vllm.args import parse_args
 from dynamo.vllm.backend_args import (
     DisaggregationMode,
     DynamoVllmArgGroup,
@@ -467,41 +466,6 @@ def test_no_op_when_classify_worker_disabled(self):
         config._validate_classify_worker_exclusivity()
 
 
-@pytest.mark.usefixtures("vllm_cpu_platform_when_no_accelerator")
-class TestParseArgsLoraExclusivity:
-    """The --enable-lora exclusivity rules must fire on the real CLI path.
-
-    The tests above call the validators directly with engine_args already set,
-    so they pass even when parse_args never supplies engine_args before
-    validate() runs. These drive the whole command line instead, which is the
-    only place that ordering is observable.
-    """
-
-    @staticmethod
-    def _parse(extra_argv):
-        return parse_args(["--model", "Qwen/Qwen3-0.6B", *extra_argv])
-
-    def test_realtime_with_enable_lora_is_rejected(self):
-        with pytest.raises(ValueError, match="enable-lora"):
-            self._parse(["--realtime", "--enable-lora"])
-
-    def test_classify_worker_with_enable_lora_is_rejected(self):
-        """Kept separate from the --realtime case: the classify rule may be
-        removed once LoRA is supported on pooling-family workers, and the
-        --realtime rule is independent of that."""
-        with pytest.raises(ValueError, match="enable-lora"):
-            self._parse(["--classify-worker", "--enable-lora"])
-
-    def test_enable_lora_alone_is_accepted(self):
-        config = self._parse(["--enable-lora"])
-        assert config.engine_args.enable_lora is True
-
-    def test_realtime_alone_is_accepted(self):
-        config = self._parse(["--realtime"])
-        assert config.realtime is True
-        assert not config.engine_args.enable_lora
-
-
 class TestValidateCustomEncoder:
     """--custom-encoder-class is an in-process, aggregated-only multimodal
     component, so validation must require --enable-multimodal and reject any

components/src/dynamo/vllm/tests/test_vllm_parse_args_lora.py

@@ -0,0 +1,48 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+
+from dynamo.vllm.args import parse_args
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.vllm,
+    pytest.mark.pre_merge,
+    pytest.mark.gpu_0,
+    pytest.mark.multimodal,
+    pytest.mark.usefixtures("vllm_cpu_platform_when_no_accelerator"),
+]
+
+
+class TestParseArgsLoraExclusivity:
+    """The --enable-lora exclusivity rules must fire on the real CLI path.
+
+    Direct validator tests supply engine_args themselves, so they pass even
+    when parse_args never supplies engine_args before validate() runs. These
+    tests drive the whole command line to exercise that ordering.
+    """
+
+    @staticmethod
+    def _parse(extra_argv):
+        return parse_args(["--model", "Qwen/Qwen3-0.6B", *extra_argv])
+
+    def test_realtime_with_enable_lora_is_rejected(self):
+        with pytest.raises(ValueError, match="enable-lora"):
+            self._parse(["--realtime", "--enable-lora"])
+
+    def test_classify_worker_with_enable_lora_is_rejected(self):
+        """Kept separate from the --realtime case: the classify rule may be
+        removed once LoRA is supported on pooling-family workers, and the
+        --realtime rule is independent of that."""
+        with pytest.raises(ValueError, match="enable-lora"):
+            self._parse(["--classify-worker", "--enable-lora"])
+
+    def test_enable_lora_alone_is_accepted(self):
+        config = self._parse(["--enable-lora"])
+        assert config.engine_args.enable_lora is True
+
+    def test_realtime_alone_is_accepted(self):
+        config = self._parse(["--realtime"])
+        assert config.realtime is True
+        assert not config.engine_args.enable_lora