fix: propagate user-supplied tags to created resources

sagemaker-train/src/sagemaker/train/base_trainer.py

@@ -1209,6 +1209,7 @@ def _yaml_safe_default(value):
             sagemaker_session=sagemaker_session,
             role=role,
             base_job_name=base_job_name,
+            tags=self.tags,
         )
 
         # Validate data paths exist before submission

sagemaker-train/src/sagemaker/train/dpo_trainer.py

@@ -346,6 +346,9 @@ def train(self,
         vpc_config = self.networking if self.networking else None
         tags = _get_jumpstart_tags(self._model_name, get_sagemaker_hub_name())
 
+        # Merge user-provided tags with the JumpStart tags
+        tags.extend(self.tags or [])
+
         # Build TrainingJob.create() arguments
         create_args = {
             "training_job_name": current_training_job_name,

sagemaker-train/src/sagemaker/train/evaluate/base_evaluator.py

@@ -138,6 +138,8 @@ class BaseEvaluator(BaseModel):
             3. Model package group name string (will fetch the object and extract ARN)
             Required when model is a JumpStart model ID. Optional when model is a ModelPackage
             ARN/object (will be inferred automatically).
+        tags (Optional[List[TagsDict]]): Tags applied to the evaluation pipeline when it is
+            created, which cascade to the pipeline's step jobs.
     """
 
     region: Optional[str] = None
@@ -153,6 +155,7 @@ class BaseEvaluator(BaseModel):
     networking: Optional[VpcConfig] = None
     kms_key_id: Optional[str] = None
     model_package_group: Optional[Union[str, ModelPackageGroup]] = None
+    tags: Optional[List[TagsDict]] = None
     compute: Optional[Union[Compute, HyperPodCompute]] = None
     training_image: Optional[str] = None
     recipe: Optional[str] = None
@@ -987,6 +990,9 @@ def _start_execution(
         if self._is_jumpstart_model:
             from sagemaker.core.jumpstart.utils import add_jumpstart_model_info_tags
             tags = add_jumpstart_model_info_tags(tags, self.model, "*")
+
+        # Merge user-provided tags
+        tags.extend(self.tags or [])
 
         execution = EvaluationPipelineExecution.start(
             eval_type=eval_type,

sagemaker-train/src/sagemaker/train/evaluate/multi_turn_rl_evaluator.py

@@ -44,6 +44,35 @@
 _MAX_STOPPING_CONDITION_SECONDS = 72 * 60 * 60
 
 
+def _tags_with_capitalized_keys(tags: Optional[List[Dict[str, str]]]) -> List[Dict[str, str]]:
+    """Convert tags to the capitalized ``Key``/``Value`` form this evaluator's paths require.
+
+    Both the raw boto3 ``CreatePipeline`` call and the ``Tags`` block rendered into the
+    pipeline definition use the API's capitalized form, whereas the other evaluators hand
+    tags to the pydantic-validated ``Pipeline.create``, which takes the lowercase form.
+    Either form is accepted here so that the inherited ``tags`` field behaves the same way
+    across evaluator subclasses.
+
+    Args:
+        tags: Tags using either ``key``/``value`` or ``Key``/``Value``, or None.
+
+    Returns:
+        The tags in capitalized form; empty when none were supplied. Entries missing a key
+        or value are skipped, since SageMaker rejects them.
+    """
+    normalized = []
+    for tag in tags or []:
+        if isinstance(tag, dict):
+            key = tag.get("Key", tag.get("key"))
+            value = tag.get("Value", tag.get("value"))
+        else:
+            key = getattr(tag, "key", None)
+            value = getattr(tag, "value", None)
+        if key is not None and value is not None:
+            normalized.append({"Key": key, "Value": value})
+    return normalized
+
+
 class MultiTurnRLEvaluator(BaseEvaluator):
     """Evaluate a multi-turn RL agent model against a held-out prompt dataset.
 
@@ -532,7 +561,7 @@ def _build_job_config_doc(include_mpc: bool, mlflow_run_name: str) -> str:
             "vpc_config": bool(networking),
             "vpc_security_group_ids": vpc_security_group_ids,
             "vpc_subnets": vpc_subnets,
-            "tags": self.tags,
+            "tags": _tags_with_capitalized_keys(self.tags) or None,
             # Pre-stringified JobConfigDocument for the templates.
             "job_config_document_str": job_config_doc_str,
             "job_config_document_ft_str": job_config_doc_ft_str,
@@ -699,6 +728,11 @@ def _start_mtrl_execution(self, pipeline_definition, name, role_arn, region):
         pipeline_prefix = _get_pipeline_name_prefix(EvalType.MTRL)
         pipeline_name = pipeline_prefix
 
+        # Customer tags are merged into the pipeline tags. This path uses raw boto3, which
+        # requires the API's capitalized Key/Value form.
+        pipeline_tags = [{"Key": _TAG_SAGEMAKER_MODEL_EVALUATION, "Value": "true"}]
+        pipeline_tags.extend(_tags_with_capitalized_keys(self.tags))
+
         # Search for existing MTRL pipeline
         existing_pipeline_name = None
         try:
@@ -725,7 +759,7 @@ def _start_mtrl_execution(self, pipeline_definition, name, role_arn, region):
                 PipelineDisplayName=pipeline_name,
                 PipelineDescription="MTRL evaluation pipeline",
                 ClientRequestToken=str(uuid.uuid4()),
-                Tags=[{"Key": _TAG_SAGEMAKER_MODEL_EVALUATION, "Value": "true"}],
+                Tags=pipeline_tags,
             )
             _logger.info(f"Created pipeline: {pipeline_name}")
 

sagemaker-train/src/sagemaker/train/multi_turn_rl_trainer.py

@@ -321,13 +321,17 @@ def train(
 
         tags = _get_jumpstart_tags(self._model_name, get_sagemaker_hub_name())
 
+        # Merge user-provided tags with the JumpStart tags
+        tags.extend(self.tags or [])
+
         try:
             job = Job.create(
                 job_name=current_job_name,
                 job_category=JOB_CATEGORY,
                 role_arn=role,
                 job_config_schema_version=JOB_CONFIG_SCHEMA_VERSION,
                 job_config_document=job_config_doc,
+                tags=tags,
                 session=sagemaker_session.boto_session,
                 region=sagemaker_session.boto_session.region_name,
             )

sagemaker-train/src/sagemaker/train/rlaif_trainer.py

@@ -310,6 +310,9 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validati
         vpc_config = self.networking if self.networking else None
         tags = _get_jumpstart_tags(self._model_name, get_sagemaker_hub_name())
 
+        # Merge user-provided tags with the JumpStart tags
+        tags.extend(self.tags or [])
+
         # Build TrainingJob.create() arguments
         create_args = {
             "training_job_name": current_training_job_name,

sagemaker-train/src/sagemaker/train/rlvr_trainer.py

@@ -520,6 +520,9 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None,
         vpc_config = self.networking if self.networking else None
         tags = _get_jumpstart_tags(self._model_name, get_sagemaker_hub_name())
 
+        # Merge user-provided tags with the JumpStart tags
+        tags.extend(self.tags or [])
+
         # Build TrainingJob.create() arguments
         create_args = {
             "training_job_name": current_training_job_name,

sagemaker-train/src/sagemaker/train/sft_trainer.py

@@ -414,6 +414,9 @@ def train(self, training_dataset: Optional[Union[str, DataSet]] = None, validati
         vpc_config = self.networking if self.networking else None
         tags = _get_jumpstart_tags(self._model_name, get_sagemaker_hub_name())
 
+        # Merge user-provided tags with the JumpStart tags
+        tags.extend(self.tags or [])
+
         # Build TrainingJob.create() arguments
         create_args = {
             "training_job_name": current_training_job_name,

sagemaker-train/tests/unit/train/evaluate/test_base_evaluator.py

@@ -14,7 +14,7 @@
 from __future__ import absolute_import
 
 import pytest
-from unittest.mock import patch, MagicMock, Mock
+from unittest.mock import patch, MagicMock, Mock, PropertyMock
 from pydantic import ValidationError
 
 from sagemaker.core.shapes import VpcConfig
@@ -24,6 +24,7 @@
 from sagemaker.train.base_trainer import BaseTrainer
 
 from sagemaker.train.evaluate.base_evaluator import BaseEvaluator
+from sagemaker.train.evaluate.constants import EvalType
 
 
 # Test constants
@@ -1840,3 +1841,57 @@ def test_fields_without_spec_type_not_coerced(self):
             {"x": {"default": "5"}}, semantic_values={"x": "5"}
         )
         assert value_map["x"] == "5"
+
+
+class TestStartExecutionTags:
+    """Tests for merging the ``tags`` field into the evaluation pipeline tags."""
+
+    @patch("sagemaker.train.common_utils.model_resolution._resolve_base_model")
+    def _make_evaluator(self, mock_resolve, mock_session, mock_model_info, tags=None):
+        mock_resolve.return_value = mock_model_info
+        return BaseEvaluator(
+            model=DEFAULT_MODEL,
+            s3_output_path=DEFAULT_S3_OUTPUT,
+            mlflow_resource_arn=DEFAULT_MLFLOW_ARN,
+            model_package_group=DEFAULT_MODEL_PACKAGE_GROUP_ARN,
+            sagemaker_session=mock_session,
+            tags=tags,
+        )
+
+    def _start(self, evaluator):
+        """Run _start_execution with the pipeline submit mocked out."""
+        with (
+            patch(
+                "sagemaker.train.evaluate.execution.EvaluationPipelineExecution"
+            ) as mock_execution,
+            patch.object(
+                BaseEvaluator, "_is_jumpstart_model", new_callable=PropertyMock
+            ) as mock_is_js,
+        ):
+            mock_is_js.return_value = False
+            evaluator._start_execution(
+                eval_type=EvalType.BENCHMARK,
+                name="test-eval",
+                pipeline_definition='{"Steps": []}',
+                role_arn=DEFAULT_ROLE_ARN,
+                region=DEFAULT_REGION,
+            )
+            return mock_execution.start.call_args.kwargs["tags"]
+
+    def test_user_tags_reach_the_pipeline(self, mock_session, mock_model_info):
+        """Tags set on the evaluator must be passed to the pipeline execution."""
+        evaluator = self._make_evaluator(
+            mock_session=mock_session,
+            mock_model_info=mock_model_info,
+            tags=[{"key": "sagemaker:project-id", "value": "p-12345"}],
+        )
+
+        assert self._start(evaluator) == [{"key": "sagemaker:project-id", "value": "p-12345"}]
+
+    def test_no_tags_yields_empty_list(self, mock_session, mock_model_info):
+        """Omitting tags must not fail, and must not invent any tags."""
+        evaluator = self._make_evaluator(
+            mock_session=mock_session, mock_model_info=mock_model_info, tags=None
+        )
+
+        assert self._start(evaluator) == []

sagemaker-train/tests/unit/train/evaluate/test_execution.py

@@ -129,6 +129,51 @@ def test_create_pipeline_success(self, mock_pipeline_class, mock_get_name, mock_
         )
         assert result == mock_pipeline
 
+    @patch("sagemaker.train.evaluate.execution._get_pipeline_name")
+    @patch("sagemaker.train.evaluate.execution.Pipeline")
+    def test_create_pipeline_propagates_user_tags(
+        self, mock_pipeline_class, mock_get_name, mock_session
+    ):
+        """User tags must reach Pipeline.create alongside the evaluation discovery tag."""
+        mock_get_name.return_value = DEFAULT_PIPELINE_NAME
+        mock_pipeline_class.create.return_value = MagicMock()
+
+        _create_evaluation_pipeline(
+            eval_type=EvalType.BENCHMARK,
+            role_arn=DEFAULT_ROLE,
+            pipeline_definition=DEFAULT_PIPELINE_DEFINITION,
+            session=mock_session,
+            region=DEFAULT_REGION,
+            tags=[{"key": "sagemaker:project-id", "value": "p-12345"}],
+        )
+
+        created_tags = mock_pipeline_class.create.call_args.kwargs["tags"]
+        pairs = [(t.key, t.value) for t in created_tags]
+        assert ("sagemaker:project-id", "p-12345") in pairs
+        # The evaluation discovery tag must still be present.
+        assert any(key == "SagemakerModelEvaluation" for key, _ in pairs)
+
+    @patch("sagemaker.train.evaluate.execution._get_pipeline_name")
+    @patch("sagemaker.train.evaluate.execution.Pipeline")
+    def test_create_pipeline_accepts_capitalized_user_tags(
+        self, mock_pipeline_class, mock_get_name, mock_session
+    ):
+        """Capitalized user tags must also be converted into Tag objects."""
+        mock_get_name.return_value = DEFAULT_PIPELINE_NAME
+        mock_pipeline_class.create.return_value = MagicMock()
+
+        _create_evaluation_pipeline(
+            eval_type=EvalType.BENCHMARK,
+            role_arn=DEFAULT_ROLE,
+            pipeline_definition=DEFAULT_PIPELINE_DEFINITION,
+            session=mock_session,
+            region=DEFAULT_REGION,
+            tags=[{"Key": "sagemaker:project-id", "Value": "p-12345"}],
+        )
+
+        pairs = [(t.key, t.value) for t in mock_pipeline_class.create.call_args.kwargs["tags"]]
+        assert ("sagemaker:project-id", "p-12345") in pairs
+
     @patch("sagemaker.train.evaluate.execution.Pipeline")
     def test_create_pipeline_waits_for_status(self, mock_pipeline_class, mock_session):
         """Test that pipeline waits for active status."""