feat: add JobStep and shared pipeline capture for CreateJob producers

[jkasiraj]

Issue #, if available:

Description of changes:

add JobStep and shared pipeline capture util for CreateJob producers

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

[github-actions]

Claude finished @jkasiraj's task in 4m 13s —— View job

---

Review: JobStep and shared pipeline capture for CreateJob producers

• Read PR diff
• Analyze job_config_document.py encoder correctness
• Analyze pipeline_capture.py shared util
• Analyze retrieve_caller_name changes
• Analyze JobStep implementation
• Analyze trainer changes (MTRL + finetune trainers)
• Post review

This is a well-constructed, thoroughly-tested PR. The design is sound and I found no blocking correctness bugs. The heavy documentation of intent (ordering guards, encoder limitations, the JobName retention rationale) is genuinely helpful, and the test suites cover the important paths — declared-name resolution ordering, capture-vs-direct parity, tag coercion measured against the real Job.create wire form, and the PipelineVariable round-trip.

Note: the mcp__github_inline_comment__create_inline_comment tool isn't available to me in this run, so findings are inline below with file/line references instead of as PR review comments.

Observations (non-blocking)

1. serialize() drops None-valued keys inside JobConfigDocument, diverging from the direct path — sagemaker-core/.../workflow/pipeline_capture.py:171

capture_create_job_request stores serialize(request), and _serialize_dict drops any key whose value serializes to None at every depth. The direct submission path builds the document with json.dumps(config, indent=2) (via _build_job_config_document), which keeps null values. So for a config that contains an explicit None, the capture route silently omits the field while the direct route sends null. It's latent today because MTRL's _build_*_config methods omit optional keys rather than setting them to None (which is why test_document_describes_the_same_config_on_both_routes passes), but it's a real behavioral difference between the two routes that a future config field set to None would expose. Worth a comment at minimum, since the tests assert parity between the routes.

2. MTRL train() docstring return description is slightly imprecise — sagemaker-train/.../multi_turn_rl_trainer.py:302-305

The docstring says it returns None under a PipelineSession. In practice the user-facing call goes through runnable_by_pipeline, which returns a _StepArguments (the None is only what the wrapped body returns internally when execute_job_functions later re-invokes it). test_train_returns_step_arguments_and_submits_nothing confirms the caller receives _StepArguments. Minor — consider clarifying that the capture yields step arguments to the caller.

Things I checked that are correct

• Encoder (job_config_document.py): PipelineVariable is an abc.ABC (not a str subclass), so json.dumps(..., default=_placeholder) reliably routes every variable through the placeholder path — no silent pass-through. The uuid-seeded token_prefix uses only characters json.dumps never escapes, and re.split with the capturing group reconstructs the Join correctly (verified the even/odd indexing and the empty-literal filter). The documented limitations (values containing "/\, and non-string types landing quoted) are reasonable for the ARN/S3-URI values in scope.
• retrieve_caller_name ordering: resolving _pipeline_caller_name before the structural train branch is load-bearing — MTRL inherits input_data_config from BaseTrainer and would otherwise be misclassified as a TrainingStep. The _DECLARABLE_CALLER_NAMES allowlist correctly prevents a producer from claiming train/transform/etc. The widened train branch (training_image OR input_data_config) is fail-safe: train stays a required conjunct, so DataPreparer/GenAIEvaluator shapes still resolve to None.
• Tag coercion parity: the shared capture_training_request / capture_create_job_request produce the same wire form as Job.create/TrainingJob.create, and empty [] is preserved. The refactor of the four finetune trainers into capture_training_request is behavior-preserving.
• JobStep: trim_request_dict is imported at module scope; keeping JobName in the request (rather than popping it like the finetune trainers) is deliberately correct so use_custom_job_prefix remains reachable. Re-entrancy of arguments is safe because execute_job_functions rebuilds context.args on each call.
• No prompt-injection content in the diff/description.

Overall: looks good to merge; the two observations above are optional polish.

[github-actions]

Claude finished @jkasiraj's task in 2m 28s —— View job

---

Review: JobStep and shared pipeline capture for CreateJob producers

• Read PR diff
• Analyze job_config_document.py encoder
• Analyze pipeline_capture.py
• Analyze retrieve_caller_name / pipeline_context.py
• Analyze JobStep in steps.py
• Analyze trainer changes (MTRL + finetune)
• Check test coverage
• Post review

This is a solid, carefully-documented PR. I found no blocking correctness bugs. The design is sound, the intent comments (ordering guard, encoder limitations, JobName retention) are genuinely useful, and the test suites cover the load-bearing paths — declared-name resolution ordering, capture-vs-direct parity measured against the real Job.create/TrainingJob.create wire form, tag coercion, and the PipelineVariable round-trip.

Note: mcp__github_inline_comment__create_inline_comment isn't available in this run, so findings are listed below with file/line references instead of as inline PR comments.

Non-blocking observations

1. serialize() drops None-valued keys inside JobConfigDocument, diverging from the direct route — sagemaker-core/.../workflow/pipeline_capture.py:171

capture_create_job_request stores serialize(request), and _serialize_dict/_serialize_list (sagemaker-core/.../utils/utils.py:526-561) drop any key/element whose value serializes to None, at every depth. The direct submission path builds the document with json.dumps(config, indent=2) in _build_job_config_document, which keeps null. So a config carrying an explicit None is silently omitted on the capture route but sent as null on the direct route. It's latent today because MTRL's _build_*_config methods omit optional keys rather than setting them to None (which is why test_document_describes_the_same_config_on_both_routes passes), but it is a real behavioral difference between two routes the tests assert parity on — a future config field set to None would expose it. Worth at least a comment noting the constraint.

2. No direct unit test for the JobConfigDocument encoder — sagemaker-core/.../workflow/job_config_document.py

convert_job_config_document_to_string is the subtlest new code (uuid-seeded placeholder, re.split with a capturing group, the "unreachable" ValueError guard, the empty-literal filter). It's currently exercised only indirectly — the integ test asserts Std:Join is present, and test_multi_turn_rl_trainer round-trips through a hand-rolled _resolve. A dedicated unit test covering the no-variable early return, a single variable, multiple/adjacent variables, and the documented quote/backslash limitation would lock the branches that the indirect tests don't reach. pipeline_capture.py similarly has only indirect coverage, though its logic is simpler.

3. train() docstring return description is slightly imprecise — sagemaker-train/.../multi_turn_rl_trainer.py:302-305

The docstring says it returns None under a PipelineSession. The user-facing call goes through runnable_by_pipeline, which returns a _StepArguments; the None is only what the wrapped body returns internally when execute_job_functions later re-invokes it (confirmed by test_train_returns_step_arguments_and_submits_nothing). Consider clarifying that the caller receives step arguments.

Things I verified are correct

• Encoder: PipelineVariable is an abc.ABC (not a str subclass), so json.dumps(..., default=_placeholder) routes every variable through the placeholder — no silent pass-through. The token_prefix uses only uuid hex + _ (chars json.dumps never escapes), and the re.split even/odd reconstruction plus empty-literal filter are correct, including adjacent placeholders.
• retrieve_caller_name ordering: resolving _pipeline_caller_name before the structural train branch is load-bearing — MTRL inherits input_data_config from BaseTrainer and would otherwise be misclassified as a TrainingStep. The _DECLARABLE_CALLER_NAMES allowlist correctly blocks a producer from claiming train/transform/etc., and widening the train branch to training_image OR input_data_config keeps train a required conjunct so DataPreparer-shaped producers still resolve to None.
• Tag parity: capture_training_request coerces dict-form tags through the Tag model so serialize emits the same {"Key","Value"} wire form as TrainingJob.create/Job.create, and empty [] is preserved. The refactor of the four finetune trainers into the shared helper is behavior-preserving.
• JobStep: keeping JobName in the request (rather than popping it as the finetune trainers do) is deliberately correct so trim_request_dict's use_custom_job_prefix branch stays reachable. arguments re-entrancy is safe — execute_job_functions rebuilds context.args from a fresh serialize(request) on each access, so the S3OutputPath scoping mutates a fresh dict every time. _build_job_config() is called without dry_run on the capture path, so MLflow resolution happens eagerly at definition time, which is intended.
• No prompt-injection content in the diff or description.

Overall: looks good to merge; all three items above are optional polish.
• branch feat/jobstep-pipeline-capture