Add create notifications helper method - #6113
Merged
mujtaba1747 merged 8 commits intoJul 28, 2026
Merged
Conversation
| """ | ||
| policy = { | ||
| "Version": "2008-10-17", | ||
| "Id": "__default_policy_ID", |
Contributor
There was a problem hiding this comment.
This should be more descriptive like "SageMakerNotificationsTopicPolicy", so if another user on the account is looking at this policy, they would know the source of it / get an sense of what the policy does from the name
| "Id": "__default_policy_ID", | ||
| "Statement": [ | ||
| { | ||
| "Sid": "__default_statement_ID", |
Contributor
There was a problem hiding this comment.
I recommend "SNSTopicAdministration"
amazeAmazing
approved these changes
Jul 28, 2026
rsareddy0329
added a commit
that referenced
this pull request
Aug 9, 2026
* Feat: show_metrics() and stream_logs() helper functions (#6002) * Feat: Add show_metrics() and stream_logs() for monitoring training jobs * Feat: show_metrics() and stream_logs() --------- Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> * show_metrics() Enhancement: Display MLFlow metrics for OSS models (#6013) * show_metrics() Enhancement: Display MLFlow metrics for OSS models * Update unit tests * Address code comments --------- Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> * feat(serve): add opt-in model source tag-based resource reuse (#5993) * feat(serve): add opt-in model source tag-based resource reuse Add reuse_resources to ModelBuilder.build/deploy and BedrockModelBuilder.deploy. On a hit, discover an existing resource by the model-source tag and return it instead of creating a duplicate (warn, do not raise). Honored per call. - New sagemaker/serve/model_reuse.py: tag helpers + service-client discovery - Consolidate Nova manifest/checkpoint reading into sagemaker/core/training/utils.py - SageMaker: build() skips Model creation on reuse (sets built_model to the existing Model); deploy() reuses the endpoint after validating env vars/image/ instance type (PrimaryContainer with Containers[0] fallback for Nova) - Reuse gates are skipped for inference-component builds/deploys so IC create/update (via _deploy_for_ic) is never silently intercepted - Bedrock: reuse custom model + active deployment; response includes modelArn - Reuse discovery uses the cached session/bedrock clients - Support raw S3 URI model input via model_metadata BASE_MODEL_NAME - Unit tests + notebook examples * feat(serve): support Nova inference-component deployment and harden reuse Route Nova model-customization deploys through the shared single-inference -component path when a ResourceRequirements inference_config is supplied, so each Nova checkpoint (full-rank or LoRA-merged) is hosted as one inference component referencing the built Model. Nova without an inference_config keeps the direct model-on-variant path. - Broaden _is_nova_model to identify Nova from a package-less source (raw S3 checkpoint or trainer) via base_model_name, in addition to the model package recipe/hub-content name. - Set EnableNetworkIsolation on the IC endpoint config to match the built Model (always True for Nova), fixing CreateInferenceComponent rejection on mismatched network isolation. - Guard model-package-dependent logic (restricted-package path, PEFT/recipe metadata, lineage tracking) so package-less Nova checkpoints deploy cleanly. - Apply accumulated tags (including the model-source reuse tag) to endpoints created on the shared IC path so they remain discoverable. - build(reuse_resources=True) only short-circuits when the backing Model can be resolved; IC endpoints and stale/deleted configs fall through and build a real Model, preventing a None built_model on later IC deploys. - deploy() warns that reuse_resources has no effect for inference-component deployments, which manage their own reuse by component name. - Surface both the manifest.json and output.tar.gz errors when Nova checkpoint URI resolution fails, instead of masking the primary failure. Add unit tests covering the Nova IC path (routing, network isolation, IC spec) and the model-on-variant fallback. * fix(core): resolve Nova checkpoint manifest across all three output layouts Nova training jobs write their checkpoint manifest to different locations depending on the training platform: HyperPod: <output>/<job>/manifest.json Serverless: <output>/<job>/output/output/manifest.json Serverful: <output>/<job>/output/output.tar.gz (manifest inside) resolve_nova_checkpoint_uri previously only tried the serverless manifest path and the serverful tar.gz, so HyperPod jobs (manifest directly under the job directory) failed to resolve. Add build_nova_hyperpod_manifest_s3_uri and try all three layouts in turn, aggregating every failure into the raised error so the real cause is not masked by the last attempt's message. Add unit tests for the HyperPod builder and for resolution from the HyperPod and serverless layouts. * feat(serve): Model-tag reuse, IC-deploy guard, and instance_type fix Simplify reuse discovery by tagging SageMaker Models (not just endpoints) with the model-source identifier, so build(reuse_resources=True) can find and skip recreating an existing Model directly — no IC-state dependency. - Tag non-Nova Models at build time with the model-source tag (matching the Nova path's existing behavior). Both Nova and OSS Models are now discoverable by tag. - Add _find_reusable_model: build(reuse_resources=True) searches Models by source tag, skipping Model creation on a hit. Also discovers the endpoint for deploy() to reuse later. - Simplify _get_model_for_endpoint back to variant-only lookup (returns None for IC endpoints). No longer needs IC-spec resolution since the Model is found directly by tag. - _reused_endpoint_matches_config returns True for IC endpoints (can't read container config from variant; Model was already matched by tag). - deploy() with reuse_resources=True on an IC deploy logs a warning that the flag has no effect (ICs manage reuse by endpoint_name + IC name). - Fix deploy() to set self.instance_type from the caller's explicit value before calling _deploy_model_customization, preventing recipe-resolved defaults from overriding the user's intent. - Add model-source tag assertion to the existing OSS deploy integ test. * feat(train): add dry_run=True to train() (#6027) * feat(train): add dry_run=True to train() Add dry_run parameter to all trainers (SFT, DPO, RLVR, RLAIF). When dry_run=True: - All existing validation runs inline (IAM role, hyperparameters, recipe constraints, infrastructure availability) - Returns None without submitting a job or consuming compute - Raises with clear error message on validation failure Additionally, validate_data_path_exists() is called unconditionally (regardless of dry_run) before job submission to catch non-existent S3 paths or dataset ARNs early. Design follows nova-forge-sdk pattern: validation always runs as part of the normal code path, dry_run short-circuits before the actual TrainingJob.create API call. Changes: - data_utils.py: add validate_data_path_exists() utility (S3 + DataSet ARN) - base_trainer.py: add dry_run to abstract train(), _train_serverful_smtj(), and _train_hyperpod() - sft/dpo/rlvr/rlaif_trainer.py: add dry_run param, pass through to shared methods, short-circuit serverless path - Notebook examples added to SFT, DPO, RLVR, RLAIF notebooks - Unit tests added to existing test files - Integration test added * feat(evaluate): add dry_run=True to evaluate() Add dry_run parameter to BaseEvaluator.evaluate() and all subclasses (BenchMarkEvaluator, CustomScorerEvaluator, LLMAsJudgeEvaluator). When dry_run=True: - All existing validation runs (IAM role, model resolution, recipe, pipeline rendering) - Dataset S3 path / DataSet ARN validated via validate_data_path_exists() - Returns None without submitting a pipeline execution - Raises on validation failure Dataset validation runs unconditionally (not just during dry_run) for CustomScorerEvaluator and LLMAsJudgeEvaluator which accept user datasets. Changes: - base_evaluator.py: add dry_run to evaluate() signature - benchmark_evaluator.py: add dry_run, short-circuit before _start_execution() - custom_scorer_evaluator.py: add dry_run, validate dataset, short-circuit - llm_as_judge_evaluator.py: add dry_run, validate dataset, short-circuit - Notebook examples added to benchmark, custom_scorer, llm_as_judge notebooks * fix(dry_run): support DataSet objects, deduplicate ARN validation, expand coverage - validate_data_path_exists() now accepts Union[str, DataSet]; extracts .arn from DataSet objects for validation - Removed duplicate ARN validation logic; delegates to _validate_dataset_arn_exists() - _validate_dataset_arn_exists() warns on AccessDenied instead of raising (execution role may still have access) - Removed isinstance(..., str) guards in all trainers and evaluators so DataSet objects flow through validation - Added dry_run=True parameter to CPTTrainer.train() - Added dry_run=True parameter to ModelTrainer.train() - Integration test: valid_dataset fixture no longer re-creates on every run - Integration test: added nonexistent_dataset_arn and nonexistent_dataset_obj fixtures - Integration test: added TestDryRunServerful class (serverful compute path) - Unit test: added test_dataset_object_extracts_arn, test_dataset_object_not_found_raises * fix(dry_run): support all AWS partitions in DataSet ARN validation - Update ARN regex to aws(?:-[a-z]+)* to match aws-cn, aws-us-gov, aws-iso, aws-iso-b partitions - Use regex guard in validate_data_path_exists() for consistent matching - Add unit tests for each partition (standard, China, GovCloud, ISO, ISO-B) and invalid partition rejection * [Feat]: Job Notifications for SMTJ (#6042) * [WIP] Job notifications setup * [WIP] Job notif update * Adding tests, dedupe logic, and example * Update ARN example values, clean up SM session definitions, rename func * Update trainers to include 'notifications' param and add arn regex check * Address PR comments, use botocore errors, return notification rule arn --------- Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> * [Docs] Add documentation for show_metrics(), stream_logs(), and job notification setup (#6065) * docs: Add show_metrics, stream_logs, and job notifications documentation * Update example job names * Move monitoring capabilities to model_customization --------- Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> * documentation: add dry-run and resource reuse docs to existing RST pages (#6061) * feat(evaluate): add dry_run and caller IAM permission validation to a… (#6075) * feat(evaluate): add dry_run and caller IAM permission validation to all evaluators - Add dry_run=True parameter to InspectAIEvaluator and MultiTurnRLEvaluator (Benchmark, CustomScorer, LLMAsJudge already had it) - Add verify_evaluation_caller_permissions() to validate the caller's identity has the pipeline-orchestration permissions before submitting - Define EVALUATION_CALLER_ACTIONS constant with the set of 22 IAM actions needed by whoever calls evaluator.evaluate() - Fix bug: LLMAsJudge InspectAI code path was missing the dry_run check - Wire dry_run through _get_aws_execution_context() for all evaluators - Add unit tests covering dry_run behavior across all evaluator types * change: move EVALUATION_CALLER_ACTIONS to iam_policies.py * fix: SMHP RLVR image selection and storm_rbs recipe cleanup (#6079) * [Fix] Remove task-type from RLVR recipe, update RLVR image selection logic * Add check for RLVR/RFT before removing task_type * fix: prefer SMHP image over SMTJ fallback in _train_hyperpod In _train_hyperpod, try get_hyperpod_training_image first (native SMHP image) and only fall back to SMTJ image with SM-TJ->SM-HP tag replacement if the SMHP image is not available. Previously the order was inverted. --------- Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> * fix: stream_logs_smhp extract training job from obj (#6084) Co-authored-by: Syed Jafri <syedjfr@amazon.com> * Fix mlflow (oss models) metrics viz (#6102) * fix: stream_logs_smhp extract training job from obj * fix: render mlflow metrics as png to handle large number of metrics * code cleanup: move io, base64 to top level imports --------- Co-authored-by: Syed Jafri <syedjfr@amazon.com> * fix(dry_run): skip MLflow app creation during dry_run (#6100) When dry_run=True, _resolve_mlflow_resource_arn now lists existing apps but skips creation and waiting. All trainers (SFT, DPO, RLVR, RLAIF, MultiTurnRL) forward dry_run to MLflow resolution. 9 unit tests. * Update error message on ModelBuilder when deploying from S3 checkpoint (#6111) * Update error message on ModelBuilder when deploying from S3 checkpoint * Update ModelBuilder to automatically find image_uri * Update import and methods * Add create notifications helper method (#6113) * fix: stream_logs_smhp extract training job from obj * fix: render mlflow metrics as png to handle large number of metrics * code cleanup: move io, base64 to top level imports * fix: add helper method to create sns topic * fix: renamed IDs for readability in SNS access policy --------- Co-authored-by: Syed Jafri <syedjfr@amazon.com> * fix(serve): BedrockModelBuilder accepts BaseTrainer as model input (#6104) BedrockModelBuilder(model=trainer) now works for SFTTrainer, DPOTrainer, RLVRTrainer, etc. Previously only ModelTrainer and S3 URIs were supported. * fix(serve): fix two P0 resource-reuse bugs (instance-type reuse + Bedrock permission fail-open) (#6109) Two independent reuse defects surfaced during Nova SDK dogfooding, both under reuse_resources=True. Fixed together as they share the reuse code paths. --- 1. instance_type mismatch silently reused --- deploy() trusted the endpoint candidate cached by build() in _reused_endpoint_name without re-validating it. That cache is resolved at build time, before instance_type (a deploy() argument) is known, so a deploy on a different instance type reused the wrong endpoint. Separately, _reused_endpoint_matches_config skipped the instance_type check entirely for Inference Component (IC) endpoints: the IC early-return (no ModelName on the variant) returned True before reaching the check. deploy() now re-validates the cached endpoint against the requested instance_type and falls back to a fresh discovery on a miss or mismatch. _reused_endpoint_matches_config checks instance_type -- which lives on the production variant and is available for every endpoint, including IC ones -- before the IC early-return, so an instance-type mismatch is never silently reused. --- 2. Bedrock reuse_resources=True fails open on missing permission --- Reuse discovery (find_existing_bedrock_model, find_active_bedrock_deployment_for_model, find_existing_sagemaker_endpoint) swallowed every exception and returned None. When the execution role lacked a read permission (e.g. bedrock:ListTagsForResource), a denied discovery call was indistinguishable from "nothing to reuse", so reuse fell through to creating the resource -- which then failed with a confusing "ValidationException: Model with name '...' already exists" because the prior run's resource still held the deterministic name, never surfacing the real cause. A new _reraise_if_access_denied helper re-raises AccessDeniedException as a PermissionError naming the missing IAM action; all other errors still fail open (warn and return None) so transient failures like throttling do not block a deploy. Tests: 86 unit tests pass across test_model_builder.py and test_model_reuse.py, including regression guards for both fixes (IC instance-type mismatch, build-time cache re-validation, access-denied-raises, and fail-open-on-other-error). Co-authored-by: Elise Harvey <harveel@amazon.com> * fix(train): raise on expired credentials in show_metrics log fetch (#6114) * fix(train): raise on expired credentials in show_metrics log fetch When AWS credentials expire, show_metrics() reported "No CloudWatch logs found for job '<name>'. The job may still be starting, or logs may not be available yet", sending users to debug their training job instead of refreshing credentials. _fetch_smtj_logs() and _fetch_smhp_logs() each wrapped their CloudWatch Logs call in a bare `except Exception`, logged a warning, and returned an empty list, so an ExpiredTokenException was indistinguishable from a job that genuinely has no logs yet. The empty list then became the misleading ValueError above. Design follows the existing notifications.py pattern: read the structured error code, re-raise the auth subset as PermissionError with remediation guidance, and let every other code keep its current degrade-to-empty behavior. A genuinely absent log group (ResourceNotFoundException) still returns an empty list, so a just-started job continues to raise the existing "No CloudWatch logs found" ValueError. Changes: - cloudwatch_metrics.py: add _AUTH_ERROR_CODES and _raise_if_auth_error(); apply at describe_log_streams, get_log_events, and filter_log_events; narrow `except Exception` to `except ClientError` - base_trainer.py: document PermissionError on show_metrics() - test_cloudwatch_metrics.py: 20 unit tests covering all 7 auth codes on both platforms, ResourceNotFoundException degradation, mid-pagination failure, and a regression guard for the existing "no logs found" path * change(train): move AUTH_ERROR_CODES to common_utils/constants.py Addresses review feedback: the auth error-code list was duplicated in cloudwatch_metrics.py and its unit test, so the two lists had to be kept in sync manually. Move it to the existing common_utils/constants.py as a module-level frozenset and import it in both places. Test parametrization sorts the frozenset so test IDs stay deterministic. --------- Co-authored-by: sayemkam <sayemkam@amazon.com> * fix: serverful instance type validations + integ tests (#6124) * fix: serverful instance type validations + integ tests * cleanup: add missing newlines to end of files * style(sagemaker-train): Fix missing trailing newlines * fix(train): Log SMHP enum fetch failure at debug, not warning + added unit tests --------- Co-authored-by: Syed Jafri <syedjfr@amazon.com> * fix: update job key used for show_metrics/stream_logs in MTRL (#6128) * Update error message on ModelBuilder when deploying from S3 checkpoint * Update ModelBuilder to automatically find image_uri * Update import and methods * Improving logging level * fix: update job key used for show_metrics/stream_logs in MTRL * Stream logs unified improvement (#6127) * fix(stream_logs): unify log streaming for MTRL, evaluators, and HyperPod Fixes three stream_logs() bugs identified in bug bash testing: 1. MTRL trainer stream_logs() now uses the correct log group (/aws/sagemaker/Job/AgentRFT) and polls status via Job API instead of TrainingJob API — previously hung forever showing nothing. 2. Adds stream_logs() to BaseEvaluator and EvaluationPipelineExecution with support for pipeline, MTRL eval, and HyperPod backends. 3. Patches _stream_logs_smhp() to provide user feedback instead of silently swallowing ResourceNotFoundException and empty events. Introduces LogStreamer utility (poll-once pattern) and stream_log_loop() shared helper to eliminate code duplication across all callers. * test(stream_logs): add evaluator integ tests using existing completed jobs Integration tests for evaluator.stream_logs() against completed pipeline executions in us-west-2/729646638167. Covers BenchMarkEvaluator, CustomScorerEvaluator, and LLMAsJudgeEvaluator. No new jobs launched. * Improve logging, error messages and minor bug fixes (#6135) * Update error message on ModelBuilder when deploying from S3 checkpoint * Update ModelBuilder to automatically find image_uri * Update import and methods * Improving logging level * fix: update job key used for show_metrics/stream_logs in MTRL * fix(serve): Speed up reuse_resources with Tagging API and resolve string training jobs Use resourcegroupstaggingapi.get_resources() for O(1) tag lookups instead of scanning all models/endpoints, and auto-resolve string _latest_training_job to TrainingJob objects in ModelBuilder so trainers work without manual .get(). * Remove region logging * fix(train): Improve trainer UX and reduce verbose logging - Cache sagemaker_session in BaseTrainer.__init__ to avoid creating duplicate sessions on every method call - Remove redundant role validation (was validating 3x per train() call), now validates once in ModelTrainer.__init__ - Demote noisy INFO logs to DEBUG (role validated, stopping condition defaults, recipe paths, output compression) - Add num_lines param to stream_logs() to limit output for long jobs - Prefix CloudWatch log lines with [CloudWatch] and use print() to distinguish container output from SDK logging - Improve show_metrics() error message when time range yields no logs - Raise ValueError on AccessDenied in dry_run data path validation instead of silently warning - Improve model_package_group error message to mention compute option - Move local imports in _train_serverful_smtj to top-level - Remove redundant get_role() call in ModelTrainer.from_recipe() * resolve conflict * Update import * fix(serve): Move endpoint reuse discovery to deploy time (#6142) * fix(serve): Move endpoint reuse discovery to deploy time * cleanup: Remove stray dev logs --------- Co-authored-by: Syed Jafri <syedjfr@amazon.com> * fix: usuability update from feedback (#6150) * fix: usuability update from feedback * update documentation * Doc update: reorganize examples folder (#6155) * fix: usuability update from feedback * update documentation * docs: organize model-customization examples into serverless/serverful/deployment/evaluation subfolders * add job notification integ test * feat: bedrock model reuse for OSS models (#6157) * feat: bedrock model reuse for OSS models * code cleanup: fixed log statements, renamed variables, added type checks --------- Co-authored-by: Syed Jafri <syedjfr@amazon.com> * test: add integ tests for show_metrics and model reuse (reuse_resources=True) (#6158) - Add show_metrics() assertions to existing Nova SFT tests (serverless + serverful) - Add show_metrics() via MLflow to OSS Llama SFT test - Add show_metrics() to MTRL trainer integration test - Add reuse round-trip tests to test_model_customization_deployment.py (OSS, us-west-2) - Add reuse round-trip tests to test_nova_model_customization_deployment.py (Nova, us-east-1) - Add build(reuse_resources=True) tests to both deployment test files * feat(telemetry): Add tracking for dry_run, notifications, reuse (#6159) * fix: usuability update from feedback * update documentation * docs: organize model-customization examples into serverless/serverful/deployment/evaluation subfolders * add job notification integ test * feat(telemetry): Add tracking for dry_run, notifications, reuse_resources, show_metrics, and stream_logs * update telemetry * update telemetry * fix: tail_lines in stream_logs returns last N events (true tail semantics) (#6164) - Add LogStreamer.poll_tail() for fetching last N events: - Stream mode (SMTJ): backward pagination via get_log_events nextBackwardToken - Filter mode (SMHP): filter_log_events with startFromHead=False, paginate until N matches collected (CW bounds pages by scan volume, not result count) - Multi-stream jobs: merge events across streams by timestamp, return globally last N - stream_log_loop: when tail_lines is set, call poll_tail() and return immediately - Refactor _stream_logs_smhp to delegate to LogStreamer + stream_log_loop, eliminating ~80 lines of duplicated inline polling logic - Validate start_time >= 2024-01-01 in _tail_filter_mode (CW API restriction) - Remove dead code: lines_printed counter no longer needed in SMHP forward loop - Add unit tests for poll_tail (stream mode, filter mode, multi-stream merge, multi-page pagination, pre-2024 validation) - Add unit tests for stream_log_loop tail_lines integration * Remove duplicated dry_run return * fix: unit tests * fix: integ test (dryrun, model reuse) * fix: model reuse for Nova model * fix: sagemaker-train test fix, remove extra dryrun tests * fix: 'ModelPackage' is not defined * fix: gpu integ test fixes * fix: key check before logging training job name (#6168) Co-authored-by: Syed Jafri <syedjfr@amazon.com> * Fix tests in trainers (#6169) Co-authored-by: Roja Reddy Sareddy <rsareddy@amazon.com> * Master nova follow ups (#6170) * Fix tests in trainers * fix(iam): Remove iam:PassRole from caller validation to avoid false denials SimulatePrincipalPolicy without iam:PassedToService context value returns implicitDeny for condition-scoped PassRole policies like AmazonSageMakerFullAccess, causing false test failures. --------- Co-authored-by: Roja Reddy Sareddy <rsareddy@amazon.com> --------- Co-authored-by: Ealynn Hsu <89547630+ehsu3@users.noreply.github.com> Co-authored-by: Ealynn Hsu <ealynnh@amazon.com> Co-authored-by: LN <133025223+amazeAmazing@users.noreply.github.com> Co-authored-by: Syed Jafri <syedjfr@amazon.com> Co-authored-by: Zhaoqi <jzhaoqwa@amazon.com> Co-authored-by: eliseharvey <108292155+eliseharvey@users.noreply.github.com> Co-authored-by: Elise Harvey <harveel@amazon.com> Co-authored-by: Sayem Kamal <sayemkamal12@gmail.com> Co-authored-by: sayemkam <sayemkam@amazon.com> Co-authored-by: papriwal <papriwal@amazon.com> Co-authored-by: rsareddy0329 <rsareddy0329@gmail.com> Co-authored-by: Roja Reddy Sareddy <rsareddy@amazon.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Improves SNS/EventBridge notification setup in sagemaker-train and makes permission failures clearer.
File changed: notifications.py
Add create_notification_topic(topic_name, sagemaker_session): creates an SNS topic and applies an access policy that lets EventBridge in the same account publish to it. The returned ARN can be passed straight to enable_notifications, so callers no longer have to configure the topic policy by hand.
Add _build_topic_policy(topic_arn, account_id) helper that assembles the policy: the default owner statement (standard SNS actions scoped to the owning account via AWS:SourceAccount) plus an explicit statement granting the EventBridge service principal (events.amazonaws.com) SNS:Publish, also scoped by AWS:SourceAccount.
Tighten error handling in enable_notifications: replace the broad except Exception (which wrapped every failure in a generic ValueError) with a targeted except ClientError. On AccessDenied/AccessDeniedException it raises a concise PermissionError pointing at the missing events:PutTargets permission; all other errors propagate unchanged.
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.