diff --git a/lms/djangoapps/grades/models.py b/lms/djangoapps/grades/models.py
index 50cfa2819ca4..6b0d753bdd18 100644
--- a/lms/djangoapps/grades/models.py
+++ b/lms/djangoapps/grades/models.py
@@ -21,12 +21,21 @@
 from model_utils.models import TimeStampedModel
 from opaque_keys.edx.django.models import CourseKeyField, UsageKeyField
 from opaque_keys.edx.keys import CourseKey, UsageKey
-from openedx_events.learning.data import CourseData, PersistentCourseGradeData
-from openedx_events.learning.signals import PERSISTENT_GRADE_SUMMARY_CHANGED
+from openedx_events.learning.data import (
+    CourseData,
+    PersistentCourseGradeData,
+    PersistentSubsectionGradeData,
+    XBlockWithScoringData,
+)
+from openedx_events.learning.signals import (
+    PERSISTENT_GRADE_SUMMARY_CHANGED,
+    PERSISTENT_SUBSECTION_GRADE_CHANGED,
+)
 from simple_history.models import HistoricalRecords
 
 from lms.djangoapps.courseware.fields import UnsignedBigIntAutoField
 from lms.djangoapps.grades import events  # pylint: disable=unused-import
+from lms.djangoapps.grades.course_data import CourseData as GradesCourseData
 from lms.djangoapps.grades.signals.signals import (
     COURSE_GRADE_PASSED_FIRST_TIME,
     COURSE_GRADE_PASSED_UPDATE_IN_LEARNER_PATHWAY,
@@ -452,12 +461,15 @@ def bulk_read_grades(cls, user_id, course_key):
             )
 
     @classmethod
-    def update_or_create_grade(cls, **params):
+    def update_or_create_grade(cls, grading_policy_hash=None, **params):
         """
         Wrapper for objects.update_or_create.
         """
         cls._prepare_params(params)
-        VisibleBlocks.cached_get_or_create(params['user_id'], params['visible_blocks'])
+        # Capture the BlockRecordList before it's removed from params below, so the
+        # event can be emitted without extra queries on grade.visible_blocks
+        visible_blocks = params['visible_blocks']
+        VisibleBlocks.cached_get_or_create(params['user_id'], visible_blocks)
         cls._prepare_params_visible_blocks_id(params)
 
         # TODO: do we NEED to pop these?
@@ -478,10 +490,11 @@ def update_or_create_grade(cls, **params):
             grade.save()
 
         cls._emit_grade_calculated_event(grade)
+        cls._emit_openedx_persistent_subsection_grade_changed_event(grade, visible_blocks, grading_policy_hash)
         return grade
 
     @classmethod
-    def bulk_create_grades(cls, grade_params_iter, user_id, course_key):
+    def bulk_create_grades(cls, grade_params_iter, user_id, course_key, grading_policy_hash=None):
         """
         Bulk creation of grades.
         """
@@ -491,15 +504,17 @@ def bulk_create_grades(cls, grade_params_iter, user_id, course_key):
         PersistentSubsectionGradeOverride.prefetch(user_id, course_key)
 
         list(map(cls._prepare_params, grade_params_iter))
-        VisibleBlocks.bulk_get_or_create(
-            user_id, course_key, [params['visible_blocks'] for params in grade_params_iter]
-        )
+        # Capture the BlockRecordLists (in order) before they're removed from params below,
+        # so events can be emitted without extra queries on grade.visible_blocks
+        visible_blocks_list = [params['visible_blocks'] for params in grade_params_iter]
+        VisibleBlocks.bulk_get_or_create(user_id, course_key, visible_blocks_list)
         list(map(cls._prepare_params_visible_blocks_id, grade_params_iter))
 
         grades = [PersistentSubsectionGrade(**params) for params in grade_params_iter]
         grades = cls.objects.bulk_create(grades)
-        for grade in grades:
+        for grade, visible_blocks in zip(grades, visible_blocks_list, strict=True):
             cls._emit_grade_calculated_event(grade)
+            cls._emit_openedx_persistent_subsection_grade_changed_event(grade, visible_blocks, grading_policy_hash)
         return grades
 
     @classmethod
@@ -529,6 +544,58 @@ def _prepare_params_visible_blocks_id(cls, params):
     def _emit_grade_calculated_event(grade):
         events.subsection_grade_calculated(grade)
 
+    @staticmethod
+    def _emit_openedx_persistent_subsection_grade_changed_event(grade, visible_blocks, grading_policy_hash=None):
+        """
+        When called emits an event when a persistent subsection grade is created or updated.
+
+        ``grading_policy_hash`` should be supplied by callers that already have the course's
+        block structure loaded, so it can be reused instead of loading the course from the
+        modulestore (an extra query per event).  When not provided, it is computed here.
+        """
+        # .. event_implemented_name: PERSISTENT_SUBSECTION_GRADE_CHANGED
+        # .. event_type: org.openedx.learning.course.persistent_subsection_grade.changed.v1
+        if grading_policy_hash is None:
+            try:
+                grading_policy_hash = GradesCourseData(user=None, course_key=grade.course_id).grading_policy_hash
+            except Exception:  # pylint: disable=broad-except
+                grading_policy_hash = ""
+                log.debug(
+                    "Unable to compute grading_policy_hash for course %s",
+                    grade.course_id,
+                    exc_info=True,
+                )
+
+        xblocks_scoring_data = [
+            XBlockWithScoringData(
+                usage_key=block.locator,
+                block_type=block.locator.block_type,
+                graded=block.graded,
+                raw_possible=block.raw_possible,
+                weight=block.weight,
+            )
+            for block in visible_blocks
+        ]
+
+        PERSISTENT_SUBSECTION_GRADE_CHANGED.send_event(
+            grade=PersistentSubsectionGradeData(
+                user_id=grade.user_id,
+                course=CourseData(
+                    course_key=grade.course_id,
+                ),
+                subsection_edited_timestamp=grade.subtree_edited_timestamp,
+                grading_policy_hash=grading_policy_hash,
+                usage_key=grade.usage_key,
+                weighted_graded_earned=grade.earned_graded,
+                weighted_graded_possible=grade.possible_graded,
+                weighted_total_earned=grade.earned_all,
+                weighted_total_possible=grade.possible_all,
+                first_attempted=grade.first_attempted,
+                visible_blocks=xblocks_scoring_data,
+                visible_blocks_hash=str(grade.visible_blocks_id),
+            )
+        )
+
     @classmethod
     def _cache_key(cls, course_id):
         return f"subsection_grades_cache.{course_id}"
diff --git a/lms/djangoapps/grades/subsection_grade.py b/lms/djangoapps/grades/subsection_grade.py
index 5fab4fb5f116..fff912889d57 100644
--- a/lms/djangoapps/grades/subsection_grade.py
+++ b/lms/djangoapps/grades/subsection_grade.py
@@ -281,12 +281,17 @@ def __init__(self, subsection, course_structure, submissions_scores, csm_scores)
 
         super().__init__(subsection, all_total, graded_total)
 
-    def update_or_create_model(self, student, score_deleted=False, force_update_subsections=False):
+    def update_or_create_model(
+        self, student, score_deleted=False, force_update_subsections=False, grading_policy_hash=None
+    ):
         """
         Saves or updates the subsection grade in a persisted model.
         """
         if self._should_persist_per_attempted(score_deleted, force_update_subsections):
-            model = PersistentSubsectionGrade.update_or_create_grade(**self._persisted_model_params(student))
+            model = PersistentSubsectionGrade.update_or_create_grade(
+                grading_policy_hash=grading_policy_hash,
+                **self._persisted_model_params(student),
+            )
 
             if hasattr(model, 'override'):
                 # When we're doing an update operation, the PersistentSubsectionGrade model
@@ -299,7 +304,7 @@ def update_or_create_model(self, student, score_deleted=False, force_update_subs
             return model
 
     @classmethod
-    def bulk_create_models(cls, student, subsection_grades, course_key):
+    def bulk_create_models(cls, student, subsection_grades, course_key, grading_policy_hash=None):
         """
         Saves the subsection grade in a persisted model.
         """
@@ -309,7 +314,9 @@ def bulk_create_models(cls, student, subsection_grades, course_key):
             if subsection_grade
             if subsection_grade._should_persist_per_attempted()  # pylint: disable=protected-access
         ]
-        return PersistentSubsectionGrade.bulk_create_grades(params, student.id, course_key)
+        return PersistentSubsectionGrade.bulk_create_grades(
+            params, student.id, course_key, grading_policy_hash=grading_policy_hash
+        )
 
     def _should_persist_per_attempted(self, score_deleted=False, force_update_subsections=False):
         """
diff --git a/lms/djangoapps/grades/subsection_grade_factory.py b/lms/djangoapps/grades/subsection_grade_factory.py
index 6b9c75a7606f..67e636cfc656 100644
--- a/lms/djangoapps/grades/subsection_grade_factory.py
+++ b/lms/djangoapps/grades/subsection_grade_factory.py
@@ -57,7 +57,10 @@ def create(self, subsection, read_only=False, force_calculate=False):
                 if read_only:
                     self._unsaved_subsection_grades[subsection_grade.location] = subsection_grade
                 else:
-                    grade_model = subsection_grade.update_or_create_model(self.student)
+                    grade_model = subsection_grade.update_or_create_model(
+                        self.student,
+                        grading_policy_hash=self.course_data.grading_policy_hash,
+                    )
                     self._update_saved_subsection_grade(subsection.location, grade_model)
         return subsection_grade
 
@@ -66,7 +69,8 @@ def bulk_create_unsaved(self):
         Bulk creates all the unsaved subsection_grades to this point.
         """
         CreateSubsectionGrade.bulk_create_models(
-            self.student, list(self._unsaved_subsection_grades.values()), self.course_data.course_key
+            self.student, list(self._unsaved_subsection_grades.values()), self.course_data.course_key,
+            grading_policy_hash=self.course_data.grading_policy_hash,
         )
         self._unsaved_subsection_grades.clear()
 
@@ -100,7 +104,8 @@ def update(self, subsection, only_if_higher=None, score_deleted=False, force_upd
             grade_model = calculated_grade.update_or_create_model(
                 self.student,
                 score_deleted,
-                force_update_subsections
+                force_update_subsections,
+                grading_policy_hash=self.course_data.grading_policy_hash,
             )
             self._update_saved_subsection_grade(subsection.location, grade_model)
 
diff --git a/openedx/envs/common.py b/openedx/envs/common.py
index 31b0015d34d9..892a1001a662 100644
--- a/openedx/envs/common.py
+++ b/openedx/envs/common.py
@@ -2736,6 +2736,12 @@ def should_send_learning_badge_events(settings):
             "enabled": Derived(should_send_learning_badge_events),
         },
     },
+    "org.openedx.learning.course.persistent_subsection_grade.changed.v1": {
+        "learning-subsection-grade": {
+            "event_key_field": "grade.course.course_key",
+            "enabled": True,
+        },
+    },
 }
 
 ### event tracking
diff --git a/requirements/edx/base.txt b/requirements/edx/base.txt
index 1ea898a4b98d..7db721f3d7f6 100644
--- a/requirements/edx/base.txt
+++ b/requirements/edx/base.txt
@@ -850,7 +850,7 @@ openedx-django-require==3.0.0
     # via -r requirements/edx/kernel.in
 openedx-django-wiki==3.1.2
     # via -r requirements/edx/kernel.in
-openedx-events==11.2.0
+git+https://github.com/openedx/openedx-events.git@bmtcril/subsection-grade
     # via
     #   -r requirements/edx/kernel.in
     #   edx-enterprise
diff --git a/requirements/edx/development.txt b/requirements/edx/development.txt
index 52557cf42256..77799cf4d67b 100644
--- a/requirements/edx/development.txt
+++ b/requirements/edx/development.txt
@@ -1401,7 +1401,7 @@ openedx-django-wiki==3.1.2
     # via
     #   -r requirements/edx/doc.txt
     #   -r requirements/edx/testing.txt
-openedx-events==11.2.0
+git+https://github.com/openedx/openedx-events.git@bmtcril/subsection-grade
     # via
     #   -r requirements/edx/doc.txt
     #   -r requirements/edx/testing.txt
diff --git a/requirements/edx/doc.txt b/requirements/edx/doc.txt
index 7ebcafa83e84..d07490db6031 100644
--- a/requirements/edx/doc.txt
+++ b/requirements/edx/doc.txt
@@ -1030,7 +1030,7 @@ openedx-django-require==3.0.0
     # via -r requirements/edx/base.txt
 openedx-django-wiki==3.1.2
     # via -r requirements/edx/base.txt
-openedx-events==11.2.0
+git+https://github.com/openedx/openedx-events.git@bmtcril/subsection-grade
     # via
     #   -r requirements/edx/base.txt
     #   edx-enterprise
diff --git a/requirements/edx/kernel.in b/requirements/edx/kernel.in
index 8abecd6ea77a..ebbfa9d7f3ac 100644
--- a/requirements/edx/kernel.in
+++ b/requirements/edx/kernel.in
@@ -115,7 +115,7 @@ openedx-atlas                       # CLI tool to manage translations
 openedx-calc                        # Library supporting mathematical calculations for Open edX
 openedx-core
 openedx-django-require
-openedx-events                      # Open edX Events from Hooks Extension Framework (OEP-50)
+git+https://github.com/openedx/openedx-events.git@bmtcril/subsection-grade                     # Open edX Events from Hooks Extension Framework (OEP-50)
 openedx-filters                     # Open edX Filters from Hooks Extension Framework (OEP-50)
 openedx-forum                       # Open edX forum v2 application
 openedx-django-wiki
diff --git a/requirements/edx/testing.txt b/requirements/edx/testing.txt
index a6c1501f7a25..0ec285c582c7 100644
--- a/requirements/edx/testing.txt
+++ b/requirements/edx/testing.txt
@@ -1070,7 +1070,7 @@ openedx-django-require==3.0.0
     # via -r requirements/edx/base.txt
 openedx-django-wiki==3.1.2
     # via -r requirements/edx/base.txt
-openedx-events==11.2.0
+git+https://github.com/openedx/openedx-events.git@bmtcril/subsection-grade
     # via
     #   -r requirements/edx/base.txt
     #   edx-enterprise