fix: keep job name in pipeline request for ModelTrainer and HyperparameterTuner

Both unconditionally popped training_job_name / hyper_parameter_tuning_job_name from
the request for a PipelineSession. The TrainingStep/TuningStep already strip it via
trim_request_dict (dropped by default, prefix preserved when
PipelineDefinitionConfig(use_custom_job_prefix=True)), so popping upstream left
use_custom_job_prefix nothing to preserve and made base_job_name have no effect in
pipelines. Stop popping; let trim_request_dict handle it.

Fixes #5776, #6299

Author: Mohamed Zeidan

sagemaker-train/src/sagemaker/train/model_trainer.py

@@ -814,8 +814,10 @@ def _create_training_job_args(
             training_request["model_package_config"] = self.model_package_config
 
         if boto3 or isinstance(self.sagemaker_session, PipelineSession):
-            if isinstance(self.sagemaker_session, PipelineSession):
-                training_request.pop("training_job_name", None)
+            # Keep training_job_name in the request. The TrainingStep strips it via
+            # trim_request_dict, which drops it by default but preserves the base_job_name
+            # prefix when PipelineDefinitionConfig(use_custom_job_prefix=True). Popping it here
+            # unconditionally left use_custom_job_prefix nothing to preserve (issue #5776, #6299).
             # Convert snake_case to PascalCase for AWS API
             pipeline_request = {to_pascal_case(k): v for k, v in training_request.items()}
             serialized_request = serialize(pipeline_request)

sagemaker-train/src/sagemaker/train/tuner.py

@@ -1290,8 +1290,10 @@ def _start_tuning_job(self, inputs):
             from sagemaker.core.utils.utils import serialize
             from sagemaker.core.apiutils._boto_functions import to_pascal_case
 
-            # Remove job name for pipeline as it's auto-generated at execution time
-            tuning_request.pop("hyper_parameter_tuning_job_name", None)
+            # Keep hyper_parameter_tuning_job_name in the request. The TuningStep strips it via
+            # trim_request_dict, which drops it by default but preserves the base_tuning_job_name
+            # prefix when PipelineDefinitionConfig(use_custom_job_prefix=True). Popping it here
+            # unconditionally left use_custom_job_prefix nothing to preserve (issue #6299, #5776).
             # Convert snake_case to PascalCase for AWS API
             pipeline_request = {to_pascal_case(k): v for k, v in tuning_request.items()}
             serialized_request = serialize(pipeline_request)

sagemaker-train/tests/unit/train/test_model_trainer.py

@@ -254,6 +254,38 @@ def test_model_trainer_param_validation(test_case, modules_session):
         assert trainer.base_job_name == DEFAULT_BASE_NAME
 
 
+def test_pipeline_session_request_keeps_training_job_name(modules_session):
+    """Regression for #5776 / #6299.
+
+    Under a PipelineSession the request must KEEP TrainingJobName. The TrainingStep strips it via
+    trim_request_dict (dropped by default, prefix preserved when use_custom_job_prefix=True);
+    popping it here left use_custom_job_prefix nothing to preserve.
+    """
+    from unittest.mock import Mock
+    from sagemaker.core.workflow.pipeline_context import PipelineSession
+
+    session = Mock(spec=PipelineSession)
+    session.default_bucket.return_value = DEFAULT_BUCKET
+    session.default_bucket_prefix = DEFAULT_BUCKET_PREFIX
+    session.boto_region_name = DEFAULT_REGION
+    session.sagemaker_config = {}
+
+    trainer = ModelTrainer(
+        training_image=DEFAULT_IMAGE,
+        role=DEFAULT_ROLE,
+        compute=DEFAULT_COMPUTE_CONFIG,
+        stopping_condition=DEFAULT_STOPPING_CONDITION,
+        output_data_config=DEFAULT_OUTPUT_DATA_CONFIG,
+        base_job_name="my-prefix",
+        sagemaker_session=session,
+    )
+
+    args = trainer._create_training_job_args(input_data_config=[])
+
+    assert "TrainingJobName" in args, "TrainingJobName must survive into the pipeline request"
+    assert args["TrainingJobName"].startswith("my-prefix")
+
+
 @patch("sagemaker.train.model_trainer.TrainingJob")
 def test_train_with_default_params(mock_training_job, model_trainer):
     model_trainer.train()

sagemaker-train/tests/unit/train/test_tuner.py

@@ -598,6 +598,34 @@ def test_build_training_job_definition_includes_spot_params(self):
             definition.stopping_condition.max_wait_time_in_seconds, int
         ), "Max wait time should be set"
 
+    def test_pipeline_session_request_keeps_tuning_job_name(self):
+        """Regression for #6299 / #5776.
+
+        Under a PipelineSession the tuning request must KEEP HyperParameterTuningJobName. The
+        TuningStep strips it via trim_request_dict (dropped by default, prefix preserved when
+        use_custom_job_prefix=True); popping it here left use_custom_job_prefix nothing to preserve.
+        """
+        from unittest.mock import Mock
+        from sagemaker.core.workflow.pipeline_context import PipelineSession
+
+        pipeline_session = Mock(spec=PipelineSession)
+        mock_trainer = _create_mock_model_trainer()
+        mock_trainer.sagemaker_session = pipeline_session
+
+        tuner = HyperparameterTuner(
+            model_trainer=mock_trainer,
+            objective_metric_name="accuracy",
+            hyperparameter_ranges=_create_single_hp_range(),
+        )
+        tuner._current_job_name = "my-prefix-2026-01-01-00-00-00-000"
+
+        tuner._start_tuning_job(inputs=None)
+
+        pipeline_session._intercept_create_request.assert_called_once()
+        serialized_request = pipeline_session._intercept_create_request.call_args.args[0]
+        assert "HyperParameterTuningJobName" in serialized_request
+        assert serialized_request["HyperParameterTuningJobName"].startswith("my-prefix")
+
     def test_build_training_job_definition_includes_environment_variables(self):
         """Test that _build_training_job_definition includes environment variables.