fix(sagemaker-core): drop tags before rebuilding TransformJob in tran…

sagemaker-core/src/sagemaker/core/transformer.py

@@ -409,6 +409,8 @@ def submit(request):
         from sagemaker.core.utils.code_injection.codec import transform as transform_util
 
         transformed = transform_util(serialized_request, "CreateTransformJobRequest")
+        # Remove tags from transformed dict as TransformJob resource doesn't accept it
+        transformed.pop("tags", None)
         self.latest_transform_job = TransformJob(**transformed)
 
         if wait:

sagemaker-core/tests/unit/test_transformer.py

@@ -639,3 +639,48 @@ def test_load_config(self, mock_session):
         assert "resource_config" in config
         assert config["output_config"]["s3_output_path"] == "s3://bucket/output"
         assert config["resource_config"]["instance_count"] == 2
+
+
+class TestTransformerTransformWithTags:
+    """Regression tests for tags handling in Transformer.transform() (issue #6090)."""
+
+    def test_transform_removes_tags_before_building_transform_job(self, mock_session):
+        """Tags must be dropped from the transformed dict before TransformJob(**transformed).
+
+        TransformJob resource model has no tags field and sets extra="forbid", so leaving
+        tags in the dict raises a pydantic ValidationError after the job is already submitted.
+        """
+        transformer = Transformer(
+            model_name="test-model",
+            instance_count=1,
+            instance_type="ml.m5.xlarge",
+            output_path="s3://bucket/output",
+            tags=[{"Key": "team", "Value": "ml"}],
+            sagemaker_session=mock_session,
+        )
+
+        # Session is not a PipelineSession, so intercept just invokes submit(request).
+        mock_session._intercept_create_request = Mock(
+            side_effect=lambda request, submit, *args, **kwargs: submit(request)
+        )
+
+        with patch(
+            "sagemaker.core.utils.code_injection.codec.transform",
+            return_value={
+                "transform_job_name": "test-job",
+                "tags": [{"Key": "team", "Value": "ml"}],
+            },
+        ):
+            with patch("sagemaker.core.transformer.TransformJob") as mock_transform_job_class:
+                transformer.transform(
+                    data="s3://bucket/input",
+                    content_type="text/csv",
+                    job_name="test-job",
+                    wait=False,
+                )
+
+        # TransformJob must be constructed without the tags key.
+        mock_transform_job_class.assert_called_once()
+        call_kwargs = mock_transform_job_class.call_args[1]
+        assert "tags" not in call_kwargs
+        assert call_kwargs["transform_job_name"] == "test-job"