From 691db06405f28b9a51df597e89a7922b0506c4a8 Mon Sep 17 00:00:00 2001
From: jzhaoqwa <jzhaoqwa@amazon.com>
Date: Mon, 3 Aug 2026 17:10:28 -0700
Subject: [PATCH 1/4] fix: usuability update from feedback

---
 docs/model_customization/index.rst            |   1 +
 .../notifications_setup.rst                   | 236 ++++++++++++++++++
 .../src/sagemaker/train/base_trainer.py       |  57 +++--
 .../train/common_utils/log_streamer.py        |  13 +-
 .../train/common_utils/notifications.py       | 118 +++------
 5 files changed, 320 insertions(+), 105 deletions(-)
 create mode 100644 docs/model_customization/notifications_setup.rst

diff --git a/docs/model_customization/index.rst b/docs/model_customization/index.rst
index 80363939c8..018586d224 100644
--- a/docs/model_customization/index.rst
+++ b/docs/model_customization/index.rst
@@ -23,3 +23,4 @@ Key Benefits
    open_weight_model_customization
    nova
    evaluation
+   notifications_setup
diff --git a/docs/model_customization/notifications_setup.rst b/docs/model_customization/notifications_setup.rst
new file mode 100644
index 0000000000..cbe73b3b46
--- /dev/null
+++ b/docs/model_customization/notifications_setup.rst
@@ -0,0 +1,236 @@
+Setting Up Training Job Notifications
+========================================
+
+Get notified when your training jobs complete, fail, or stop via SNS email/SMS
+alerts. This guide walks through creating the prerequisite SNS topic and
+configuring your trainer to send notifications.
+
+Architecture
+-------------
+
+.. code-block:: text
+
+   SageMaker Training Job ──► EventBridge Rule ──► SNS Topic ──► Email/SMS/Slack
+
+The SDK creates an EventBridge rule that listens for training job status changes
+and routes them to your SNS topic. You provide the topic; the SDK handles the
+wiring.
+
+Prerequisites
+--------------
+
+You need:
+
+1. An SNS topic with a policy allowing EventBridge to publish to it
+2. A subscription on that topic (email, SMS, Slack, etc.)
+3. IAM permissions for EventBridge rule management (see below)
+
+Step 1: Create an SNS Topic
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**AWS Console**
+
+1. Go to **Amazon SNS** → **Topics** → **Create topic**
+2. Choose **Standard** type
+3. Name it (e.g., ``my-training-alerts``)
+4. Click **Create topic**
+5. Note the **Topic ARN** (e.g., ``arn:aws:sns:us-east-1:123456789012:my-training-alerts``)
+
+**AWS CLI**
+
+.. code-block:: bash
+
+   aws sns create-topic --name my-training-alerts
+
+Step 2: Allow EventBridge to Publish
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The topic needs a resource policy granting EventBridge publish access.
+
+**AWS Console**
+
+1. Open your topic → **Access policy** tab → **Edit**
+2. Add this statement to the policy's ``Statement`` array:
+
+.. code-block:: json
+
+   {
+       "Sid": "AllowEventBridgePublish",
+       "Effect": "Allow",
+       "Principal": {"Service": "events.amazonaws.com"},
+       "Action": "SNS:Publish",
+       "Resource": "arn:aws:sns:us-east-1:123456789012:my-training-alerts",
+       "Condition": {
+           "StringEquals": {"AWS:SourceAccount": "123456789012"}
+       }
+   }
+
+**AWS CLI**
+
+.. code-block:: bash
+
+   TOPIC_ARN="arn:aws:sns:us-east-1:123456789012:my-training-alerts"
+   ACCOUNT_ID="123456789012"
+
+   aws sns set-topic-attributes \
+       --topic-arn $TOPIC_ARN \
+       --attribute-name Policy \
+       --attribute-value '{
+           "Version": "2008-10-17",
+           "Statement": [{
+               "Sid": "AllowEventBridgePublish",
+               "Effect": "Allow",
+               "Principal": {"Service": "events.amazonaws.com"},
+               "Action": "SNS:Publish",
+               "Resource": "'$TOPIC_ARN'",
+               "Condition": {"StringEquals": {"AWS:SourceAccount": "'$ACCOUNT_ID'"}}
+           }]
+       }'
+
+Step 3: Subscribe to the Topic
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**Email**
+
+.. code-block:: bash
+
+   aws sns subscribe \
+       --topic-arn arn:aws:sns:us-east-1:123456789012:my-training-alerts \
+       --protocol email \
+       --notification-endpoint you@example.com
+
+Check your inbox and confirm the subscription.
+
+**SMS**
+
+.. code-block:: bash
+
+   aws sns subscribe \
+       --topic-arn arn:aws:sns:us-east-1:123456789012:my-training-alerts \
+       --protocol sms \
+       --notification-endpoint +15551234567
+
+Step 4: Use with the SDK
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Pass the topic ARN in the ``notifications`` config when constructing your trainer:
+
+.. code-block:: python
+
+   from sagemaker.train import SFTTrainer
+   from sagemaker.train.common import TrainingType
+   from sagemaker.core.training.configs import TrainingJobCompute
+
+   trainer = SFTTrainer(
+       model="amazon.nova-2-lite-v1",
+       training_type=TrainingType.LORA,
+       training_dataset="s3://my-bucket/data/train.jsonl",
+       compute=TrainingJobCompute(instance_type="ml.p4d.24xlarge"),
+       notifications={
+           "sns_topic_arn": "arn:aws:sns:us-east-1:123456789012:my-training-alerts",
+       },
+   )
+
+   trainer.train()
+
+Configuration Options
+~~~~~~~~~~~~~~~~~~~~~~
+
+The ``notifications`` dict supports:
+
+.. list-table::
+   :header-rows: 1
+   :widths: 25 10 65
+
+   * - Key
+     - Required
+     - Description
+   * - ``sns_topic_arn``
+     - Yes
+     - ARN of your SNS topic
+   * - ``events``
+     - No
+     - List of statuses to notify on. Default: ``["Completed", "Failed", "Stopped"]``.
+       Valid values: ``Completed``, ``Failed``, ``Stopped``, ``InProgress``.
+   * - ``event_bus_arn``
+     - No
+     - Custom EventBridge bus ARN. Defaults to the account's default event bus.
+   * - ``job_name_prefix``
+     - No
+     - Only notify for jobs whose name starts with this prefix.
+
+Example: notify only on failures for jobs matching a prefix:
+
+.. code-block:: python
+
+   notifications={
+       "sns_topic_arn": "arn:aws:sns:us-east-1:123456789012:my-training-alerts",
+       "events": ["Failed"],
+       "job_name_prefix": "prod-sft-",
+   }
+
+Managing Notification Rules
+-----------------------------
+
+List active rules:
+
+.. code-block:: python
+
+   rules = trainer.list_notification_rules()
+   for rule in rules:
+       print(f"{rule['name']} ({rule['state']})")
+
+Delete a rule:
+
+.. code-block:: python
+
+   trainer.delete_notification_rule(rule_arn="arn:aws:events:us-east-1:123456789012:rule/sm-pysdk-job-notif-abc123")
+
+Required IAM Permissions
+--------------------------
+
+The caller (your IAM role or user) needs these permissions:
+
+.. code-block:: json
+
+   {
+       "Version": "2012-10-17",
+       "Statement": [
+           {
+               "Effect": "Allow",
+               "Action": [
+                   "events:PutRule",
+                   "events:PutTargets",
+                   "events:ListRules",
+                   "events:ListTargetsByRule",
+                   "events:RemoveTargets",
+                   "events:DeleteRule"
+               ],
+               "Resource": "arn:aws:events:*:*:rule/sm-pysdk-job-notif-*"
+           },
+           {
+               "Effect": "Allow",
+               "Action": "sns:GetTopicAttributes",
+               "Resource": "arn:aws:sns:*:*:my-training-alerts"
+           }
+       ]
+   }
+
+Troubleshooting
+-----------------
+
+**PermissionError: Missing permissions to manage EventBridge rules**
+
+Your IAM identity needs ``events:PutRule`` and ``events:PutTargets``. Ask your
+admin to attach the policy above.
+
+**ValueError: SNS topic not found**
+
+Verify the topic ARN is correct and exists in the same region as your
+SageMaker session. Ensure you have ``sns:GetTopicAttributes`` permission.
+
+**Not receiving notifications**
+
+1. Confirm the SNS subscription is in ``Confirmed`` state (check in the Console)
+2. Verify the topic policy allows EventBridge to publish (Step 2 above)
+3. Check that the ``events`` list includes the status you're waiting for
diff --git a/sagemaker-train/src/sagemaker/train/base_trainer.py b/sagemaker-train/src/sagemaker/train/base_trainer.py
index 2d36016256..10ec7c6a27 100644
--- a/sagemaker-train/src/sagemaker/train/base_trainer.py
+++ b/sagemaker-train/src/sagemaker/train/base_trainer.py
@@ -627,7 +627,7 @@ def list_notification_rules(
             event_bus_arn=event_bus_arn,
         )
 
-    def stream_logs(self, poll: int = 5, start_time: Optional[Any] = None, num_lines: Optional[int] = None) -> None:
+    def stream_logs(self, poll: int = 5, start_time: Optional[Any] = None, tail_lines: Optional[int] = None) -> None:
         """Stream CloudWatch logs in real-time (like ``kubectl logs -f``).
 
         Continuously polls for new log events and prints them as they arrive.
@@ -641,10 +641,13 @@ def stream_logs(self, poll: int = 5, start_time: Optional[Any] = None, num_lines
                 attaching to a job that's already running. If not provided,
                 auto-resolved from the training job's start time (SMTJ) or
                 defaults to now (HyperPod).
-            num_lines: Optional maximum number of log lines to print. When
-                specified, streaming stops after this many lines have been
-                printed. Useful for long jobs where the full log is too verbose.
-                If not provided, streams all logs until the job completes.
+            tail_lines: Optional maximum number of most recent log lines to
+                print. Logs are returned in chronological order; when specified,
+                only the last ``tail_lines`` entries are shown (similar to
+                ``kubectl logs --tail``). Useful for quickly checking the latest
+                output of long-running jobs without scrolling through the full
+                history. If not provided, streams all logs until the job
+                completes.
 
         Raises:
             ValueError: If no training job has been run yet.
@@ -678,11 +681,11 @@ def stream_logs(self, poll: int = 5, start_time: Optional[Any] = None, num_lines
         compute = getattr(self, 'compute', None)
 
         if isinstance(compute, HyperPodCompute):
-            self._stream_logs_smhp(training_job, compute, poll, start_time_ms, num_lines=num_lines)
+            self._stream_logs_smhp(training_job, compute, poll, start_time_ms, tail_lines=tail_lines)
         else:
-            self._stream_logs_smtj(training_job, poll, start_time_ms, num_lines=num_lines)
+            self._stream_logs_smtj(training_job, poll, start_time_ms, tail_lines=tail_lines)
 
-    def _stream_logs_smtj(self, training_job, poll: int, start_time_ms=None, num_lines: Optional[int] = None) -> None:
+    def _stream_logs_smtj(self, training_job, poll: int, start_time_ms=None, tail_lines: Optional[int] = None) -> None:
         """Stream logs for an SMTJ training job."""
         from sagemaker.train.common_utils.log_streamer import (
             LogStreamer,
@@ -714,9 +717,9 @@ def _get_status() -> str:
             job = TrainingJob.get(training_job_name=job_name)
             return job.training_job_status
 
-        stream_log_loop(streamer, poll, _get_status, num_lines=num_lines)
+        stream_log_loop(streamer, poll, _get_status, tail_lines=tail_lines)
 
-    def _stream_logs_smhp(self, training_job, compute, poll: int, start_time_ms=None, num_lines: Optional[int] = None) -> None:
+    def _stream_logs_smhp(self, training_job, compute, poll: int, start_time_ms=None, tail_lines: Optional[int] = None) -> None:
         """Stream logs for a HyperPod job using filter_log_events polling."""
 
         if isinstance(training_job, str):
@@ -774,8 +777,8 @@ def _stream_logs_smhp(self, training_job, compute, poll: int, start_time_ms=None
                         if message:
                             print(f"{_CW_PREFIX}{message}")
                             lines_printed += 1
-                            if num_lines and lines_printed >= num_lines:
-                                logger.info(f"Reached num_lines limit ({num_lines}). Stopping log stream.")
+                            if tail_lines and lines_printed >= tail_lines:
+                                logger.info(f"Reached tail_lines limit ({tail_lines}). Stopping log stream.")
                                 return
                         ts = event.get("timestamp", 0)
                         if ts > last_timestamp:
@@ -812,8 +815,14 @@ def _stream_logs_smhp(self, training_job, compute, poll: int, start_time_ms=None
                 logger.info("Log streaming stopped by user.")
                 return
 
-    def _validate_instance_count(self, instance_count, sagemaker_session):
-        """Validate instance/node count against allowed values from SMHP recipe."""
+    def _validate_instance_count(self, instance_count, sagemaker_session, compute):
+        """Validate instance/node count against allowed values from SMHP recipe.
+
+        For HyperPod compute, raises ValueError on mismatch since the recipe's
+        node count constraints are hard requirements for distributed training.
+        For SMTJ compute, logs a warning instead since SMTJ may support counts
+        not listed in the SMHP recipe depending on the model.
+        """
         smhp_replicas_enum = _get_smhp_replicas_enum(
             model_name=self._model_name,
             customization_technique=self._customization_technique,
@@ -821,10 +830,18 @@ def _validate_instance_count(self, instance_count, sagemaker_session):
             sagemaker_session=sagemaker_session,
         )
         if smhp_replicas_enum and instance_count not in smhp_replicas_enum:
-            raise ValueError(
-                f"Node/Instance count '{instance_count}' is not supported. "
-                f"Allowed values: {sorted(smhp_replicas_enum)}."
-            )
+            if isinstance(compute, HyperPodCompute):
+                raise ValueError(
+                    f"Node/Instance count '{instance_count}' is not supported. "
+                    f"Allowed values: {sorted(smhp_replicas_enum)}."
+                )
+            else:
+                logger.warning(
+                    f"Instance count '{instance_count}' is not in the recommended values "
+                    f"{sorted(smhp_replicas_enum)} from the model recipe. "
+                    f"This may or may not work depending on the model. "
+                    f"Proceeding anyway for SMTJ compute."
+                )
         return smhp_replicas_enum
 
     def _validate_instance_type(self, instance_type, sagemaker_session):
@@ -954,7 +971,7 @@ def _channel_mount_path(dataset_uri, channel_name):
             )
 
         # Validate instance count against allowed values from SMHP recipe.
-        smhp_replicas_enum = self._validate_instance_count(compute.instance_count, sagemaker_session)
+        smhp_replicas_enum = self._validate_instance_count(compute.instance_count, sagemaker_session, compute)
 
         if smhp_replicas_enum:
             override_spec.setdefault("replicas", {})["enum"] = smhp_replicas_enum
@@ -1436,7 +1453,7 @@ def _train_hyperpod(self, training_dataset=None, validation_dataset=None,
         job_base_name = self.base_job_name or f"{self._model_name}-{self._customization_technique}"
 
         # Validate node_count against allowed values from SMHP recipe
-        self._validate_instance_count(compute.node_count, sagemaker_session)
+        self._validate_instance_count(compute.node_count, sagemaker_session, compute)
 
         # Resolve and validate the recipe (3-level merge: base → user recipe → overrides)
         try:
diff --git a/sagemaker-train/src/sagemaker/train/common_utils/log_streamer.py b/sagemaker-train/src/sagemaker/train/common_utils/log_streamer.py
index 3538c95200..570a3331e3 100644
--- a/sagemaker-train/src/sagemaker/train/common_utils/log_streamer.py
+++ b/sagemaker-train/src/sagemaker/train/common_utils/log_streamer.py
@@ -227,7 +227,7 @@ def stream_log_loop(
     streamer: LogStreamer,
     poll: int,
     status_fn: Callable[[], str],
-    num_lines: Optional[int] = None,
+    tail_lines: Optional[int] = None,
 ) -> None:
     """Run the standard log streaming loop.
 
@@ -238,19 +238,20 @@ def stream_log_loop(
     :param streamer: A configured LogStreamer instance.
     :param poll: Seconds between polls.
     :param status_fn: Callable that returns the current job status string.
-    :param num_lines: Optional maximum number of log lines to print.
-        When specified, streaming stops after this many lines.
+    :param tail_lines: Optional maximum number of most recent log lines to
+        print. When specified, streaming stops after this many lines have
+        been displayed.
     """
     _CW_PREFIX = "[CloudWatch] "
     lines_printed = 0
 
     def _print_event(ts_ms: int, message: str) -> bool:
-        """Print a log event. Returns True if num_lines limit reached."""
+        """Print a log event. Returns True if tail_lines limit reached."""
         nonlocal lines_printed
         print(f"{_CW_PREFIX}[{_format_timestamp(ts_ms)}] {message}")
         lines_printed += 1
-        if num_lines and lines_printed >= num_lines:
-            logger.info("Reached num_lines limit (%d). Stopping log stream.", num_lines)
+        if tail_lines and lines_printed >= tail_lines:
+            logger.info("Reached tail_lines limit (%d). Stopping log stream.", tail_lines)
             return True
         return False
 
diff --git a/sagemaker-train/src/sagemaker/train/common_utils/notifications.py b/sagemaker-train/src/sagemaker/train/common_utils/notifications.py
index eced509685..1f3f712559 100644
--- a/sagemaker-train/src/sagemaker/train/common_utils/notifications.py
+++ b/sagemaker-train/src/sagemaker/train/common_utils/notifications.py
@@ -3,6 +3,45 @@
 Manages EventBridge rules that route SageMaker Training Job status change
 events to user-provided SNS topics. Supports SMTJ (serverless and serverful)
 training jobs only.
+
+Prerequisites:
+    Before using notifications, you need an SNS topic with a policy that allows
+    EventBridge to publish to it. You can create one via the AWS Console, CLI,
+    or CloudFormation/CDK:
+
+    **AWS CLI example**::
+
+        # Create the topic
+        aws sns create-topic --name my-training-alerts
+
+        # Allow EventBridge to publish to it
+        aws sns set-topic-attributes \\
+            --topic-arn arn:aws:sns:<region>:<account-id>:my-training-alerts \\
+            --attribute-name Policy \\
+            --attribute-value '{
+                "Version": "2008-10-17",
+                "Statement": [{
+                    "Sid": "AllowEventBridgePublish",
+                    "Effect": "Allow",
+                    "Principal": {"Service": "events.amazonaws.com"},
+                    "Action": "SNS:Publish",
+                    "Resource": "arn:aws:sns:<region>:<account-id>:my-training-alerts",
+                    "Condition": {"StringEquals": {"AWS:SourceAccount": "<account-id>"}}
+                }]
+            }'
+
+        # Subscribe your email
+        aws sns subscribe \\
+            --topic-arn arn:aws:sns:<region>:<account-id>:my-training-alerts \\
+            --protocol email \\
+            --notification-endpoint you@example.com
+
+    Then pass the topic ARN when constructing a trainer::
+
+        trainer = SFTTrainer(
+            ...,
+            notifications={"sns_topic_arn": "arn:aws:sns:<region>:<account-id>:my-training-alerts"},
+        )
 """
 
 from __future__ import absolute_import
@@ -150,85 +189,6 @@ def _validate_sns_topic(sns_client, topic_arn: str) -> None:
         raise
 
 
-def _build_topic_policy(topic_arn: str, account_id: str) -> str:
-    """Build the SNS topic access policy for EventBridge notifications.
-
-    Includes the default owner statement (standard SNS actions scoped to the
-    topic's own account via ``AWS:SourceAccount``) plus an explicit statement
-    granting the EventBridge service principal ``SNS:Publish``.
-
-    Args:
-        topic_arn: The ARN of the SNS topic.
-        account_id: The AWS account ID that owns the topic.
-
-    Returns:
-        JSON string of the topic access policy.
-    """
-    policy = {
-        "Version": "2008-10-17",
-        "Id": "SageMakerNotificationsTopicPolicy",
-        "Statement": [
-            {
-                "Sid": "SNSTopicAdministration",
-                "Effect": "Allow",
-                "Principal": {"AWS": "*"},
-                "Action": [
-                    "SNS:Publish",
-                    "SNS:RemovePermission",
-                    "SNS:SetTopicAttributes",
-                    "SNS:DeleteTopic",
-                    "SNS:ListSubscriptionsByTopic",
-                    "SNS:GetTopicAttributes",
-                    "SNS:AddPermission",
-                    "SNS:Subscribe",
-                ],
-                "Resource": topic_arn,
-                "Condition": {"StringEquals": {"AWS:SourceAccount": account_id}},
-            },
-            {
-                "Sid": "AllowEventBridgePublish",
-                "Effect": "Allow",
-                "Principal": {"Service": "events.amazonaws.com"},
-                "Action": "SNS:Publish",
-                "Resource": topic_arn,
-                "Condition": {"StringEquals": {"AWS:SourceAccount": account_id}},
-            },
-        ],
-    }
-    return json.dumps(policy)
-
-
-def create_notification_topic(topic_name: str, sagemaker_session) -> str:
-    """Create an SNS topic configured for EventBridge notifications.
-
-    Creates a new SNS topic and sets an access policy that allows EventBridge
-    in the same account to publish to it. The returned ARN can be passed
-    straight to :func:`enable_notifications`.
-
-    Args:
-        topic_name: Name of the SNS topic to create.
-        sagemaker_session: SageMaker session (provides boto_session).
-
-    Returns:
-        The ARN of the created SNS topic.
-    """
-    region_name = sagemaker_session.boto_session.region_name
-    sns_client = sagemaker_session.boto_session.client("sns", region_name=region_name)
-
-    response = sns_client.create_topic(Name=topic_name)
-    topic_arn = response["TopicArn"]
-    # topic ARN: arn:aws:sns:<region>:<account_id>:<name>
-    account_id = topic_arn.split(":")[4]
-
-    policy = _build_topic_policy(topic_arn, account_id)
-    sns_client.set_topic_attributes(
-        TopicArn=topic_arn, AttributeName="Policy", AttributeValue=policy
-    )
-
-    logger.info(f"Created SNS topic with EventBridge publish policy: {topic_arn}")
-    return topic_arn
-
-
 def enable_notifications(
     sns_topic_arn: str,
     sagemaker_session,

From 5ef92168432c690fc6a67f2c19726b98ada33c5d Mon Sep 17 00:00:00 2001
From: jzhaoqwa <jzhaoqwa@amazon.com>
Date: Tue, 4 Aug 2026 13:41:00 -0700
Subject: [PATCH 2/4] update documentation

---
 .../notifications_setup.rst                   | 20 ++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/docs/model_customization/notifications_setup.rst b/docs/model_customization/notifications_setup.rst
index cbe73b3b46..687552b5b4 100644
--- a/docs/model_customization/notifications_setup.rst
+++ b/docs/model_customization/notifications_setup.rst
@@ -49,8 +49,9 @@ The topic needs a resource policy granting EventBridge publish access.
 
 **AWS Console**
 
-1. Open your topic → **Access policy** tab → **Edit**
-2. Add this statement to the policy's ``Statement`` array:
+1. Go to **Amazon SNS** → **Topics** → open your topic → **Access policy** tab → **Edit**
+2. Add this statement to the policy's ``Statement`` array (replace the ARN and
+   account ID with your own SNS topic ARN and AWS account ID):
 
 .. code-block:: json
 
@@ -65,6 +66,12 @@ The topic needs a resource policy granting EventBridge publish access.
        }
    }
 
+.. note::
+
+   Replace ``arn:aws:sns:us-east-1:123456789012:my-training-alerts`` with your
+   actual SNS topic ARN, and ``123456789012`` with your AWS account ID. These
+   must match so that only EventBridge in your account can publish to your topic.
+
 **AWS CLI**
 
 .. code-block:: bash
@@ -189,7 +196,8 @@ Delete a rule:
 Required IAM Permissions
 --------------------------
 
-The caller (your IAM role or user) needs these permissions:
+The caller (your IAM role or user) needs these permissions. Replace the SNS
+resource ARN with your actual topic ARN:
 
 .. code-block:: json
 
@@ -216,6 +224,12 @@ The caller (your IAM role or user) needs these permissions:
        ]
    }
 
+.. note::
+
+   The ``sns:GetTopicAttributes`` resource must match the SNS topic you created
+   in Step 1. You can use a wildcard (``arn:aws:sns:*:*:*``) for broader access
+   or scope it to your specific topic ARN for least privilege.
+
 Troubleshooting
 -----------------
 

From d4e20422d76d8cbbdae9a62edf3bdd6afe689243 Mon Sep 17 00:00:00 2001
From: jzhaoqwa <jzhaoqwa@amazon.com>
Date: Tue, 4 Aug 2026 15:17:46 -0700
Subject: [PATCH 3/4] docs: organize model-customization examples into
 serverless/serverful/deployment/evaluation subfolders

---
 docs/inference/deploy_finetuned.rst                       | 2 +-
 docs/model_customization/evaluation.rst                   | 8 ++++----
 docs/model_customization/finetuning_hyperpod.rst          | 2 +-
 docs/model_customization/finetuning_serverful.rst         | 2 +-
 docs/model_customization/model_customization.rst          | 8 ++++----
 docs/model_customization/nova.rst                         | 4 ++--
 docs/model_customization/nova_data_mixing.rst             | 2 +-
 .../open_weight_model_customization.rst                   | 2 +-
 .../bedrock-modelbuilder-deployment-nova.ipynb            | 0
 .../bedrock-modelbuilder-deployment.ipynb                 | 0
 .../model_builder_deployment_notebook.ipynb               | 0
 .../{ => evaluation}/benchmark_demo.ipynb                 | 0
 .../{ => evaluation}/custom_scorer_demo.ipynb             | 0
 .../{ => evaluation}/inspect_ai_evaluation_demo.ipynb     | 0
 .../{ => evaluation}/llm_as_judge_custom_model_demo.ipynb | 0
 .../{ => evaluation}/llm_as_judge_demo.ipynb              | 0
 .../recipe_override_evaluator_example.ipynb               | 0
 .../{ => serverful}/cpt_data_mixing_hyperpod.ipynb        | 0
 .../{ => serverful}/sft_finetuning_hyperpod.ipynb         | 0
 .../{ => serverful}/sft_finetuning_serverful_smtj.ipynb   | 0
 .../{ => serverless}/ai_registry_example.ipynb            | 0
 .../dpo_trainer_example_notebook_v3_prod.ipynb            | 0
 .../mtrl_finetuning_example_notebook_v3_prod.ipynb        | 0
 .../{ => serverless}/nova_data_mixing.ipynb               | 0
 .../recipe_override_sft_trainer_example.ipynb             | 0
 .../rlaif_finetuning_example_notebook_v3_prod.ipynb       | 0
 .../rlvr_finetuning_example_notebook_v3_prod.ipynb        | 0
 .../serverless_e2e_example.ipynb}                         | 8 ++++----
 .../sft_finetuning_example_notebook_pysdk_prod_v3.ipynb   | 0
 29 files changed, 19 insertions(+), 19 deletions(-)
 rename v3-examples/model-customization-examples/{ => deployment}/bedrock-modelbuilder-deployment-nova.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => deployment}/bedrock-modelbuilder-deployment.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => deployment}/model_builder_deployment_notebook.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/benchmark_demo.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/custom_scorer_demo.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/inspect_ai_evaluation_demo.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/llm_as_judge_custom_model_demo.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/llm_as_judge_demo.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => evaluation}/recipe_override_evaluator_example.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverful}/cpt_data_mixing_hyperpod.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverful}/sft_finetuning_hyperpod.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverful}/sft_finetuning_serverful_smtj.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/ai_registry_example.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/dpo_trainer_example_notebook_v3_prod.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/mtrl_finetuning_example_notebook_v3_prod.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/nova_data_mixing.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/recipe_override_sft_trainer_example.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/rlaif_finetuning_example_notebook_v3_prod.ipynb (100%)
 rename v3-examples/model-customization-examples/{ => serverless}/rlvr_finetuning_example_notebook_v3_prod.ipynb (100%)
 rename v3-examples/model-customization-examples/{sm-studio-nova-training-job-sample-notebook.ipynb => serverless/serverless_e2e_example.ipynb} (98%)
 rename v3-examples/model-customization-examples/{ => serverless}/sft_finetuning_example_notebook_pysdk_prod_v3.ipynb (100%)

diff --git a/docs/inference/deploy_finetuned.rst b/docs/inference/deploy_finetuned.rst
index e67389a02d..43b7cb052a 100644
--- a/docs/inference/deploy_finetuned.rst
+++ b/docs/inference/deploy_finetuned.rst
@@ -7,4 +7,4 @@ Deploy models that have been fine-tuned through the :doc:`Model Customization <.
    :maxdepth: 1
 
    Deploy to SageMaker Endpoint <../model_customization/deploy_sagemaker_endpoint>
-   Deploy to Amazon Bedrock <../../v3-examples/model-customization-examples/bedrock-modelbuilder-deployment>
+   Deploy to Amazon Bedrock <../../v3-examples/model-customization-examples/deployment/bedrock-modelbuilder-deployment>
diff --git a/docs/model_customization/evaluation.rst b/docs/model_customization/evaluation.rst
index f5e3f6a2c8..db704166cf 100644
--- a/docs/model_customization/evaluation.rst
+++ b/docs/model_customization/evaluation.rst
@@ -12,10 +12,10 @@ Launch evaluation jobs with the following options:
 .. toctree::
    :maxdepth: 1
 
-   ../../v3-examples/model-customization-examples/llm_as_judge_demo
-   ../../v3-examples/model-customization-examples/inspect_ai_evaluation_demo
-   ../../v3-examples/model-customization-examples/custom_scorer_demo
-   ../../v3-examples/model-customization-examples/benchmark_demo
+   ../../v3-examples/model-customization-examples/evaluation/llm_as_judge_demo
+   ../../v3-examples/model-customization-examples/evaluation/inspect_ai_evaluation_demo
+   ../../v3-examples/model-customization-examples/evaluation/custom_scorer_demo
+   ../../v3-examples/model-customization-examples/evaluation/benchmark_demo
    ../../v3-examples/nova-examples/evaluation-benchmark-and-custom-scorer
 
 
diff --git a/docs/model_customization/finetuning_hyperpod.rst b/docs/model_customization/finetuning_hyperpod.rst
index 5d35674a88..ec0d240de8 100644
--- a/docs/model_customization/finetuning_hyperpod.rst
+++ b/docs/model_customization/finetuning_hyperpod.rst
@@ -177,4 +177,4 @@ Interactive Notebook
 .. toctree::
    :maxdepth: 1
 
-   ../../v3-examples/model-customization-examples/sft_finetuning_hyperpod
+   ../../v3-examples/model-customization-examples/serverful/sft_finetuning_hyperpod
diff --git a/docs/model_customization/finetuning_serverful.rst b/docs/model_customization/finetuning_serverful.rst
index 49e9555947..fe3605d1e6 100644
--- a/docs/model_customization/finetuning_serverful.rst
+++ b/docs/model_customization/finetuning_serverful.rst
@@ -138,4 +138,4 @@ Interactive Notebook
 .. toctree::
    :maxdepth: 1
 
-   ../../v3-examples/model-customization-examples/sft_finetuning_serverful_smtj
+   ../../v3-examples/model-customization-examples/serverful/sft_finetuning_serverful_smtj
diff --git a/docs/model_customization/model_customization.rst b/docs/model_customization/model_customization.rst
index eeb0635e92..6e0095184d 100644
--- a/docs/model_customization/model_customization.rst
+++ b/docs/model_customization/model_customization.rst
@@ -196,9 +196,9 @@ Key Features
    :maxdepth: 1
    :caption: Customization Techniques
 
-   SFT Finetuning <../../v3-examples/model-customization-examples/sft_finetuning_example_notebook_pysdk_prod_v3>
-   DPOTrainer Finetuning <../../v3-examples/model-customization-examples/dpo_trainer_example_notebook_v3_prod>
-   RLAIF Finetuning <../../v3-examples/model-customization-examples/rlaif_finetuning_example_notebook_v3_prod>
-   RLVR Finetuning <../../v3-examples/model-customization-examples/rlvr_finetuning_example_notebook_v3_prod>
+   SFT Finetuning <../../v3-examples/model-customization-examples/serverless/sft_finetuning_example_notebook_pysdk_prod_v3>
+   DPOTrainer Finetuning <../../v3-examples/model-customization-examples/serverless/dpo_trainer_example_notebook_v3_prod>
+   RLAIF Finetuning <../../v3-examples/model-customization-examples/serverless/rlaif_finetuning_example_notebook_v3_prod>
+   RLVR Finetuning <../../v3-examples/model-customization-examples/serverless/rlvr_finetuning_example_notebook_v3_prod>
    Fine-Tuning with Serverful Training Jobs <finetuning_serverful>
    Fine-Tuning with HyperPod <finetuning_hyperpod>
diff --git a/docs/model_customization/nova.rst b/docs/model_customization/nova.rst
index 5de821d2ff..18253278ec 100644
--- a/docs/model_customization/nova.rst
+++ b/docs/model_customization/nova.rst
@@ -13,5 +13,5 @@ Adapt Amazon Nova foundation models to your specific use cases through fine-tuni
    :caption: Nova Customization Guides
 
    nova_data_mixing
-      CPT Training on HyperPod <../../v3-examples/model-customization-examples/cpt_data_mixing_hyperpod>
-   ../../v3-examples/model-customization-examples/sm-studio-nova-training-job-sample-notebook
+      CPT Training on HyperPod <../../v3-examples/model-customization-examples/serverful/cpt_data_mixing_hyperpod>
+   Serverless End-to-End Example <../../v3-examples/model-customization-examples/serverless/serverless_e2e_example>
diff --git a/docs/model_customization/nova_data_mixing.rst b/docs/model_customization/nova_data_mixing.rst
index f1bbde1a30..afec2be54c 100644
--- a/docs/model_customization/nova_data_mixing.rst
+++ b/docs/model_customization/nova_data_mixing.rst
@@ -132,4 +132,4 @@ Interactive Notebook
 ---------------------
 
 For a complete walkthrough, see the
-:doc:`Data Mixing notebook <../../v3-examples/model-customization-examples/nova_data_mixing>`.
+:doc:`Data Mixing notebook <../../v3-examples/model-customization-examples/serverless/nova_data_mixing>`.
diff --git a/docs/model_customization/open_weight_model_customization.rst b/docs/model_customization/open_weight_model_customization.rst
index c4b2098e5d..bd655f747f 100644
--- a/docs/model_customization/open_weight_model_customization.rst
+++ b/docs/model_customization/open_weight_model_customization.rst
@@ -6,5 +6,5 @@ This section walks you through the process to get started with open weight model
 .. toctree::
    :maxdepth: 1
 
-   ../../v3-examples/model-customization-examples/ai_registry_example
+   ../../v3-examples/model-customization-examples/serverless/ai_registry_example
    model_customization
diff --git a/v3-examples/model-customization-examples/bedrock-modelbuilder-deployment-nova.ipynb b/v3-examples/model-customization-examples/deployment/bedrock-modelbuilder-deployment-nova.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/bedrock-modelbuilder-deployment-nova.ipynb
rename to v3-examples/model-customization-examples/deployment/bedrock-modelbuilder-deployment-nova.ipynb
diff --git a/v3-examples/model-customization-examples/bedrock-modelbuilder-deployment.ipynb b/v3-examples/model-customization-examples/deployment/bedrock-modelbuilder-deployment.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/bedrock-modelbuilder-deployment.ipynb
rename to v3-examples/model-customization-examples/deployment/bedrock-modelbuilder-deployment.ipynb
diff --git a/v3-examples/model-customization-examples/model_builder_deployment_notebook.ipynb b/v3-examples/model-customization-examples/deployment/model_builder_deployment_notebook.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/model_builder_deployment_notebook.ipynb
rename to v3-examples/model-customization-examples/deployment/model_builder_deployment_notebook.ipynb
diff --git a/v3-examples/model-customization-examples/benchmark_demo.ipynb b/v3-examples/model-customization-examples/evaluation/benchmark_demo.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/benchmark_demo.ipynb
rename to v3-examples/model-customization-examples/evaluation/benchmark_demo.ipynb
diff --git a/v3-examples/model-customization-examples/custom_scorer_demo.ipynb b/v3-examples/model-customization-examples/evaluation/custom_scorer_demo.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/custom_scorer_demo.ipynb
rename to v3-examples/model-customization-examples/evaluation/custom_scorer_demo.ipynb
diff --git a/v3-examples/model-customization-examples/inspect_ai_evaluation_demo.ipynb b/v3-examples/model-customization-examples/evaluation/inspect_ai_evaluation_demo.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/inspect_ai_evaluation_demo.ipynb
rename to v3-examples/model-customization-examples/evaluation/inspect_ai_evaluation_demo.ipynb
diff --git a/v3-examples/model-customization-examples/llm_as_judge_custom_model_demo.ipynb b/v3-examples/model-customization-examples/evaluation/llm_as_judge_custom_model_demo.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/llm_as_judge_custom_model_demo.ipynb
rename to v3-examples/model-customization-examples/evaluation/llm_as_judge_custom_model_demo.ipynb
diff --git a/v3-examples/model-customization-examples/llm_as_judge_demo.ipynb b/v3-examples/model-customization-examples/evaluation/llm_as_judge_demo.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/llm_as_judge_demo.ipynb
rename to v3-examples/model-customization-examples/evaluation/llm_as_judge_demo.ipynb
diff --git a/v3-examples/model-customization-examples/recipe_override_evaluator_example.ipynb b/v3-examples/model-customization-examples/evaluation/recipe_override_evaluator_example.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/recipe_override_evaluator_example.ipynb
rename to v3-examples/model-customization-examples/evaluation/recipe_override_evaluator_example.ipynb
diff --git a/v3-examples/model-customization-examples/cpt_data_mixing_hyperpod.ipynb b/v3-examples/model-customization-examples/serverful/cpt_data_mixing_hyperpod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/cpt_data_mixing_hyperpod.ipynb
rename to v3-examples/model-customization-examples/serverful/cpt_data_mixing_hyperpod.ipynb
diff --git a/v3-examples/model-customization-examples/sft_finetuning_hyperpod.ipynb b/v3-examples/model-customization-examples/serverful/sft_finetuning_hyperpod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/sft_finetuning_hyperpod.ipynb
rename to v3-examples/model-customization-examples/serverful/sft_finetuning_hyperpod.ipynb
diff --git a/v3-examples/model-customization-examples/sft_finetuning_serverful_smtj.ipynb b/v3-examples/model-customization-examples/serverful/sft_finetuning_serverful_smtj.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/sft_finetuning_serverful_smtj.ipynb
rename to v3-examples/model-customization-examples/serverful/sft_finetuning_serverful_smtj.ipynb
diff --git a/v3-examples/model-customization-examples/ai_registry_example.ipynb b/v3-examples/model-customization-examples/serverless/ai_registry_example.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/ai_registry_example.ipynb
rename to v3-examples/model-customization-examples/serverless/ai_registry_example.ipynb
diff --git a/v3-examples/model-customization-examples/dpo_trainer_example_notebook_v3_prod.ipynb b/v3-examples/model-customization-examples/serverless/dpo_trainer_example_notebook_v3_prod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/dpo_trainer_example_notebook_v3_prod.ipynb
rename to v3-examples/model-customization-examples/serverless/dpo_trainer_example_notebook_v3_prod.ipynb
diff --git a/v3-examples/model-customization-examples/mtrl_finetuning_example_notebook_v3_prod.ipynb b/v3-examples/model-customization-examples/serverless/mtrl_finetuning_example_notebook_v3_prod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/mtrl_finetuning_example_notebook_v3_prod.ipynb
rename to v3-examples/model-customization-examples/serverless/mtrl_finetuning_example_notebook_v3_prod.ipynb
diff --git a/v3-examples/model-customization-examples/nova_data_mixing.ipynb b/v3-examples/model-customization-examples/serverless/nova_data_mixing.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/nova_data_mixing.ipynb
rename to v3-examples/model-customization-examples/serverless/nova_data_mixing.ipynb
diff --git a/v3-examples/model-customization-examples/recipe_override_sft_trainer_example.ipynb b/v3-examples/model-customization-examples/serverless/recipe_override_sft_trainer_example.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/recipe_override_sft_trainer_example.ipynb
rename to v3-examples/model-customization-examples/serverless/recipe_override_sft_trainer_example.ipynb
diff --git a/v3-examples/model-customization-examples/rlaif_finetuning_example_notebook_v3_prod.ipynb b/v3-examples/model-customization-examples/serverless/rlaif_finetuning_example_notebook_v3_prod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/rlaif_finetuning_example_notebook_v3_prod.ipynb
rename to v3-examples/model-customization-examples/serverless/rlaif_finetuning_example_notebook_v3_prod.ipynb
diff --git a/v3-examples/model-customization-examples/rlvr_finetuning_example_notebook_v3_prod.ipynb b/v3-examples/model-customization-examples/serverless/rlvr_finetuning_example_notebook_v3_prod.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/rlvr_finetuning_example_notebook_v3_prod.ipynb
rename to v3-examples/model-customization-examples/serverless/rlvr_finetuning_example_notebook_v3_prod.ipynb
diff --git a/v3-examples/model-customization-examples/sm-studio-nova-training-job-sample-notebook.ipynb b/v3-examples/model-customization-examples/serverless/serverless_e2e_example.ipynb
similarity index 98%
rename from v3-examples/model-customization-examples/sm-studio-nova-training-job-sample-notebook.ipynb
rename to v3-examples/model-customization-examples/serverless/serverless_e2e_example.ipynb
index d4a6c65fc9..212d2d2a71 100644
--- a/v3-examples/model-customization-examples/sm-studio-nova-training-job-sample-notebook.ipynb
+++ b/v3-examples/model-customization-examples/serverless/serverless_e2e_example.ipynb
@@ -14,7 +14,7 @@
     "tags": []
    },
    "source": [
-    "# Model Customization using SageMaker Training Job"
+    "# Serverless Fine-Tuning and Deployment (End-to-End)"
    ]
   },
   {
@@ -803,9 +803,9 @@
    "metadata": {},
    "source": [
     "After you create a custom model, you can set up inference using one of the following options:\n",
-    "1. **Purchase Provisioned Throughput** – Purchase Provisioned Throughput for your model to set up dedicated compute capacity with guaranteed throughput for consistent performance and lower latency.\n",
+    "1. **Purchase Provisioned Throughput** \u2013 Purchase Provisioned Throughput for your model to set up dedicated compute capacity with guaranteed throughput for consistent performance and lower latency.\n",
     "For more information about Provisioned Throughput, see [Increase model invocation capacity with Provisioned Throughput in Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/prov-throughput.html). For more information about using custom models with Provisioned Throughput, [see Purchase Provisioned Throughput for a custom model](https://docs.aws.amazon.com/bedrock/latest/userguide/custom-model-use-pt.html).\n",
-    "2. **Deploy custom model for on-demand inference (only LoRA fine-tuned Amazon Nova models)** – To set up on-demand inference, you deploy the custom model with a custom model deployment. After you deploy the model, you invoke it using the ARN for the custom model deployment. With on-demand inference, you only pay for what you use and you don't need to set up provisioned compute resources.\n",
+    "2. **Deploy custom model for on-demand inference (only LoRA fine-tuned Amazon Nova models)** \u2013 To set up on-demand inference, you deploy the custom model with a custom model deployment. After you deploy the model, you invoke it using the ARN for the custom model deployment. With on-demand inference, you only pay for what you use and you don't need to set up provisioned compute resources.\n",
     "For more information about deploying custom models for on-demand inference, see [Deploy a custom model for on-demand inference](https://docs.aws.amazon.com/bedrock/latest/userguide/deploy-custom-model-on-demand.html)."
    ]
   },
@@ -1099,4 +1099,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 5
-}
+}
\ No newline at end of file
diff --git a/v3-examples/model-customization-examples/sft_finetuning_example_notebook_pysdk_prod_v3.ipynb b/v3-examples/model-customization-examples/serverless/sft_finetuning_example_notebook_pysdk_prod_v3.ipynb
similarity index 100%
rename from v3-examples/model-customization-examples/sft_finetuning_example_notebook_pysdk_prod_v3.ipynb
rename to v3-examples/model-customization-examples/serverless/sft_finetuning_example_notebook_pysdk_prod_v3.ipynb

From 1675758f977727ac0a1f5e8264caba9a727302b1 Mon Sep 17 00:00:00 2001
From: jzhaoqwa <jzhaoqwa@amazon.com>
Date: Tue, 4 Aug 2026 16:36:49 -0700
Subject: [PATCH 4/4] add job notification integ test

---
 .../tests/integ/train/test_notifications.py   | 345 ++++++++++++++++++
 1 file changed, 345 insertions(+)
 create mode 100644 sagemaker-train/tests/integ/train/test_notifications.py

diff --git a/sagemaker-train/tests/integ/train/test_notifications.py b/sagemaker-train/tests/integ/train/test_notifications.py
new file mode 100644
index 0000000000..789391755a
--- /dev/null
+++ b/sagemaker-train/tests/integ/train/test_notifications.py
@@ -0,0 +1,345 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+#     http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Integration test for training job notifications (EventBridge + SNS).
+
+Verifies the full end-to-end notification flow:
+1. Creating a trainer with `notifications` config creates an EventBridge rule
+2. The rule targets the correct SNS topic
+3. Stopping the job delivers a notification message through EventBridge → SNS → SQS
+4. The message contains the correct job name and status
+5. Cleanup removes the rule and temporary SQS queue
+
+Prerequisites:
+    - Active AWS credentials in us-east-1
+    - SNS topic: arn:aws:sns:us-east-1:784379639078:fine-tune-integ-test-job-notification
+      with EventBridge publish access policy already attached
+    - IAM permissions: events:PutRule, events:PutTargets, events:ListRules,
+      events:ListTargetsByRule, events:RemoveTargets, events:DeleteRule,
+      sqs:CreateQueue, sqs:DeleteQueue, sqs:GetQueueAttributes,
+      sqs:ReceiveMessage, sqs:SetQueueAttributes, sns:Subscribe, sns:Unsubscribe
+
+Run with:
+    export AWS_DEFAULT_REGION=us-east-1
+    pytest tests/integ/train/test_notifications.py -v -s
+"""
+from __future__ import absolute_import
+
+import json
+import logging
+import os
+import time
+import random
+
+import boto3
+import pytest
+
+from sagemaker.core.helper.session_helper import Session
+from sagemaker.train import SFTTrainer
+from sagemaker.train.common import TrainingType
+
+logger = logging.getLogger(__name__)
+
+# Test configuration
+REGION = "us-east-1"
+SNS_TOPIC_ARN = "arn:aws:sns:us-east-1:784379639078:fine-tune-integ-test-job-notification"
+ACCOUNT_ID = "784379639078"
+DATA_PREFIX = "notifications-integ"
+DATA_S3_KEY = f"{DATA_PREFIX}/sft_sample_data.jsonl"
+
+# Local sample data file (reuse existing test data)
+DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "data", "train")
+LOCAL_TRAINING_DATA = os.path.join(DATA_DIR, "sft_smtj_sample_data.jsonl")
+
+
+@pytest.fixture(scope="module")
+def sm_session():
+    """Create a SageMaker session in us-east-1."""
+    boto_session = boto3.Session(region_name=REGION)
+    return Session(boto_session=boto_session)
+
+
+@pytest.fixture(scope="module")
+def training_data_uri(sm_session):
+    """Upload training data to S3 if not present, return the S3 URI."""
+    bucket = sm_session.default_bucket()
+    s3_client = sm_session.boto_session.client("s3")
+
+    s3_uri = f"s3://{bucket}/{DATA_S3_KEY}"
+    try:
+        s3_client.head_object(Bucket=bucket, Key=DATA_S3_KEY)
+        logger.info(f"Training data already at {s3_uri}")
+    except s3_client.exceptions.ClientError:
+        logger.info(f"Uploading training data to {s3_uri}")
+        s3_client.upload_file(LOCAL_TRAINING_DATA, bucket, DATA_S3_KEY)
+
+    return s3_uri
+
+
+@pytest.fixture(scope="module")
+def sqs_subscriber(sm_session):
+    """Create a temporary SQS queue subscribed to the SNS topic for verification.
+
+    Yields a dict with queue_url and subscription_arn.
+    Cleans up the queue and subscription after the test module.
+    """
+    sqs_client = sm_session.boto_session.client("sqs", region_name=REGION)
+    sns_client = sm_session.boto_session.client("sns", region_name=REGION)
+
+    queue_name = f"notif-integ-test-{int(time.time())}-{random.randint(1000, 9999)}"
+
+    # Create SQS queue
+    queue_response = sqs_client.create_queue(
+        QueueName=queue_name,
+        Attributes={"MessageRetentionPeriod": "300"},  # 5 min retention
+    )
+    queue_url = queue_response["QueueUrl"]
+    logger.info(f"Created SQS queue: {queue_url}")
+
+    # Get queue ARN
+    attrs = sqs_client.get_queue_attributes(
+        QueueUrl=queue_url, AttributeNames=["QueueArn"]
+    )
+    queue_arn = attrs["Attributes"]["QueueArn"]
+
+    # Allow SNS to send messages to this queue
+    policy = json.dumps({
+        "Version": "2012-10-17",
+        "Statement": [{
+            "Sid": "AllowSNSPublish",
+            "Effect": "Allow",
+            "Principal": {"Service": "sns.amazonaws.com"},
+            "Action": "sqs:SendMessage",
+            "Resource": queue_arn,
+            "Condition": {
+                "ArnEquals": {"aws:SourceArn": SNS_TOPIC_ARN}
+            },
+        }],
+    })
+    sqs_client.set_queue_attributes(
+        QueueUrl=queue_url,
+        Attributes={"Policy": policy},
+    )
+
+    # Subscribe queue to SNS topic
+    sub_response = sns_client.subscribe(
+        TopicArn=SNS_TOPIC_ARN,
+        Protocol="sqs",
+        Endpoint=queue_arn,
+        Attributes={"RawMessageDelivery": "true"},
+    )
+    subscription_arn = sub_response["SubscriptionArn"]
+    logger.info(f"Subscribed SQS to SNS: {subscription_arn}")
+
+    yield {
+        "queue_url": queue_url,
+        "queue_arn": queue_arn,
+        "subscription_arn": subscription_arn,
+    }
+
+    # Cleanup
+    try:
+        sns_client.unsubscribe(SubscriptionArn=subscription_arn)
+        logger.info(f"Unsubscribed: {subscription_arn}")
+    except Exception as e:
+        logger.warning(f"Failed to unsubscribe: {e}")
+
+    try:
+        sqs_client.delete_queue(QueueUrl=queue_url)
+        logger.info(f"Deleted queue: {queue_url}")
+    except Exception as e:
+        logger.warning(f"Failed to delete queue: {e}")
+
+
+@pytest.mark.us_east_1
+def test_notifications_creates_eventbridge_rule_and_cleanup(
+    sm_session, training_data_uri, sqs_subscriber
+):
+    """Test end-to-end notification flow: EventBridge rule → SNS → SQS message.
+
+    Flow:
+    1. Create SFTTrainer with notifications config
+    2. Verify EventBridge rule was created with correct SNS target
+    3. Submit a serverless training job (non-blocking)
+    4. Stop the job to trigger a "Stopped" event
+    5. Poll SQS queue for the notification message
+    6. Assert message contains job name and status
+    7. Clean up the EventBridge rule
+    """
+    unique_id = f"{int(time.time())}-{random.randint(1000, 9999)}"
+    job_name_prefix = f"notif-integ-{unique_id}"
+
+    bucket = sm_session.default_bucket()
+
+    sft_trainer = SFTTrainer(
+        model="amazon.nova-micro-v1",
+        training_type=TrainingType.LORA,
+        training_dataset=training_data_uri,
+        s3_output_path=f"s3://{bucket}/{DATA_PREFIX}/output/",
+        model_package_group="sdk-test-finetuned-models",
+        sagemaker_session=sm_session,
+        notifications={
+            "sns_topic_arn": SNS_TOPIC_ARN,
+            "events": ["Completed", "Failed", "Stopped"],
+            "job_name_prefix": job_name_prefix,
+        },
+        base_job_name=job_name_prefix,
+    )
+
+    # Verify notification rule ARN was set
+    assert sft_trainer.notification_rule_arn is not None, (
+        "Expected notification_rule_arn to be set after trainer construction"
+    )
+    rule_arn = sft_trainer.notification_rule_arn
+    logger.info(f"EventBridge rule created: {rule_arn}")
+
+    # Verify the rule exists via EventBridge API
+    events_client = sm_session.boto_session.client("events", region_name=REGION)
+    rule_name = rule_arn.rsplit("/", 1)[-1] if "/" in rule_arn else rule_arn.rsplit(":", 1)[-1]
+
+    # Try extracting rule name from ARN format: arn:aws:events:region:account:rule/rule-name
+    if "/rule/" in rule_arn:
+        rule_name = rule_arn.split("/rule/")[-1]
+
+    rules_response = events_client.list_rules(NamePrefix="sm-pysdk-job-notif")
+    rule_names = [r["Name"] for r in rules_response["Rules"]]
+    logger.info(f"Found rules: {rule_names}")
+
+    # Find our rule
+    matching_rules = [r for r in rules_response["Rules"] if r["Arn"] == rule_arn]
+    assert len(matching_rules) == 1, (
+        f"Expected exactly 1 rule matching ARN {rule_arn}, found {len(matching_rules)}"
+    )
+    rule = matching_rules[0]
+    assert rule["State"] == "ENABLED"
+    logger.info(f"Rule verified: {rule['Name']} (State={rule['State']})")
+
+    # Verify the rule targets our SNS topic
+    targets_response = events_client.list_targets_by_rule(Rule=rule["Name"])
+    targets = targets_response["Targets"]
+    assert len(targets) >= 1, "Expected at least 1 target on the rule"
+
+    sns_targets = [t for t in targets if t["Arn"] == SNS_TOPIC_ARN]
+    assert len(sns_targets) == 1, (
+        f"Expected SNS topic {SNS_TOPIC_ARN} as target, got: {[t['Arn'] for t in targets]}"
+    )
+    logger.info(f"Target verified: {sns_targets[0]['Arn']}")
+
+    # Submit a training job (serverless, non-blocking)
+    training_job = sft_trainer.train(wait=False)
+    assert training_job is not None
+    logger.info(f"Training job submitted: {training_job.training_job_name}")
+
+    # Wait briefly for the job to start, then stop it
+    time.sleep(30)
+    sm_client = sm_session.boto_session.client("sagemaker", region_name=REGION)
+
+    try:
+        sm_client.stop_training_job(TrainingJobName=training_job.training_job_name)
+        logger.info(f"Stop requested for: {training_job.training_job_name}")
+    except Exception as e:
+        logger.warning(f"Could not stop job (may already be terminal): {e}")
+
+    # Poll until terminal
+    max_wait = 300  # 5 minutes
+    start = time.time()
+    while time.time() - start < max_wait:
+        training_job.refresh()
+        status = training_job.training_job_status
+        if status in ("Completed", "Failed", "Stopped"):
+            break
+        logger.info(f"Status: {status} ({int(time.time() - start)}s)")
+        time.sleep(15)
+
+    logger.info(
+        f"Job final status: {training_job.training_job_status} "
+        f"(expected 'Stopped' or 'Failed')"
+    )
+    # The job should be Stopped (or Failed if it never started)
+    assert training_job.training_job_status in ("Stopped", "Failed"), (
+        f"Unexpected final status: {training_job.training_job_status}"
+    )
+
+    # Poll SQS queue for the notification message
+    sqs_client = sm_session.boto_session.client("sqs", region_name=REGION)
+    queue_url = sqs_subscriber["queue_url"]
+
+    notification_received = False
+    poll_start = time.time()
+    poll_timeout = 120  # 2 minutes for event propagation
+
+    while time.time() - poll_start < poll_timeout:
+        response = sqs_client.receive_message(
+            QueueUrl=queue_url,
+            MaxNumberOfMessages=10,
+            WaitTimeSeconds=10,
+        )
+
+        messages = response.get("Messages", [])
+        for msg in messages:
+            body = msg["Body"]
+            logger.info(f"Received SQS message: {body}")
+
+            # Parse the notification payload
+            try:
+                payload = json.loads(body)
+            except json.JSONDecodeError:
+                # Might be a raw string
+                payload = {"raw": body}
+
+            # The notification format from InputTransformer has Job, Status, etc.
+            job_name = payload.get("Job", "")
+            status = payload.get("Status", "")
+
+            if training_job.training_job_name in body:
+                notification_received = True
+                logger.info(
+                    f"Notification matched! Job={job_name}, Status={status}"
+                )
+
+                # Verify the message content
+                assert training_job.training_job_name == job_name or \
+                    training_job.training_job_name in body, (
+                    f"Expected job name '{training_job.training_job_name}' in message"
+                )
+                assert status in ("Stopped", "Failed") or \
+                    "Stopped" in body or "Failed" in body, (
+                    f"Expected 'Stopped' or 'Failed' status in message, got: {body}"
+                )
+                break
+
+            # Delete processed message
+            sqs_client.delete_message(
+                QueueUrl=queue_url,
+                ReceiptHandle=msg["ReceiptHandle"],
+            )
+
+        if notification_received:
+            break
+
+    assert notification_received, (
+        f"No notification received for job {training_job.training_job_name} "
+        f"within {poll_timeout}s. The EventBridge → SNS → SQS pipeline did not deliver."
+    )
+    logger.info("End-to-end notification delivery verified!")
+
+    # Clean up: delete the EventBridge rule
+    deleted_name = sft_trainer.delete_notification_rule(rule_arn=rule_arn)
+    logger.info(f"Deleted rule: {deleted_name}")
+
+    # Verify rule is gone
+    rules_after = events_client.list_rules(NamePrefix="sm-pysdk-job-notif")
+    remaining_arns = [r["Arn"] for r in rules_after["Rules"]]
+    assert rule_arn not in remaining_arns, (
+        f"Rule {rule_arn} should have been deleted but still exists"
+    )
+    logger.info("Cleanup verified: rule no longer exists")