feat(model): add shared NGC fetching support with vLLM/SGLang integration

components/src/dynamo/common/model_fetch.py

@@ -17,6 +17,17 @@
 
 logger = logging.getLogger(__name__)
 
+NGC_URI_PREFIX = "ngc://"
+
+
+def needs_local_model_path(model: str) -> bool:
+    """Whether an engine must be given the fetched local path instead of ``model``.
+
+    Engines resolve Hugging Face repo IDs against the HF cache themselves, but
+    cannot resolve ``ngc://`` URIs.
+    """
+    return model.startswith(NGC_URI_PREFIX)
+
 
 def _fetch_model_process_main(
     remote_name: str,

components/src/dynamo/common/tests/test_model_fetch.py

@@ -0,0 +1,22 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for identifying model sources that engines cannot resolve themselves."""
+
+import pytest
+
+from dynamo.common.model_fetch import needs_local_model_path
+
+pytestmark = [pytest.mark.pre_merge, pytest.mark.unit, pytest.mark.gpu_0]
+
+
+@pytest.mark.parametrize(
+    ("model", "expected"),
+    [
+        ("ngc://nvstaging/nim/nemotron-3-ultra:hf-rl-052726-nvfp4-e9744cc", True),
+        ("Qwen/Qwen3-0.6B", False),
+        ("/data/llms/Qwen3-0.6B", False),
+    ],
+)
+def test_needs_local_model_path(model: str, expected: bool) -> None:
+    assert needs_local_model_path(model) is expected

components/src/dynamo/sglang/args.py

@@ -26,7 +26,7 @@
 from dynamo.common.configuration.groups.runtime_args import DynamoRuntimeArgGroup
 from dynamo.common.configuration.utils import split_served_model_names
 from dynamo.common.constants import DisaggregationMode
-from dynamo.common.model_fetch import fetch_model
+from dynamo.common.model_fetch import fetch_model, needs_local_model_path
 from dynamo.common.snapshot.lifecycle import (
     configure_snapshot_capture_env,
     is_snapshot_enabled,
@@ -62,6 +62,8 @@ class DynamoConfig(DynamoRuntimeConfig, DynamoSGLangConfig):
 
     component: str
     diffusion_worker: bool = False
+    # Preserve NGC identity for registration while SGLang loads the local path.
+    model_source_uri: Optional[str] = None
     # Whether this worker publishes KV events. Distinct from the router-side
     # `use_kv_events` on `router_advertisement`, which means the router
     # subscribes to them -- the reason the two live on separate objects.
@@ -194,10 +196,16 @@ def is_object_storage_path(model_path: str) -> bool:
 
 
 def should_fetch_model(args: Any, model_path: str) -> bool:
+    """Prefetch NGC and native-loader models, leaving HF P2P to ModelExpress."""
     if os.path.exists(model_path):
         return False
     if is_object_storage_path(model_path):
         return False
+    # SGLang resolves config/tokenizer files before invoking its weight loader.
+    # NGC therefore needs a local directory even with the ModelExpress plugin;
+    # fetch weights too so the plugin's native fallback can read that directory.
+    if needs_local_model_path(model_path):
+        return True
     return not use_modelexpress_remote_instance(args)
 
 
@@ -665,19 +673,30 @@ async def parse_args(args: list[str]) -> Config:
                 served_names[1:],
             )
 
+    if needs_local_model_path(model_path) and (
+        not served_names or ":" in served_names[0]
+    ):
+        raise ValueError(
+            "NGC models require a served model name without ':' in SGLang. "
+            "Pass --served-model-name my-model."
+        )
+
     # Name the model — falls back to model_path only if neither
     # --served-model-name nor an env var supplied one.
     if not parsed_args.served_model_name:
         parsed_args.served_model_name = model_path
     # Download the model if necessary using modelexpress.
-    # We don't set `parsed_args.model_path` to the local path fetch_model returns
-    # because sglang will send this to its pipeline-parallel workers, which may
-    # not have the local path.
+    # For HF names we don't set `parsed_args.model_path` to the local path
+    # fetch_model returns, because sglang will send this to its pipeline-parallel
+    # workers, which may not have the local path.
     # sglang will attempt to download the model again, but find it in the HF cache.
-    # For non-HF models use a path instead of an HF name, and ensure all workers have
-    # that path (ideally via a shared folder).
+    # sglang cannot resolve `ngc://` names, so those use the local path instead;
+    # every worker needs that path (ideally via a shared folder).
     if should_fetch_model(parsed_args, model_path):
-        await fetch_model(model_path)
+        local_path = await fetch_model(model_path)
+        if needs_local_model_path(model_path):
+            dynamo_config.model_source_uri = model_path
+            parsed_args.model_path = model_path = local_path
 
     snapshot_enabled = is_snapshot_enabled()
     if snapshot_enabled:

components/src/dynamo/sglang/register.py

@@ -222,7 +222,8 @@ async def _register_model_with_runtime_config(
             input_type,
             output_type,
             endpoint,
-            _register_model_source_path(engine, server_args),
+            dynamo_args.model_source_uri
+            or _register_model_source_path(engine, server_args),
             server_args.served_model_name,
             kv_cache_block_size=kv_cache_block_size,
             runtime_config=runtime_config,

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

@@ -9,6 +9,7 @@
 import sys
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 import torch
@@ -1638,6 +1639,70 @@ def test_should_fetch_model_skips_sglang_modelexpress_remote_instance():
     assert should_fetch_model(args, "Qwen/Qwen3-0.6B") is False
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("load_format", ["auto", "remote_instance"])
+async def test_parse_args_resolves_ngc_before_server_args(
+    monkeypatch, tmp_path, load_format
+):
+    model_uri = "ngc://test-org/test-team/test-model:v1"
+    local_path = str(tmp_path)
+    fetch = AsyncMock(return_value=local_path)
+    monkeypatch.setattr(sglang_args, "fetch_model", fetch)
+    monkeypatch.delenv(SNAPSHOT_CONTROL_DIR_ENV, raising=False)
+
+    def resolve(parsed_args):
+        assert parsed_args.model_path == local_path
+        return _dcp_server_args_stub(
+            model_path=parsed_args.model_path,
+            served_model_name=parsed_args.served_model_name,
+            load_format=parsed_args.load_format,
+            remote_instance_weight_loader_backend=(
+                parsed_args.remote_instance_weight_loader_backend
+            ),
+        )
+
+    monkeypatch.setattr(sglang_args.ServerArgs, "from_cli_args", resolve)
+    argv = [
+        "--model",
+        model_uri,
+        "--load-format",
+        load_format,
+        "--served-model-name",
+        "my-model,alternate:model",
+    ]
+    if load_format == "remote_instance":
+        argv.extend(["--remote-instance-weight-loader-backend", "modelexpress"])
+
+    config = await parse_args(argv)
+
+    # A full download supplies the plugin's native fallback as well as config.
+    fetch.assert_awaited_once_with(model_uri)
+    assert config.server_args.model_path == local_path
+    assert config.dynamo_args.model_source_uri == model_uri
+    assert config.server_args.served_model_name == "my-model"
+    assert config.dynamo_args.served_model_aliases == ["alternate:model"]
+    assert config.server_args.load_format == load_format
+    if load_format == "remote_instance":
+        assert use_modelexpress_remote_instance(config.server_args)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("served_model_name", [None, " ", "model:adapter"])
+async def test_ngc_requires_valid_served_name_before_fetch(
+    monkeypatch, served_model_name
+):
+    fetch = AsyncMock()
+    monkeypatch.setattr(sglang_args, "fetch_model", fetch)
+    argv = ["--model", "ngc://test-org/test-team/test-model:v1"]
+    if served_model_name is not None:
+        argv.extend(["--served-model-name", served_model_name])
+
+    with pytest.raises(ValueError, match="--served-model-name my-model"):
+        await parse_args(argv)
+
+    fetch.assert_not_awaited()
+
+
 @pytest.mark.parametrize(
     "model_path",
     [
@@ -1659,7 +1724,11 @@ def test_should_fetch_model_keeps_default_non_local_fetch():
 
 
 @pytest.mark.asyncio
-async def test_register_model_uses_metadata_only_for_sglang_modelexpress(monkeypatch):
+@pytest.mark.parametrize("model_source", ["hf", "local", "ngc"])
+@pytest.mark.parametrize("worker_name", ["Aggregated", "Encode"])
+async def test_register_model_preserves_source_with_sglang_modelexpress(
+    monkeypatch, tmp_path, model_source, worker_name
+):
     if sglang_register is None:
         pytest.skip("dynamo.sglang.register is unavailable")
 
@@ -1669,35 +1738,44 @@ async def fake_get_runtime_config(engine, server_args, dynamo_args):
         return None
 
     async def fake_register_model(*args, **kwargs):
+        captured["args"] = args
         captured["kwargs"] = kwargs
 
     monkeypatch.setattr(sglang_register, "get_runtime_config", fake_get_runtime_config)
     monkeypatch.setattr(sglang_register, "register_model", fake_register_model)
 
+    model_path = "Qwen/Qwen3-0.6B" if model_source == "hf" else str(tmp_path)
+    model_uri = (
+        "ngc://test-org/test-team/test-model:v1" if model_source == "ngc" else None
+    )
     server_args = SimpleNamespace(
-        model_path="Qwen/Qwen3-0.6B",
-        served_model_name="Qwen/Qwen3-0.6B",
+        model_path=model_path,
+        served_model_name="my-model",
         context_length=4096,
         page_size=64,
         dcp_size=8,
         load_format="remote_instance",
         remote_instance_weight_loader_backend="modelexpress",
     )
     dynamo_args = SimpleNamespace(
+        model_source_uri=model_uri,
         use_sglang_tokenizer=False,
         frontend_decoding=False,
         custom_jinja_template=None,
     )
 
     result = await sglang_register._register_model_with_runtime_config(
-        engine=SimpleNamespace(),
+        engine=None if worker_name == "Encode" else SimpleNamespace(),
         endpoint=SimpleNamespace(),
         server_args=server_args,
         dynamo_args=dynamo_args,
-        worker_type=sglang_register.WorkerType.Aggregated,
+        worker_type=getattr(sglang_register.WorkerType, worker_name),
     )
 
     assert result is True
+    assert captured["args"][3] == (model_uri or model_path)
+    assert captured["args"][4] == "my-model"
+    assert server_args.model_path == model_path
     assert captured["kwargs"]["ignore_weights"] is True
     assert captured["kwargs"]["kv_cache_block_size"] == 512
 
@@ -1732,6 +1810,7 @@ async def fake_register_model(*args, **kwargs):
         remote_instance_weight_loader_backend=None,
     )
     dynamo_args = SimpleNamespace(
+        model_source_uri=None,
         use_sglang_tokenizer=False,
         frontend_decoding=False,
         custom_jinja_template=None,

components/src/dynamo/vllm/args.py

@@ -28,6 +28,7 @@
     DynamoRuntimeConfig,
 )
 from dynamo.common.configuration.utils import split_served_model_names
+from dynamo.common.model_fetch import fetch_model, needs_local_model_path
 from dynamo.common.utils.runtime import parse_endpoint
 from dynamo.vllm.backend_args import DynamoVllmArgGroup, DynamoVllmConfig
 from dynamo.vllm.benchmark_points import RANDOM_KDA_WORKER
@@ -66,6 +67,12 @@ class Config(DynamoRuntimeConfig, DynamoVllmConfig):
     # rest vLLM args
     engine_args: AsyncEngineArgs
 
+    @property
+    def model_source_path(self) -> str:
+        return (
+            self.engine_args.model if needs_local_model_path(self.model) else self.model
+        )
+
     def validate(self) -> None:
         DynamoRuntimeConfig.validate(self)
         DynamoVllmConfig.validate(self)
@@ -78,13 +85,38 @@ def _preprocess_for_encode_config(config: Config) -> Dict[str, Any]:
 
 
 def parse_args(argv: list[str] | None = None) -> Config:
-    """Parse command-line arguments for the vLLM backend.
+    """Parse vLLM arguments without fetching; workers use parse_args_with_model_fetch.
 
     Args:
         argv: Command-line arguments.  ``None`` means ``sys.argv[1:]``.
     Returns:
         Config: Parsed configuration object.
     """
+    return _build_config(*_parse_cli_args(argv))
+
+
+async def parse_args_with_model_fetch(argv: list[str] | None = None) -> Config:
+    """Parse worker arguments, resolving NGC sources before engine validation.
+
+    Only metadata is fetched here; worker startup fetches weights after
+    validation. Hugging Face sources retain their existing acquisition path.
+
+    Args:
+        argv: Command-line arguments. ``None`` means ``sys.argv[1:]``.
+
+    Returns:
+        The validated configuration, retaining the original source in ``model``.
+    """
+    dynamo_config, vllm_args = _parse_cli_args(argv)
+    if needs_local_model_path(dynamo_config.model):
+        vllm_args.model = await fetch_model(dynamo_config.model, ignore_weights=True)
+        if not vllm_args.served_model_name:
+            vllm_args.served_model_name = [dynamo_config.model]
+    return _build_config(dynamo_config, vllm_args)
+
+
+def _parse_cli_args(argv: list[str] | None) -> tuple[Config, argparse.Namespace]:
+    """Split Dynamo and vLLM arguments while preserving the original model source."""
     dynamo_runtime_argspec = DynamoRuntimeArgGroup()
     dynamo_vllm_argspec = DynamoVllmArgGroup()
 
@@ -128,7 +160,10 @@ def parse_args(argv: list[str] | None = None) -> Config:
     # vllm will update the model name to the full path of the model, which will break the dynamo logic,
     # as we use the model name as served_model_name (if served_model_name is not set)
     dynamo_config.model = vllm_args.model
+    return dynamo_config, vllm_args
+
 
+def _build_config(dynamo_config: Config, vllm_args: argparse.Namespace) -> Config:
     enable_kv_cache_metadata_compat()
     engine_config = AsyncEngineArgs.from_cli_args(vllm_args)
 

components/src/dynamo/vllm/handlers.py

@@ -1273,7 +1273,7 @@ def __init__(
         self._deferred_aborts: dict[str, _DeferredAbort] = {}
 
         self._multimodal_request_processor = VllmMultimodalRequestProcessor(
-            model=config.model,
+            model=config.model_source_path,
             engine_client=engine,
             enable_multimodal=enable_multimodal,
             enable_frontend_decoding=enable_frontend_decoding,
@@ -1362,15 +1362,15 @@ def _load_custom_encoder(self, config: Config) -> None:
             config.engine_args,
         )
         encoder = AsyncVisionEncoder(backend)
-        encoder.load(config.model)
+        encoder.load(config.model_source_path)
         # Assign only after a successful load so a failed load (which already shut
         # its own thread down) leaves _custom_encoder None.
         self._custom_encoder = encoder
         self._custom_encoder_adapter = adapter
         logger.info(
             "Loaded CustomEncoder %s from %s with %s",
             custom_encoder_class,
-            config.model,
+            config.model_source_path,
             type(adapter).__name__,
         )