[azure-ai-ml] Fix internal pipeline Command dropping interactive services during serialization

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features Added
 
 ### Bugs Fixed
+- Fixed internal pipeline `Command` node dropping node-level interactive `services` (SSH, JupyterLab, TensorBoard, VS Code, etc.) during serialization, which prevented interactive endpoints from being created for Singularity jobs. The `services` are now serialized into the pipeline REST request and round-tripped on deserialization, matching the public `Command` node behavior.
 - Fixed `MLClient.jobs.create_or_update`, `archive`, and `restore` failing for previously-fetched jobs across all job types by routing metadata-only edits through the RunHistory PATCH endpoint.
 - Fixed `DeploymentTemplate.creation_context` always being `None` when retrieved via `get()` or `list()`. The created/modified timestamps and identity returned by the service (as `createdTime` / `modifiedTime` / `createdBy`) are now populated on `creation_context`, making `DeploymentTemplate` consistent with `Model` and `Environment`.
 - Fixed `deployment_templates.list(name=...)` raising `AttributeError: 'str' object has no attribute 'request_timeout'`. In the list response, `requestSettings` / `livenessProbe` / `readinessProbe` arrive as stringified dicts nested under `properties`; these are now parsed before conversion, giving `list()` parity with `models.list()` / `environments.list()`.

sdk/ml/azure-ai-ml/api.metadata.yml

@@ -1,3 +1,3 @@
 apiMdSha256: 81371e37ca26ea3c761bdb2104b7a0b89a0026216a97510bc16f533cf936e2d9
-parserVersion: 0.3.28
+parserVersion: 0.3.30
 pythonVersion: 3.12.10

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/command.py

@@ -10,7 +10,9 @@
 from ..._schema import PathAwareSchema
 from ..._schema.core.fields import DistributionField
 from ...entities import CommandJobLimits, JobResourceConfiguration
-from ...entities._util import get_rest_dict_for_node_attrs
+from ...entities._builders.command import _resolve_job_services
+from ...entities._job.job_service import JobServiceBase
+from ...entities._util import from_rest_dict_to_dummy_rest_object, get_rest_dict_for_node_attrs
 from .._schema.component import NodeType
 from ..entities.component import InternalComponent
 from ..entities.node import InternalBaseNode
@@ -24,13 +26,15 @@ class Command(InternalBaseNode):
 
     def __init__(self, **kwargs):
         node_type = kwargs.pop("type", None) or NodeType.COMMAND
+        services = kwargs.pop("services", None)
         super(Command, self).__init__(type=node_type, **kwargs)
         self._init = True
         self._resources = kwargs.pop("resources", JobResourceConfiguration())
         self._compute = kwargs.pop("compute", None)
         self._environment = kwargs.pop("environment", None)
         self._environment_variables = kwargs.pop("environment_variables", None)
         self._limits = kwargs.pop("limits", CommandJobLimits())
+        self._services = _resolve_job_services(services)
         self._init = False
 
     @property
@@ -108,6 +112,30 @@ def resources(self) -> JobResourceConfiguration:
     def resources(self, value: JobResourceConfiguration):
         self._resources = value
 
+    @property
+    def services(self) -> Optional[Dict]:
+        """The interactive services for the node.
+
+        This is an experimental parameter, and may change at any time.
+        Please see https://aka.ms/azuremlexperimental for more information.
+
+        :return: The interactive services for the node.
+        :rtype: Optional[Dict]
+        """
+        return self._services
+
+    @services.setter
+    def services(self, value: Optional[Dict]) -> None:
+        """Sets the interactive services for the node.
+
+        This is an experimental parameter, and may change at any time.
+        Please see https://aka.ms/azuremlexperimental for more information.
+
+        :param value: The interactive services for the node.
+        :type value: Optional[Dict]
+        """
+        self._services = _resolve_job_services(value)
+
     @classmethod
     def _picked_fields_from_dict_to_rest_object(cls) -> List[str]:
         return ["environment", "limits", "resources", "environment_variables"]
@@ -126,6 +154,9 @@ def _to_rest_object(self, **kwargs) -> dict:
                 "resources": get_rest_dict_for_node_attrs(self.resources, clear_empty_value=True),
             }
         )
+        services = get_rest_dict_for_node_attrs(self.services)
+        if services is not None:
+            rest_obj["services"] = services
         return rest_obj
 
     @classmethod
@@ -138,6 +169,17 @@ def _from_rest_object_to_init_params(cls, obj):
         # handle limits
         if "limits" in obj and obj["limits"]:
             obj["limits"] = CommandJobLimits._from_rest_object(obj["limits"])
+
+        # services, pipeline node rest object are dicts while _from_rest_job_services expect RestJobService
+        if "services" in obj and obj["services"]:
+            services = {}
+            for service_name, service in obj["services"].items():
+                # in rest object of a pipeline job, service will be transferred to a dict as
+                # it's attributes of a node, but JobService._from_rest_object expect a
+                # RestJobService, so we need to convert it back. Here we convert the dict to a
+                # dummy rest object which may work as a RestJobService instead.
+                services[service_name] = from_rest_dict_to_dummy_rest_object(service)
+            obj["services"] = JobServiceBase._from_rest_job_services(services)
         return obj
 
 

sdk/ml/azure-ai-ml/dev_requirements.txt

@@ -21,7 +21,7 @@ azure-mgmt-resourcegraph<9.0.0,>=2.0.0
 azure-mgmt-resource<23.0.0,>=3.0.0
 pytest-reportlog
 python-dotenv
-azureml-dataprep-rslex>=2.22.0; platform_python_implementation == "CPython" and python_version < "3.13"
+azureml-dataprep-rslex>=2.22.0; platform_python_implementation == "CPython" and python_version < "3.13" and (sys_platform != "darwin" or platform_machine != "arm64")
 azureml-dataprep-rslex>=2.22.0; platform_python_implementation == "PyPy" and python_version < "3.10"
 pip
 setuptools==77.0.3

sdk/ml/azure-ai-ml/tests/internal/unittests/test_pipeline_job.py

@@ -428,6 +428,30 @@ def test_singularity_component_in_pipeline(self):
             }
         }
 
+    def test_interactive_services_in_internal_command_node(self):
+        # regression test for ICM 822596251 / Bug 5466757: internal Command node dropped
+        # node-level interactive services (ssh/jupyterlab/etc.) during serialization.
+        from azure.ai.ml.entities import JupyterLabJobService, SshJobService
+
+        yaml_path = "./tests/test_configs/internal/command-component-ls/ls_command_component.yaml"
+        node_func: CommandComponent = load_component(yaml_path)
+        node = node_func()
+        node.services = {
+            "my_ssh": SshJobService(),
+            "my_jupyter": JupyterLabJobService(),
+        }
+
+        rest_obj = node._to_rest_object()
+        assert rest_obj["services"] == {
+            "my_ssh": {"job_service_type": "SSH"},
+            "my_jupyter": {"job_service_type": "JupyterLab"},
+        }
+
+        # services should round-trip back into typed JobService objects
+        init_params = Command._from_rest_object_to_init_params(dict(rest_obj))
+        assert isinstance(init_params["services"]["my_ssh"], SshJobService)
+        assert isinstance(init_params["services"]["my_jupyter"], JupyterLabJobService)
+
     def test_load_pipeline_job_with_internal_components_as_node(self):
         yaml_path = Path("./tests/test_configs/internal/helloworld/helloworld_component_scope.yml")
         scope_internal_func = load_component(source=yaml_path)