fix(train): complete PipelineSession support for V3 trainers (SFT/DPO…

[nayan3107]

…/RLAIF/RLVR)

Add @runnable_by_pipeline decorator and PascalCase request serialization so TrainingStep can consume step_args from V3 fine-tuning trainers.

Changes:

• Add @runnable_by_pipeline decorator on train() for all 4 trainers
• Build PascalCase serialized request (remove session/region, pop job name)
• Fix Tags to PascalCase (JumpStart returns lowercase key/value)
• Add source_code=None to BaseTrainer (required by get_code_hash)
• Add consumer tests (TrainingStep.arguments validation)
• Add regular session regression tests

Fixes: #6163

Issue #, if available:

Description of changes:

fix(train): complete PipelineSession support for V3 trainers

Follow-up to #6213. Completes PipelineSession integration so V3 fine-tuning
trainers (SFTTrainer, DPOTrainer, RLAIFTrainer, RLVRTrainer) work with
TrainingStep in SageMaker Pipelines.

Fixes #6163

Changes

• @runnable_by_pipeline decorator on train() for all 4 trainers — matches
  ModelTrainer/Processor/Transformer pattern. Decorator captures function reference

  • trainer instance so TrainingStep can re-execute during pipeline compilation.

• PascalCase request serialization inside train() body — removes non-serializable
  session/region, pops training_job_name (engine generates at runtime), converts
  snake_case keys to PascalCase via to_pascal_case() + serialize().

• Tags PascalCase fix — _get_jumpstart_tags() returns lowercase key/value
  dicts (not Tag objects), so _serialize_dict() doesn't auto-convert them. Manually
  converts to Key/Value before serialization.

• source_code = None in BaseTrainer — get_code_hash() in pipeline compilation
  accesses model_trainer.source_code directly. ModelTrainer defines this field;
  BaseTrainer didn't. Fine-tuning trainers use built-in containers with no customer code.

Testing

Unit tests (12 new, all existing pass):

• Producer tests: train() returns _StepArguments with func + trainer captured
• Consumer tests: TrainingStep(step_args=...).arguments produces valid PascalCase dict
  with no session/region leaked, Tags in correct Key/Value format
• Regular session tests: TrainingJob.create() still called normally (no regression)

Manual E2E (real AWS, us-west-2, account 243179453435):

session = PipelineSession()
trainer = SFTTrainer(model='meta-textgeneration-llama-3-2-1b-instruct', ...)
result = trainer.train()                    # _StepArguments (no job)
step = TrainingStep(name="SFT", step_args=result)
step.arguments                              # PascalCase dict ✅
pipeline = Pipeline(steps=[step], sagemaker_session=session)
pipeline.upsert(role_arn="...")             # CreatePipeline ✅
execution = pipeline.start()               # Execution started ✅

Pipeline created and execution rg29qr88isq5 started successfully.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.