fix(train): add PipelineSession support to V3 trainers (SFT/DPO/RLAIF…

[nayan3107]

…/RLVR)

When a PipelineSession is passed as sagemaker_session, the serverless training path in SFTTrainer, DPOTrainer, RLAIFTrainer, and RLVRTrainer now intercepts the CreateTrainingJob request and returns step arguments instead of immediately launching a training job.

This enables V3 trainers to be used with SageMaker Pipelines TrainingStep, matching the existing behavior of ModelTrainer, Processor, Transformer, and HyperparameterTuner.

The fix follows the established SDK pattern: isinstance check for PipelineSession, call _intercept_create_request with the request args, and return session.context (the captured step arguments).

Issue #, if available:
Fixes: #6163

Description of changes:

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

Testing

Unit Tests (167 pass, 0 regressions):

• All existing SFT/DPO/RLAIF/RLVR trainer tests pass unchanged
• 4 new TestXxxTrainerPipelineSession classes added verifying:
  • TrainingJob.create() is NOT called when PipelineSession is used
  • session._intercept_create_request() IS called with correct create_args and func_name="train"
  • Return value is session.context (_JobStepArguments) — usable with TrainingStep

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

from sagemaker.core.workflow.pipeline_context import PipelineSession
from sagemaker.train.sft_trainer import SFTTrainer
from sagemaker.train.common import TrainingType

session = PipelineSession()
trainer = SFTTrainer(
    model='meta-textgeneration-llama-3-2-1b-instruct',
    training_type=TrainingType.LORA,
    training_dataset='s3://my-bucket/train.jsonl',
    model_package_group='my-group',
    sagemaker_session=session,
    accept_eula=True,
)
result = trainer.train()
# type(result) = _JobStepArguments
# result.caller_name = 'train'
# No CreateTrainingJob API call made. No billing.

Verified scenarios:

Scenario	Result
SFTTrainer + PipelineSession	✅ Returns step_args, no job launched
SFTTrainer + regular Session	✅ Launches job normally (existing behavior)
DPO/RLAIF/RLVR + PipelineSession	✅ Returns step_args (unit tested)
All 137 pre-existing trainer unit tests	✅ Pass (no regressions)