feat: add training job observability prints for script mode

Author: joshuatowner

manual_tests/pr_description_eval.md

@@ -0,0 +1,59 @@
+# Replace eval Rich spinner with plain-text observability output
+
+Replaces the Rich Live spinner in the terminal `wait()` path with plain-text status prints that stream correctly to agents and piped processes.
+
+## Problem
+
+The eval `wait()` terminal path used a Rich Live spinner that:
+- Outputs ANSI escape sequences agents can't parse
+- Uses `transient=True` so output disappears after completion
+- Only shows "Current status: Executing" with no step details
+
+## Changes
+
+- Remove Rich Live spinner from terminal `wait()` path
+- Print pipeline name and execution ARN at start
+- Print step transitions with ✓/⋯, durations, and job ARNs on each poll
+- Print "Running... Xs" elapsed timer for in-progress steps
+- Print S3 output path on success
+- Print failed step details with log group and CloudWatch link on failure
+- All prints use `flush=True` for agent compatibility
+- Log group derived from job ARN type (training/processing/transform)
+
+Jupyter path is unchanged.
+
+## Example output
+
+```
+Evaluation started: eval-meta-5190c33c
+Pipeline: SagemakerEvaluation-LLMAJEvaluation-565d3249
+Execution ARN: arn:aws:sagemaker:us-west-2:099324990371:pipeline/.../execution/rstzamudixmz
+
+--------------------------------------
+
+Status Transitions:
+  ✓ CreateEvaluationAction: Succeeded (3.0s)
+  ⋯ EvaluateCustomInferenceModel: Executing (Running... 17s)
+    Job ARN: arn:aws:sagemaker:us-west-2:099324990371:training-job/CustomInference-rstzamudixmz-6ZP4YbszZc
+
+Status: Executing (Elapsed: 21.5s)
+```
+
+On failure:
+```
+Failed
+Failure reason: Step failure: One or multiple steps failed.
+
+Failed step: EvaluateCustomInferenceModel
+Failure reason: ClientError: No S3 objects found under S3 URL...
+Job ARN: arn:aws:sagemaker:us-west-2:099324990371:training-job/CustomInference-xyz
+Log group: /aws/sagemaker/TrainingJobs
+Log stream prefix: CustomInference-xyz
+CloudWatch Logs: https://us-west-2.console.aws.amazon.com/cloudwatch/...
+```
+
+## Testing
+
+- 382 existing unit tests pass (0 regressions)
+- 4 new tests covering start, step transitions, success, and failure output
+- Manually validated with real LLM-as-Judge eval jobs

manual_tests/pr_description_poll.md

@@ -0,0 +1,26 @@
+# Expose poll interval parameter on all trainers
+
+Adds a `poll` parameter to `train()` on SFTTrainer, DPOTrainer, RLVRTrainer, and RLAIFTrainer, allowing callers to control how frequently the SDK polls for training job status updates.
+
+## Problem
+
+The `trainer_wait.wait()` function already accepts a `poll` parameter (default 5s), but there was no way to configure it from the trainer API. Users running in agent/script contexts want longer intervals (e.g., 30s) to reduce output noise and API calls.
+
+## Changes
+
+- Add `poll: int = 5` parameter to `train()` on all 4 trainers
+- Pass through to `trainer_wait.wait()` via `wait_kwargs`
+- Add docstring for the new parameter
+
+```python
+# Before: no way to control poll interval
+trainer.train(wait=True)
+
+# After: configurable
+trainer.train(wait=True, poll=30)
+```
+
+## Testing
+
+- 93 existing trainer unit tests pass (assertions updated to include `poll=5`)
+- Manually validated with SFT and RLAIF jobs showing correct 10s intervals

manual_tests/pr_description_training.md

@@ -0,0 +1,38 @@
+# Add training job observability prints for script mode
+
+Adds structured output at job start and on failure so users and agents always know where to find logs without scrolling or guessing.
+
+## Problem
+
+The terminal wait path printed only `TrainingJob Name:` at start and raised an exception on failure with no debug links. Users had to manually construct CloudWatch URLs or find the job in the console.
+
+## Changes
+
+- Print job name, log group, and log stream prefix at start
+- Print failure reason, log group, stream prefix, and CloudWatch Logs URL on failure
+- Add `flush=True` to all prints for agent compatibility
+- Preserve existing behavior: MLflow link on success, loss metrics by epoch
+
+## Example output
+
+At start:
+```
+Training job started: meta-textgeneration-llama-3-1-8b-instruct-sft-20260514
+Log group: /aws/sagemaker/TrainingJobs
+Log stream prefix: meta-textgeneration-llama-3-1-8b-instruct-sft-20260514
+```
+
+On failure:
+```
+Failure reason: ClientError: Unable to download training image...
+
+Log group: /aws/sagemaker/TrainingJobs
+Log stream prefix: meta-textgeneration-llama-3-1-8b-instruct-sft-20260514
+CloudWatch Logs: https://us-west-2.console.aws.amazon.com/cloudwatch/home?region=us-west-2#logsV2:log-groups/log-group/$252Faws$252Fsagemaker$252FTrainingJobs$3FlogStreamNameFilter$3Dmeta-textgeneration-llama-3-1-8b-instruct-sft-20260514
+```
+
+## Testing
+
+- 277 existing unit tests pass (0 regressions)
+- 4 new tests covering start prints, failure debug info, and CloudWatch URL generation
+- Manually validated with SFT and DPO jobs

sagemaker-train/src/sagemaker/train/common_utils/trainer_wait.py

@@ -419,7 +419,9 @@ def get_cached_mlflow_url():
                         raise TimeoutExceededError(resource_type="TrainingJob", status=status)
 
         else:
-            print(f"\nTrainingJob Name: {training_job.training_job_name}")
+            print(f"\nTraining job started: {training_job.training_job_name}", flush=True)
+            print(f"Log group: /aws/sagemaker/TrainingJobs", flush=True)
+            print(f"Log stream prefix: {training_job.training_job_name}", flush=True)
             iteration = 0
             while True:
                 iteration += 1
@@ -432,8 +434,8 @@ def get_cached_mlflow_url():
 
                 # Show transitions with checkmarks
                 if training_job.secondary_status_transitions:
-                    print("\n--------------------------------------\n")
-                    print("Status Transitions:")
+                    print("\n--------------------------------------\n", flush=True)
+                    print("Status Transitions:", flush=True)
                     for trans in training_job.secondary_status_transitions:
                         duration, check = _calculate_transition_duration(trans)
 
@@ -452,28 +454,39 @@ def get_cached_mlflow_url():
                             if progress_pct is not None:
                                 step_msg += f" - {progress_pct:.1f}%{progress_text.replace(chr(10), ', ')}"
 
-                        print(step_msg)
-                print(f"\nStatus: {status} - {secondary_status} (Elapsed: {elapsed:.1f}s)")
+                        print(step_msg, flush=True)
+                print(f"\nStatus: {status} - {secondary_status} (Elapsed: {elapsed:.1f}s)", flush=True)
 
                 if status in ["Completed", "Failed", "Stopped"]:
                     if status == "Completed":
                         if mlflow_url:
-                            print(f"\n✓ Training completed! View metrics in MLflow: {mlflow_url}")
+                            print(f"\n✓ Training completed! View metrics in MLflow: {mlflow_url}", flush=True)
                         try:
                             steps_per_epoch = training_job.progress_info.total_step_count_per_epoch
                             loss_metrics_by_epoch = metrics_util._get_loss_metrics_by_epoch(run_name=mlflow_run_name,
                                                                                            steps_per_epoch=steps_per_epoch)
                             if loss_metrics_by_epoch:
-                                print("\n------------ Loss Metrics by Epoch ------------")
+                                print("\n------------ Loss Metrics by Epoch ------------", flush=True)
                                 for epoch, metrics in list(loss_metrics_by_epoch.items())[:-1]:
-                                    print(f"Epoch {epoch}: {metrics}")
-                                print("----------------------------------------------")
+                                    print(f"Epoch {epoch}: {metrics}", flush=True)
+                                print("----------------------------------------------", flush=True)
                         except Exception:
                             pass
+                    if status == "Failed":
+                        failure_reason = training_job.failure_reason
+                        if failure_reason and not _is_unassigned_attribute(failure_reason):
+                            print(f"\nFailure reason: {failure_reason}", flush=True)
+                        print(f"\nLog group: /aws/sagemaker/TrainingJobs", flush=True)
+                        print(f"Log stream prefix: {training_job.training_job_name}", flush=True)
+                        from sagemaker.train.common_utils.metrics_visualizer import get_cloudwatch_logs_url
+                        cw_url = get_cloudwatch_logs_url(training_job.training_job_arn)
+                        if cw_url:
+                            print(f"CloudWatch Logs: {cw_url}", flush=True)
+                        raise FailedStatusError(resource_type="TrainingJob", status=status, reason=failure_reason)
                     return
 
                 failure_reason = training_job.failure_reason
-                if status == "Failed" or (failure_reason and not _is_unassigned_attribute(failure_reason)):
+                if failure_reason and not _is_unassigned_attribute(failure_reason):
                     raise FailedStatusError(resource_type="TrainingJob", status=status, reason=failure_reason)
 
                 if timeout and elapsed >= timeout:

sagemaker-train/tests/unit/train/common_utils/test_trainer_wait_observability.py

@@ -0,0 +1,78 @@
+"""Tests for training job observability prints in script/terminal mode."""
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from sagemaker.train.common_utils.trainer_wait import wait, _is_unassigned_attribute
+
+
+class MockUnassigned:
+    pass
+
+
+class MockTrainingJob:
+    def __init__(self, status="Completed", failure_reason=None):
+        self.training_job_name = "test-sft-job-2026"
+        self.training_job_arn = "arn:aws:sagemaker:us-west-2:123456789:training-job/test-sft-job-2026"
+        self.training_job_status = status
+        self.secondary_status = "Training"
+        self.secondary_status_transitions = []
+        self.progress_info = MockUnassigned()
+        self.failure_reason = failure_reason
+        self.mlflow_config = MockUnassigned()
+        self._call_count = 0
+
+    def refresh(self):
+        self._call_count += 1
+
+
+class TestTrainingObservabilityAtStart:
+    """Test that job info is printed at start in terminal mode."""
+
+    @patch("sagemaker.train.common_utils.trainer_wait._is_jupyter_environment", return_value=False)
+    @patch("sagemaker.train.common_utils.trainer_wait._setup_mlflow_integration", return_value=(None, None, None))
+    def test_prints_job_info_at_start(self, mock_mlflow, mock_jupyter, capsys):
+        job = MockTrainingJob(status="Completed")
+        wait(job, poll=0, timeout=1)
+        captured = capsys.readouterr()
+        assert "Training job started: test-sft-job-2026" in captured.out
+        assert "Log group: /aws/sagemaker/TrainingJobs" in captured.out
+        assert "Log stream prefix: test-sft-job-2026" in captured.out
+
+
+class TestTrainingObservabilityOnFailure:
+    """Test that debug info is printed on failure."""
+
+    @patch("sagemaker.train.common_utils.trainer_wait._is_jupyter_environment", return_value=False)
+    @patch("sagemaker.train.common_utils.trainer_wait._setup_mlflow_integration", return_value=(None, None, None))
+    def test_prints_debug_info_on_failure(self, mock_mlflow, mock_jupyter, capsys):
+        job = MockTrainingJob(status="Failed", failure_reason="OOM error")
+        with pytest.raises(Exception):
+            wait(job, poll=0, timeout=1)
+        captured = capsys.readouterr()
+        assert "Failure reason: OOM error" in captured.out
+        assert "Log group: /aws/sagemaker/TrainingJobs" in captured.out
+        assert "Log stream prefix: test-sft-job-2026" in captured.out
+        assert "CloudWatch Logs:" in captured.out
+
+    @patch("sagemaker.train.common_utils.trainer_wait._is_jupyter_environment", return_value=False)
+    @patch("sagemaker.train.common_utils.trainer_wait._setup_mlflow_integration", return_value=(None, None, None))
+    def test_prints_cloudwatch_url_on_failure(self, mock_mlflow, mock_jupyter, capsys):
+        job = MockTrainingJob(status="Failed", failure_reason="ClientError")
+        with pytest.raises(Exception):
+            wait(job, poll=0, timeout=1)
+        captured = capsys.readouterr()
+        assert "us-west-2.console.aws.amazon.com/cloudwatch" in captured.out
+
+
+class TestTrainingObservabilityOnSuccess:
+    """Test that MLflow link is printed on success (existing behavior preserved)."""
+
+    @patch("sagemaker.train.common_utils.trainer_wait._is_jupyter_environment", return_value=False)
+    @patch("sagemaker.train.common_utils.trainer_wait._setup_mlflow_integration", return_value=("https://mlflow.example.com", None, None))
+    def test_prints_mlflow_on_success(self, mock_mlflow, mock_jupyter, capsys):
+        job = MockTrainingJob(status="Completed")
+        wait(job, poll=0, timeout=1)
+        captured = capsys.readouterr()
+        assert "mlflow.example.com" in captured.out