[ml] Fix jobs partial update across all job types

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

@@ -5,6 +5,7 @@
 ### Features Added
 
 ### Bugs Fixed
+- 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.
 
 ### Other Changes
 

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

@@ -10934,6 +10934,18 @@ namespace azure.ai.ml.operations
         @monitor_with_activity(ops_logger, 'Job.Stream', ActivityType.PUBLICAPI)
         def stream(self, name: str) -> None: ...
 
+        @distributed_trace
+        @monitor_with_telemetry_mixin(ops_logger, 'Job.Update', ActivityType.PUBLICAPI)
+        def update(
+                self, 
+                name: str, 
+                *, 
+                description: Optional[str] = ..., 
+                display_name: Optional[str] = ..., 
+                properties: Optional[Dict[str, str]] = ..., 
+                tags: Optional[Dict[str, str]] = ...
+            ) -> Job: ...
+
         @distributed_trace
         @monitor_with_telemetry_mixin(ops_logger, 'Job.Validate', ActivityType.PUBLICAPI)
         def validate(

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

@@ -1,3 +1,3 @@
-apiMdSha256: dc079c94f44c28406c444f39bd5501c997e56e593fd93c3176b08ca3c3ef04d3
+apiMdSha256: 7262033cadec8b494dfa9cca72d97a9dad84e1e465ce8c455782d9fab275a81f
 parserVersion: 0.3.28
 pythonVersion: 3.12.10

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py

@@ -654,7 +654,7 @@ def error_func(msg: str, no_personal_data_msg: str) -> ValidationException:
 
     @distributed_trace
     @monitor_with_telemetry_mixin(ops_logger, "Job.CreateOrUpdate", ActivityType.PUBLICAPI)
-    def create_or_update(
+    def create_or_update(  # pylint: disable=too-many-branches
         self,
         job: Job,
         *,
@@ -723,6 +723,67 @@ def create_or_update(
         if experiment_name is not None:
             job.experiment_name = experiment_name
 
+        # ---------------- TEMPORARY HOTFIX (metadata-only shortcut) ----------------
+        # Root bug: for a Job fetched via jobs.get(name) then mutated and passed to
+        # create_or_update, the serialize + MFE PUT round-trip fails for AutoML,
+        # Command, Pipeline, Sweep, Spark (each with a different serializer /
+        # comparator error). Until MFE + per-jobType serializers are fixed upstream,
+        # detect the "fetched job with only metadata changes" case and route through
+        # the RunHistory PATCH endpoint (same call used by ML Studio's portal for
+        # inline edits, and by the new jobs.update() API added alongside this fix).
+        #
+        # Shortcut is taken only when:
+        #   * the incoming job carries an ARM resource id (=> it was fetched, not
+        #     freshly created), AND
+        #   * the caller is not trying to change compute or experiment_name (these
+        #     are not supported by the shortcut and were rejected by MFE anyway).
+        # Any failure inside the shortcut falls through to the original path so
+        # brand-new job creation behavior is preserved bit-for-bit.
+        if getattr(job, "id", None) and getattr(job, "name", None) and compute is None and experiment_name is None:
+            job_name: str = cast(str, job.name)
+            patch_succeeded = False
+            try:
+                job_object = self._get_job(job_name)
+                if job_object.properties.job_type == RestJobType.PIPELINE:
+                    job_object = self._get_job_2401(job_name)
+                if _is_pipeline_child_job(job_object):
+                    raise PipelineChildJobError(job_id=job_object.id)
+
+                from azure.ai.ml._restclient.runhistory.models import CreateRun
+
+                self._runs_operations._operation.add_or_modify_by_experiment_name(
+                    subscription_id=self._operation_scope.subscription_id,
+                    resource_group_name=self._operation_scope.resource_group_name,
+                    workspace_name=self._workspace_name,
+                    experiment_name=job_object.properties.experiment_name,
+                    run_id=job_name,
+                    body=CreateRun(
+                        display_name=getattr(job, "display_name", None),
+                        description=getattr(job, "description", None),
+                        tags=getattr(job, "tags", None),
+                        properties=getattr(job, "properties", None),
+                    ),
+                )
+                patch_succeeded = True
+            except PipelineChildJobError:
+                raise
+            except Exception:  # pylint: disable=broad-exception-caught
+                # Only failures BEFORE and INCLUDING the RunHistory PATCH are swallowed
+                # (job not found, transient RunHistory error, etc.) so we can fall
+                # through to the original code path without regressing creation.
+                pass
+
+            if patch_succeeded:
+                # PATCH already persisted the metadata change server-side. Any failure
+                # in the refresh/resolve below must surface directly — falling back to
+                # the original MFE PUT path would re-execute the very round-trip this
+                # hotfix exists to avoid and would surface a misleading error.
+                refreshed = self._get_job(job_name)
+                if refreshed.properties.job_type == RestJobType.PIPELINE:
+                    refreshed = self._get_job_2401(job_name)
+                return self._resolve_azureml_id(Job._from_rest_object(refreshed))
+        # ------------------ END TEMPORARY HOTFIX --------------------------------
+
         if job.compute == LOCAL_COMPUTE_TARGET:
             job.environment_variables[COMMON_RUNTIME_ENV_VAR] = "true"  # type: ignore
 
@@ -850,8 +911,34 @@ def _archive_or_restore(self, name: str, is_archived: bool) -> None:
             job_object = self._get_job_2401(name)
         if _is_pipeline_child_job(job_object):
             raise PipelineChildJobError(job_id=job_object.id)
-        job_object.properties.is_archived = is_archived
 
+        # ---------------- TEMPORARY HOTFIX (archive/restore shortcut) ----------------
+        # The legacy _create_or_update_with_different_version_api PUT round-trip fails
+        # on AutoML, Command, Sweep, and Spark jobs for the same serializer/comparator
+        # reasons as create_or_update (see the shortcut in create_or_update above).
+        # RunHistory's "hidden" flag is the same server-side field surfaced by MFE as
+        # properties.isArchived (verified via probe_runhistory_hidden.py), so we can
+        # PATCH it directly - which is what ML Studio's portal does for the Archive
+        # button. Any failure falls through to the original path so behavior is
+        # preserved for Pipeline (which already works via _get_job_2401).
+        try:
+            from azure.ai.ml._restclient.runhistory.models import CreateRun
+
+            self._runs_operations._operation.add_or_modify_by_experiment_name(
+                subscription_id=self._operation_scope.subscription_id,
+                resource_group_name=self._operation_scope.resource_group_name,
+                workspace_name=self._workspace_name,
+                experiment_name=job_object.properties.experiment_name,
+                run_id=name,
+                body=CreateRun(hidden=is_archived),
+            )
+            return
+        except Exception:  # pylint: disable=broad-exception-caught
+            # Fall through to the original code path on any RunHistory error.
+            pass
+        # ------------------ END TEMPORARY HOTFIX --------------------------------
+
+        job_object.properties.is_archived = is_archived
         self._create_or_update_with_different_version_api(rest_job_resource=job_object)
 
     @distributed_trace
@@ -896,6 +983,71 @@ def restore(self, name: str) -> None:
 
         self._archive_or_restore(name=name, is_archived=False)
 
+    @distributed_trace
+    @monitor_with_telemetry_mixin(ops_logger, "Job.Update", ActivityType.PUBLICAPI)
+    def update(
+        self,
+        name: str,
+        *,
+        display_name: Optional[str] = None,
+        description: Optional[str] = None,
+        tags: Optional[Dict[str, str]] = None,
+        properties: Optional[Dict[str, str]] = None,
+    ) -> Job:
+        """Partially updates the display_name, description, tags or properties of an existing job.
+
+        This routes through the RunHistory PATCH endpoint that Azure ML Studio's portal uses
+        for inline edits, which works uniformly for all job types (AutoML, Command, Pipeline,
+        Sweep, Spark, Parallel) and merges tags/properties rather than replacing them.
+
+        :param name: The name of the job to update.
+        :type name: str
+        :keyword display_name: New display name for the job. If None, leaves the current value unchanged.
+        :paramtype display_name: Optional[str]
+        :keyword description: New description for the job. If None, leaves the current value unchanged.
+        :paramtype description: Optional[str]
+        :keyword tags: Tags to merge into the job's existing tags.
+        :paramtype tags: Optional[Dict[str, str]]
+        :keyword properties: Properties to merge into the job's existing properties.
+        :paramtype properties: Optional[Dict[str, str]]
+        :raises ~azure.ai.ml.exceptions.UserErrorException: Raised if no updatable field is supplied.
+        :raises ~azure.ai.ml.exceptions.PipelineChildJobError: Raised if the target job is a pipeline child job.
+        :return: The refreshed Job entity.
+        :rtype: ~azure.ai.ml.entities.Job
+        """
+        if display_name is None and description is None and tags is None and properties is None:
+            raise UserErrorException(
+                "jobs.update() requires at least one of "
+                "display_name, description, tags or properties to be specified."
+            )
+
+        job_object = self._get_job(name)
+        if job_object.properties.job_type == RestJobType.PIPELINE:
+            job_object = self._get_job_2401(name)
+        if _is_pipeline_child_job(job_object):
+            raise PipelineChildJobError(job_id=job_object.id)
+
+        from azure.ai.ml._restclient.runhistory.models import CreateRun
+
+        self._runs_operations._operation.add_or_modify_by_experiment_name(
+            subscription_id=self._operation_scope.subscription_id,
+            resource_group_name=self._operation_scope.resource_group_name,
+            workspace_name=self._workspace_name,
+            experiment_name=job_object.properties.experiment_name,
+            run_id=name,
+            body=CreateRun(
+                display_name=display_name,
+                description=description,
+                tags=tags,
+                properties=properties,
+            ),
+        )
+
+        refreshed = self._get_job(name)
+        if refreshed.properties.job_type == RestJobType.PIPELINE:
+            refreshed = self._get_job_2401(name)
+        return self._resolve_azureml_id(Job._from_rest_object(refreshed))
+
     @distributed_trace
     @monitor_with_activity(ops_logger, "Job.Stream", ActivityType.PUBLICAPI)
     def stream(self, name: str) -> None:

sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py

@@ -266,6 +266,168 @@ def test_restore(self, mock_method, mock_job_operation: JobOperations) -> None:
         mock_job_operation.service_client_01_2024_preview.jobs.get.assert_called_once()
         mock_job_operation._operation_2023_02_preview.create_or_update.assert_called_once()
 
+    # -------------- jobs.update() (new public API) --------------
+
+    def test_update_no_field_raises_user_error(self, mock_job_operation: JobOperations) -> None:
+        """update() with no updatable keyword must raise UserErrorException without any service calls."""
+        from azure.ai.ml.exceptions import UserErrorException
+
+        with pytest.raises(UserErrorException, match="at least one"):
+            mock_job_operation.update(name="random_name")
+
+    @patch.object(Job, "_from_rest_object")
+    @patch.object(JobOperations, "_resolve_azureml_id")
+    @patch.object(JobOperations, "_get_job")
+    def test_update_routes_through_runhistory_patch(
+        self, mock_get_job, mock_resolve, mock_from_rest, mock_job_operation: JobOperations
+    ) -> None:
+        """update() with fields must invoke add_or_modify_by_experiment_name exactly once with
+        the correct experiment_name / run_id and a CreateRun body carrying the supplied fields.
+        The returned entity must also be routed through _resolve_azureml_id so callers get the
+        same resolved view they'd get from jobs.get()."""
+        from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType
+
+        fake_job = Mock()
+        fake_job.properties.job_type = RestJobType.COMMAND
+        fake_job.properties.experiment_name = "exp1"
+        mock_get_job.return_value = fake_job
+        resolved_job = Command(component=None)
+        mock_from_rest.return_value = Command(component=None)
+        mock_resolve.return_value = resolved_job
+        # Force the property to return a fresh mock so add_or_modify_by_experiment_name is
+        # assertable (bypasses the lazy RunOperations construction that would fail on mocks).
+        mock_job_operation._runs_operations_client = Mock()
+
+        result = mock_job_operation.update(
+            name="random_name",
+            display_name="new dn",
+            description="new desc",
+            tags={"k": "v"},
+            properties={"pk": "pv"},
+        )
+
+        add_mod = mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name
+        add_mod.assert_called_once()
+        call_kwargs = add_mod.call_args.kwargs
+        assert call_kwargs["experiment_name"] == "exp1"
+        assert call_kwargs["run_id"] == "random_name"
+        body = call_kwargs["body"]
+        assert body.display_name == "new dn"
+        assert body.description == "new desc"
+        assert body.tags == {"k": "v"}
+        assert body.properties == {"pk": "pv"}
+        mock_resolve.assert_called_once()
+        assert result is resolved_job
+
+    @patch.object(Job, "_from_rest_object")
+    @patch.object(JobOperations, "_resolve_azureml_id")
+    @patch.object(JobOperations, "_get_job_2401")
+    @patch.object(JobOperations, "_get_job")
+    def test_update_pipeline_uses_2401_refetch(
+        self,
+        mock_get_job,
+        mock_get_job_2401,
+        mock_resolve,
+        mock_from_rest,
+        mock_job_operation: JobOperations,
+    ) -> None:
+        """PIPELINE jobs must be re-fetched via _get_job_2401 to obtain the non-projected view
+        before the RunHistory PATCH is issued. The refreshed entity is then resolved."""
+        from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType
+
+        pipeline_job = Mock()
+        pipeline_job.properties.job_type = RestJobType.PIPELINE
+        pipeline_job.properties.experiment_name = "exp1"
+        mock_get_job.return_value = pipeline_job
+        mock_get_job_2401.return_value = pipeline_job
+        mock_from_rest.return_value = Command(component=None)
+        mock_resolve.return_value = Command(component=None)
+        mock_job_operation._runs_operations_client = Mock()
+
+        mock_job_operation.update(name="random_name", display_name="new dn")
+
+        # _get_job_2401 is called at least once (initial resolve + refresh both hit it for
+        # PIPELINE jobs); the RunHistory PATCH must still be issued.
+        assert mock_get_job_2401.call_count >= 1
+        mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name.assert_called_once()
+        mock_resolve.assert_called_once()
+
+    @patch.object(JobOperations, "_get_job_2401")
+    @patch.object(JobOperations, "_get_job")
+    def test_update_pipeline_child_raises(
+        self, mock_get_job, mock_get_job_2401, mock_job_operation: JobOperations
+    ) -> None:
+        """A pipeline child job (properties is None on the 2401 view) must raise
+        PipelineChildJobError and must NOT issue any RunHistory PATCH."""
+        from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType
+        from azure.ai.ml.exceptions import PipelineChildJobError
+
+        parent_view = Mock()
+        parent_view.properties.job_type = RestJobType.PIPELINE
+        mock_get_job.return_value = parent_view
+
+        child = Mock()
+        child.properties = None  # _is_pipeline_child_job -> True
+        child.id = "/fake/child/id"
+        mock_get_job_2401.return_value = child
+
+        mock_job_operation._runs_operations_client = Mock()
+
+        with pytest.raises(PipelineChildJobError):
+            mock_job_operation.update(name="child_run", description="x")
+
+        mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name.assert_not_called()
+
+    # -------------- create_or_update metadata-only shortcut --------------
+
+    @patch.object(Job, "_from_rest_object")
+    @patch.object(JobOperations, "_resolve_azureml_id")
+    @patch.object(JobOperations, "_get_job")
+    def test_create_or_update_metadata_shortcut_uses_runhistory_patch(
+        self,
+        mock_get_job,
+        mock_resolve,
+        mock_from_rest,
+        mock_job_operation: JobOperations,
+    ) -> None:
+        """For a Job that was previously fetched (has an ARM id) and is being resubmitted with
+        only metadata edits (no compute/experiment_name change), create_or_update must route
+        through the RunHistory PATCH shortcut and skip the legacy MFE PUT round-trip."""
+        from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType
+
+        fake_job = Mock()
+        fake_job.properties.job_type = RestJobType.COMMAND
+        fake_job.properties.experiment_name = "exp1"
+        mock_get_job.return_value = fake_job
+        mock_from_rest.return_value = Command(component=None)
+        mock_resolve.return_value = Command(component=None)
+        mock_job_operation._runs_operations_client = Mock()
+
+        input_job = Mock()
+        input_job.id = (
+            "/subscriptions/x/resourceGroups/y/providers/Microsoft.MachineLearningServices"
+            "/workspaces/z/jobs/random_name"
+        )
+        input_job.name = "random_name"
+        input_job.display_name = "edited dn"
+        input_job.description = "edited desc"
+        input_job.tags = {"k": "v"}
+        input_job.properties = {"pk": "pv"}
+        input_job.compute = None
+
+        mock_job_operation.create_or_update(job=input_job)
+
+        add_mod = mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name
+        add_mod.assert_called_once()
+        body = add_mod.call_args.kwargs["body"]
+        assert body.display_name == "edited dn"
+        assert body.description == "edited desc"
+        assert body.tags == {"k": "v"}
+        assert body.properties == {"pk": "pv"}
+        # Legacy MFE PUT paths must be untouched when the shortcut succeeds.
+        mock_job_operation.service_client_01_2024_preview.jobs.create_or_update.assert_not_called()
+        mock_job_operation._operation_2023_02_preview.create_or_update.assert_not_called()
+
     @pytest.mark.parametrize(
         "corrupt_job_data",
         [