Master nova follow ups

docs/model_customization/deploy_sagemaker_endpoint.rst

@@ -75,3 +75,65 @@ production deployments.
    model_builder = ModelBuilder(model=model_package)
    model = model_builder.build(model_name="my-registered-model")
    endpoint = model_builder.deploy(endpoint_name="my-endpoint")
+
+
+Reuse Deployed Resources
+--------------------------
+
+Pass ``reuse_resources=True`` to ``build()`` and ``deploy()`` to avoid creating duplicate
+endpoints when deploying the same model source multiple times.
+
+On first deploy, the SDK tags the endpoint with ``sagemaker.amazonaws.com/model-source``
+derived from the model's stable source identifier (model package ARN, escrow URI, S3 path,
+or JumpStart model ID). On subsequent deploys with ``reuse_resources=True``, the SDK discovers
+the existing endpoint by that tag and returns it instead of creating a new one.
+
+.. list-table::
+   :header-rows: 1
+
+   * - Model input style
+     - Source ID used for tag
+   * - ``ModelPackage``
+     - Model package ARN
+   * - ``BaseTrainer`` (with completed training job)
+     - Model package ARN if available, otherwise escrow resolution
+   * - ``TrainingJob``
+     - Via model package ARN or escrow resolution
+   * - Raw S3 URI string
+     - The S3 path itself
+   * - JumpStart model ID string
+     - The model ID string
+
+.. code-block:: python
+
+   from sagemaker.serve import ModelBuilder
+
+   builder = ModelBuilder(
+       model=my_trainer,  # or ModelPackage, TrainingJob, S3 URI, etc.
+       role_arn="arn:aws:iam::123456789012:role/MySageMakerRole",
+       instance_type="ml.p4d.24xlarge",
+       image_uri="my-inference-image:latest",
+   )
+
+   # build() checks for an existing Model with matching source tag
+   builder.build(region="us-east-1", reuse_resources=True)
+
+   # deploy() checks for an existing Endpoint with matching source tag
+   endpoint = builder.deploy(
+       endpoint_name="my-endpoint",
+       instance_type="ml.p4d.24xlarge",
+       reuse_resources=True,
+   )
+   # If a match is found: returns the existing endpoint
+   # If no match: creates a new endpoint as normal
+
+Without ``reuse_resources=True`` (the default), every deploy creates a new endpoint. The
+model-source tag is still applied so that future deploys with reuse enabled can discover it.
+
+.. note::
+
+   The ``reuse_resources`` flag must be passed to each call independently — it is not
+   inherited between ``build()`` and ``deploy()``.
+
+   Inference component builds (``modelbuilder_list``) manage their own reuse by component
+   name and bypass the endpoint-return reuse gate.

docs/model_customization/evaluation.rst

@@ -17,3 +17,44 @@ Launch evaluation jobs with the following options:
    ../../v3-examples/model-customization-examples/custom_scorer_demo
    ../../v3-examples/model-customization-examples/benchmark_demo
    ../../v3-examples/nova-examples/evaluation-benchmark-and-custom-scorer
+
+
+Dry-Run Validation
+-------------------
+
+Pass ``dry_run=True`` to ``evaluate()`` to validate your evaluation configuration without
+submitting a job or consuming compute. The SDK runs all validation (IAM role resolution,
+model resolution, dataset path existence) and then stops before launching the evaluation
+pipeline. Returns ``None`` on success, raises ``ValueError`` on validation failure.
+
+Supported on all evaluators: ``BenchMarkEvaluator``, ``CustomScorerEvaluator``, and
+``LLMAsJudgeEvaluator``.
+
+.. code-block:: python
+
+   from sagemaker.train.evaluate import BenchMarkEvaluator, get_benchmarks
+
+   Benchmark = get_benchmarks()
+
+   evaluator = BenchMarkEvaluator(
+       benchmark=Benchmark.MMLU,
+       model="arn:aws:sagemaker:us-east-1:123456789012:model-package/my-models/3",
+       s3_output_path="s3://my-bucket/eval-output/",
+   )
+
+   # Validate without launching — returns None on success
+   evaluator.evaluate(dry_run=True)
+
+.. code-block:: python
+
+   from sagemaker.train.evaluate import CustomScorerEvaluator
+
+   evaluator = CustomScorerEvaluator(
+       model="my-model-package-arn",
+       evaluation_dataset="s3://my-bucket/eval-data.jsonl",
+       s3_output_path="s3://my-bucket/custom-eval-output/",
+       scorer_function=my_scorer,
+   )
+
+   # Raises ValueError if dataset path does not exist
+   evaluator.evaluate(dry_run=True)

docs/model_customization/model_customization.rst

@@ -54,6 +54,143 @@ Key Features
   with clear precedence. Use ``get_resolved_recipe()`` to inspect the merged configuration
   before job submission. See :doc:`finetuning_serverful` and :doc:`finetuning_hyperpod` for examples.
 
+**Dry-Run Validation**
+  Pass ``dry_run=True`` to ``train()`` to run the validation steps without submitting a job 
+  or consuming compute. Returns ``None`` on success, raises ``ValueError`` on validation failure.
+
+  Supported on all trainers (SFT, DPO, RLVR, RLAIF, CPT) and ``ModelTrainer.train()``.
+  Works across serverless, serverful (``TrainingJobCompute``), and HyperPod
+  (``HyperPodCompute``) compute modes. Validates S3 URIs, DataSet ARNs, and ``DataSet``
+  objects.
+
+  Also available on evaluators — see :doc:`evaluation` for details.
+
+  .. code-block:: python
+
+     from sagemaker.train import SFTTrainer
+     from sagemaker.train.common import TrainingType
+
+     trainer = SFTTrainer(
+         model="meta-textgeneration-llama-3-2-1b-instruct",
+         training_type=TrainingType.LORA,
+         model_package_group="my-finetuned-models",
+         training_dataset="s3://my-bucket/train.jsonl",
+         accept_eula=True,
+     )
+
+     # Validate without submitting — returns None on success
+     trainer.train(dry_run=True)
+
+**Job Notifications**
+  Receive SNS notifications when training jobs complete, fail, or stop. Uses EventBridge
+  rules to route SageMaker Training Job status changes to your SNS topic. 
+
+  The SDK creates one rule per unique config (topic + events + prefix). Re-running with 
+  the same config reuses the existing rule. Different configs create separate rules.
+
+  .. note::
+    Supported for SMTJ (serverful and serverless) compute only. HyperPod is not currently supported.
+
+  .. code-block:: python
+
+    trainer = SFTTrainer(
+        model="nova-textgeneration-micro",
+        training_dataset="s3://my-bucket/train.jsonl",
+        accept_eula=True,
+        notifications={
+            "sns_topic_arn": "arn:aws:sns:us-east-1:123456789012:my-topic",  # Required
+            "events": ["Completed", "Failed"],    # Optional (default: Completed, Failed, Stopped)
+            "job_name_prefix": "my-team-sft-",    # Optional: filter by job name
+            "event_bus_arn": "arn:aws:events:us-east-1:123456789012:event-bus/custom-bus" # Optional
+        },
+    )
+    job = trainer.train(wait=False)  # Notification sent on completion
+
+    # Access the rule ARN
+    print(trainer.notification_rule_arn)
+
+    # List and manage rules
+    rules = trainer.list_notification_rules()
+    trainer.delete_notification_rule(rule_arn=trainer.notification_rule_arn)
+
+  **Prerequisites:**
+
+  - An SNS topic (and subscription) with a resource policy allowing ``events.amazonaws.com``. 
+    To set up a topic and subscription, see `Creating an SNS topic and subscription <https://docs.aws.amazon.com/sns/latest/dg/sns-create-subscribe-endpoint-to-topic.html>`_. 
+  - IAM permissions: ``events:PutRule``, ``events:PutTargets``, ``events:ListRules``,
+    ``events:RemoveTargets``, ``events:DeleteRule``
+
+**Monitoring: show_metrics()**
+  Plot training metrics after a job completes. Works across all compute types.
+
+  - **Nova models**: Metrics parsed from CloudWatch logs.
+  - **OSS models**: Metrics pulled from MLflow.
+
+  .. code-block:: python
+
+    # Plot all available metrics
+    df = trainer.show_metrics()
+
+    # Plot specific metrics
+    df = trainer.show_metrics(metrics=["training_loss", "lr"])
+
+    # Filter by step range
+    df = trainer.show_metrics(starting_step=10, ending_step=100)
+
+    # Filter by time window
+    from datetime import datetime
+    df = trainer.show_metrics(
+        start_time=datetime(2026, 1, 1, 10, 0, 0),
+        end_time=datetime(2026, 1, 1, 12, 0, 0),
+    )
+
+  **After a kernel restart:**
+
+  .. code-block:: python
+
+    # Standalone (SMTJ)
+    from sagemaker.train import plot_training_metrics
+    plot_training_metrics("my-sft-job")
+
+    # Re-attach (HyperPod — needs cluster name for log group resolution)
+    from sagemaker.train import SFTTrainer
+    from sagemaker.core.training.configs import HyperPodCompute
+
+    trainer = SFTTrainer(
+        model="nova-textgeneration-micro",
+        training_dataset="s3://unused",
+        compute=HyperPodCompute(cluster_name="my-cluster", instance_type="ml.p5.48xlarge")
+    )
+    trainer._latest_training_job = "my-hp-job"
+    df = trainer.show_metrics()
+
+**Monitoring: stream_logs()**
+  Stream CloudWatch logs in real-time while a job is running.
+
+  .. code-block:: python
+
+    # Start training non-blocking
+    job = trainer.train(wait=False)
+
+    # Stream logs (blocks until job completes or Ctrl+C)
+    trainer.stream_logs()
+
+    # Custom polling interval (seconds)
+    trainer.stream_logs(poll=10)
+
+    # Stream from a specific start time - providing this will speed up execution.
+    from datetime import datetime
+    trainer.stream_logs(start_time=datetime(2026, 1, 1, 15, 0, 0))
+
+  .. note::
+
+    - **SMTJ**: Streaming auto-stops when the job reaches a terminal state.
+    - **HyperPod**: Streaming runs until you press Ctrl+C. Logs may take a few minutes
+      to propagate to CloudWatch on first run.
+
+
+----
+
 
 .. toctree::
    :maxdepth: 1

sagemaker-core/src/sagemaker/core/helper/iam_policies.py

@@ -813,3 +813,36 @@
         },
     },
 }
+
+# Actions the *caller* must have to orchestrate Pipeline-based evaluations
+# directly. These actions must be held by whoever calls evaluator.evaluate(),
+# NOT by the job execution role (which is covered by role_type="training").
+# See verify_evaluation_caller_permissions() in iam_role_resolver.
+EVALUATION_CALLER_ACTIONS = (
+    # Pipeline orchestration
+    "sagemaker:CreatePipeline",
+    "sagemaker:UpdatePipeline",
+    "sagemaker:DescribePipeline",
+    "sagemaker:ListPipelines",
+    "sagemaker:StartPipelineExecution",
+    "sagemaker:DescribePipelineExecution",
+    "sagemaker:ListPipelineExecutionSteps",
+    "sagemaker:StopPipelineExecution",
+    "sagemaker:ListTags",
+    "sagemaker:AddTags",
+    "sagemaker:DescribeTrainingJob",
+    "iam:PassRole",
+    # Model resolution (DescribeHubContent called at construction time under caller creds)
+    "sagemaker:DescribeHubContent",
+    "sagemaker:ListHubContents",
+    "sagemaker:DescribeHub",
+    "sagemaker:ListHubs",
+    # Lineage (artifact creation/lookup runs under caller before pipeline starts)
+    "sagemaker:CreateArtifact",
+    "sagemaker:ListArtifacts",
+    "sagemaker:DescribeArtifact",
+    # S3 access (config/benchmark upload + output path validation)
+    "s3:PutObject",
+    "s3:GetObject",
+    "s3:ListBucket",
+)

sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

@@ -42,6 +42,12 @@
     "eks:AccessKubernetesApi",
 )
 
+# Actions the *caller* must have to orchestrate Pipeline-based evaluations
+# directly. These actions must be held by whoever calls evaluator.evaluate(),
+# NOT by the job execution role (which is covered by role_type="training").
+# See verify_evaluation_caller_permissions().
+from sagemaker.core.helper.iam_policies import EVALUATION_CALLER_ACTIONS
+
 
 class RoleValidationError(Exception):
     """Raised when the resolved IAM role lacks the permissions/trust an operation needs.
@@ -677,6 +683,81 @@ def verify_hyperpod_connect_permissions(
     return True
 
 
+def verify_evaluation_caller_permissions(
+    sagemaker_session=None,
+) -> Optional[bool]:
+    """Verify the caller can orchestrate SageMaker Pipeline-based evaluations.
+
+    The evaluate module submits work via SageMaker Pipelines — creating, updating,
+    starting, and describing pipelines and their executions. These actions run under
+    the *caller's* credentials (the notebook user, Lambda, or CI role), NOT under
+    the job execution role passed to the pipeline. This function simulates the
+    required pipeline-orchestration actions on the caller identity and raises
+    :class:`RoleValidationError` when they are missing.
+
+    Args:
+        sagemaker_session: SageMaker session (used to get the boto session).
+
+    Returns:
+        True  — all evaluation caller actions are allowed.
+        None  — could not be determined (caller is not a role, or cannot simulate).
+
+    Raises:
+        RoleValidationError: If permissions are definitively denied.
+    """
+    boto_session = _get_boto_session(sagemaker_session)
+    sts_client = boto_session.client("sts")
+    iam_client = boto_session.client("iam")
+
+    caller_identity = sts_client.get_caller_identity()
+    caller_arn = caller_identity["Arn"]
+    account_id = caller_identity["Account"]
+    partition = _partition_from_arn(caller_arn)
+
+    caller_role_arn = _resolve_caller_role_arn(iam_client, caller_arn, account_id, partition)
+    if not caller_role_arn:
+        logger.info(
+            "Could not resolve a caller role to verify evaluation pipeline "
+            "permissions; errors will surface at pipeline creation time."
+        )
+        return None
+
+    try:
+        denied = _simulate_denied_actions(
+            iam_client, caller_role_arn, list(EVALUATION_CALLER_ACTIONS)
+        )
+    except ClientError as e:
+        error_code = e.response.get("Error", {}).get("Code", "")
+        if error_code in ("AccessDenied", "AccessDeniedException"):
+            logger.info(
+                "Cannot simulate evaluation caller permissions for '%s' (access "
+                "denied to iam:SimulatePrincipalPolicy); errors will surface at "
+                "pipeline creation time.",
+                caller_role_arn,
+            )
+            return None
+        raise
+
+    if denied:
+        message = (
+            f"Your identity '{caller_role_arn}' is missing IAM permissions required "
+            f"to orchestrate SageMaker Pipeline-based evaluations: "
+            f"{', '.join(denied)}. "
+            f"The evaluation execution role was resolved successfully, but creating "
+            f"and starting the evaluation pipeline runs as YOUR credentials. "
+            f"Grant these actions to your identity (scoped to "
+            f"arn:{partition}:sagemaker:*:{account_id}:pipeline/*) or use the "
+            f"AmazonSageMakerFullAccess managed policy."
+        )
+        raise RoleValidationError(message)
+
+    logger.info(
+        "Caller '%s' has the evaluation pipeline orchestration permissions.",
+        caller_role_arn,
+    )
+    return True
+
+
 # ---------------------------------------------------------------------------
 # Opt-in IAM execution-role creation.
 #