diff --git a/docker-compose.yml b/docker-compose.yml
index 2da0a6bb6..6e2b7746a 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -99,6 +99,42 @@ services:
       CELERY_BROKER_PASSWORD: password
       DJANGO_SETTINGS_MODULE: enterprise_access.settings.devstack
 
+  consume_enterprise_groups_lifecycle:
+    image: edxops/enterprise-access-dev
+    container_name: enterprise_access.consume_enterprise_groups_lifecycle
+    volumes:
+      - .:/edx/app/enterprise-access/
+      - ../src:/edx/src
+    command: >
+      bash -c '
+        make requirements &&
+        pip install edx-event-bus-kafka &&
+        pip install openedx-events>=10.4.0 &&
+        pip install "confluent-kafka[avro,schema-registry]" &&
+        while true; do python /edx/app/enterprise-access/manage.py consume_events -t enterprise-groups-lifecycle -g enterprise_access_dev; sleep 2; done
+      '
+    ports:
+      - "18273:18273"
+    depends_on:
+      - mysql80
+      - memcache
+    networks:
+      - devstack_default
+    stdin_open: true
+    tty: true
+    environment:
+      CELERY_ALWAYS_EAGER: 'false'
+      CELERY_BROKER_TRANSPORT: redis
+      CELERY_BROKER_HOSTNAME: edx.devstack.redis:6379
+      CELERY_BROKER_VHOST: 0
+      CELERY_BROKER_PASSWORD: password
+      EVENT_BUS_KAFKA_SCHEMA_REGISTRY_URL: 'http://edx.devstack.schema-registry:8081'
+      EVENT_BUS_KAFKA_BOOTSTRAP_SERVERS: 'edx.devstack.kafka:29092'
+      EVENT_BUS_PRODUCER: 'edx_event_bus_kafka.create_producer'
+      EVENT_BUS_CONSUMER: 'edx_event_bus_kafka.KafkaEventConsumer'
+      EVENT_BUS_TOPIC_PREFIX: 'dev'
+      DJANGO_SETTINGS_MODULE: enterprise_access.settings.devstack
+
 networks:
   devstack_default:
     external: true
diff --git a/enterprise_access/apps/subsidy_access_policy/apps.py b/enterprise_access/apps/subsidy_access_policy/apps.py
index a5ec244f3..b8bd89f53 100644
--- a/enterprise_access/apps/subsidy_access_policy/apps.py
+++ b/enterprise_access/apps/subsidy_access_policy/apps.py
@@ -4,5 +4,15 @@
 
 
 class SubsidyAccessPolicyConfig(AppConfig):
+    """
+    Initialization app for enterprise_access.apps.subsidy_access_policy.
+    Necessary so that django signals in this app are registered.
+    """
     default_auto_field = 'django.db.models.BigAutoField'
     name = 'enterprise_access.apps.subsidy_access_policy'
+
+    def ready(self):
+        super().ready()
+
+        # pylint: disable=unused-import, import-outside-toplevel
+        import enterprise_access.apps.subsidy_access_policy.signals
diff --git a/enterprise_access/apps/subsidy_access_policy/models.py b/enterprise_access/apps/subsidy_access_policy/models.py
index 217eba87e..d6134a920 100644
--- a/enterprise_access/apps/subsidy_access_policy/models.py
+++ b/enterprise_access/apps/subsidy_access_policy/models.py
@@ -1952,6 +1952,17 @@ class Meta:
         help_text='The uuid that uniquely identifies the associated group.',
     )
 
+    @classmethod
+    def cascade_delete_for_group_uuid(cls, group_uuid):
+        """
+        Delete all associations for a remote EnterpriseGroup.
+
+        Called by the domain signal `handle_enterprise_group_deleted`.
+        """
+        return cls.objects.filter(
+            enterprise_group_uuid=group_uuid
+        ).delete()
+
 
 class ForcedPolicyRedemption(TimeStampedModel):
     """
diff --git a/enterprise_access/apps/subsidy_access_policy/signals.py b/enterprise_access/apps/subsidy_access_policy/signals.py
new file mode 100644
index 000000000..61bb86421
--- /dev/null
+++ b/enterprise_access/apps/subsidy_access_policy/signals.py
@@ -0,0 +1,29 @@
+"""
+Signal handlers for subsidy_access_policy app.
+"""
+import logging
+
+from django.dispatch import receiver
+from openedx_events.enterprise.data import EnterpriseGroup
+from openedx_events.enterprise.signals import ENTERPRISE_GROUP_DELETED
+
+from enterprise_access.apps.subsidy_access_policy.models import PolicyGroupAssociation
+
+logger = logging.getLogger(__name__)
+
+
+@receiver(ENTERPRISE_GROUP_DELETED)
+def handle_enterprise_group_deleted(**kwargs):
+    """
+    OEP-49 event handler to update assignment status for reversed transaction.
+    """
+    logger.info('Received ENTERPRISE_GROUP_DELETED signal with data: %s', kwargs)
+    group = kwargs.get('enterprise_group')
+    if not group or not isinstance(group, EnterpriseGroup):
+        logger.error('ENTERPRISE_GROUP_DELETED signal missing or invalid enterprise_group: %s', kwargs)
+        raise ValueError('Missing or invalid enterprise_group in signal')
+
+    group_uuid = group.uuid
+
+    deletions = PolicyGroupAssociation.cascade_delete_for_group_uuid(group_uuid)
+    logger.info('PolicyGroupAssociation records deleted: %s', deletions)
diff --git a/enterprise_access/apps/subsidy_access_policy/tests/factories.py b/enterprise_access/apps/subsidy_access_policy/tests/factories.py
index a163ea239..6c962b8d2 100644
--- a/enterprise_access/apps/subsidy_access_policy/tests/factories.py
+++ b/enterprise_access/apps/subsidy_access_policy/tests/factories.py
@@ -79,7 +79,7 @@ class Meta:
         model = PolicyGroupAssociation
 
     enterprise_group_uuid = factory.LazyFunction(uuid4)
-    subsidy_access_policy = factory.SubFactory(SubsidyAccessPolicyFactory)
+    subsidy_access_policy = factory.SubFactory(PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory)
 
 
 class ForcedPolicyRedemptionFactory(factory.django.DjangoModelFactory):
diff --git a/enterprise_access/apps/subsidy_access_policy/tests/test_models.py b/enterprise_access/apps/subsidy_access_policy/tests/test_models.py
index 5dbc7d93f..92896205e 100644
--- a/enterprise_access/apps/subsidy_access_policy/tests/test_models.py
+++ b/enterprise_access/apps/subsidy_access_policy/tests/test_models.py
@@ -49,6 +49,7 @@
     REQUEST_CACHE_NAMESPACE,
     PerLearnerEnrollmentCreditAccessPolicy,
     PerLearnerSpendCreditAccessPolicy,
+    PolicyGroupAssociation,
     SubsidyAccessPolicy,
     SubsidyAccessPolicyLockAttemptFailed
 )
@@ -1902,3 +1903,27 @@ def test_save(self):
 
         self.assertEqual(policy.enterprise_group_uuid, self.group_uuid)
         self.assertIsNotNone(policy.subsidy_access_policy)
+
+    def test_cascade_delete_for_group_uuid_should_delete_correct_associations(self):
+        """
+        Test that deleting a group uuid will delete all associations
+        with that group uuid, and no others.
+        """
+        policy1 = PolicyGroupAssociationFactory(
+            enterprise_group_uuid=self.group_uuid,
+            subsidy_access_policy=self.access_policy,
+        )
+        policy2 = PolicyGroupAssociationFactory(
+            enterprise_group_uuid=uuid4(),
+            subsidy_access_policy=self.access_policy,
+        )
+
+        # Ensure both policies are created
+        self.assertEqual(PolicyGroupAssociation.objects.count(), 2)
+
+        # Delete the first policy group association
+        policy1.delete()
+
+        # Ensure only the second policy remains
+        self.assertEqual(PolicyGroupAssociation.objects.count(), 1)
+        self.assertEqual(PolicyGroupAssociation.objects.first(), policy2)
diff --git a/enterprise_access/apps/subsidy_access_policy/tests/test_signals.py b/enterprise_access/apps/subsidy_access_policy/tests/test_signals.py
new file mode 100644
index 000000000..efa73f133
--- /dev/null
+++ b/enterprise_access/apps/subsidy_access_policy/tests/test_signals.py
@@ -0,0 +1,132 @@
+"""
+Tests for subsidy_access_policy signals and handlers.
+"""
+import uuid
+
+from django.test import TestCase
+from openedx_events.enterprise.data import EnterpriseGroup
+from openedx_events.enterprise.signals import ENTERPRISE_GROUP_DELETED
+
+from enterprise_access.apps.subsidy_access_policy.models import PolicyGroupAssociation
+from enterprise_access.apps.subsidy_access_policy.signals import handle_enterprise_group_deleted
+from enterprise_access.apps.subsidy_access_policy.tests.factories import (
+    PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory,
+    PolicyGroupAssociationFactory
+)
+
+
+class TestEnterpriseGroupDeletedSignal(TestCase):
+    """
+    Tests for the ENTERPRISE_GROUP_DELETED signal handler.
+    """
+
+    def setUp(self):
+        """
+        Set up test data for the test cases.
+        """
+        super().setUp()
+        self.group_uuid_1 = uuid.uuid4()
+        self.group_uuid_2 = uuid.uuid4()
+        self.group_uuid_3 = uuid.uuid4()
+
+        # Create policies to associate with groups
+        self.policy_1 = PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory()
+        self.policy_2 = PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory()
+        self.policy_3 = PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory()
+
+        # Create policy-group associations
+        self.association_1 = PolicyGroupAssociationFactory.create(
+            subsidy_access_policy=self.policy_1,
+            enterprise_group_uuid=self.group_uuid_1
+        )
+        self.association_2 = PolicyGroupAssociationFactory.create(
+            subsidy_access_policy=self.policy_2,
+            enterprise_group_uuid=self.group_uuid_2
+        )
+        # Different group that shouldn't be affected
+        self.association_3 = PolicyGroupAssociationFactory.create(
+            subsidy_access_policy=self.policy_3,
+            enterprise_group_uuid=self.group_uuid_3
+        )
+
+    def test_handle_enterprise_group_deleted_direct_call(self):
+        """
+        Test that the signal handler correctly deletes associations when called directly.
+        """
+        # Set up a mock enterprise group
+        mock_enterprise_group = EnterpriseGroup(uuid=self.group_uuid_1)
+
+        # Verify that associations for group_uuid_1 exist before deletion
+        self.assertTrue(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_1).exists(),
+            "Associations for group_uuid_1 should exist before deletion"
+        )
+
+        # Call the signal handler directly
+        handle_enterprise_group_deleted(enterprise_group=mock_enterprise_group)
+
+        # Verify that associations for group_uuid_1 are deleted
+        self.assertFalse(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_1).exists(),
+            "Associations for deleted group should be removed"
+        )
+
+        # Verify that associations for other groups are not affected
+        self.assertTrue(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_2).exists(),
+            "Associations for unrelated groups should not be affected"
+        )
+
+    def test_handle_enterprise_group_deleted_via_signal(self):
+        """
+        Test that the signal handler correctly responds to the ENTERPRISE_GROUP_DELETED signal.
+        """
+        # Set up a mock enterprise group
+        mock_enterprise_group = EnterpriseGroup(uuid=self.group_uuid_2)
+
+        # Verify that associations for group_uuid_1 exist before deletion
+        self.assertTrue(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_2).exists(),
+            "Associations for group_uuid_1 should exist before deletion"
+        )
+
+        # Send the signal
+        ENTERPRISE_GROUP_DELETED.send_event(enterprise_group=mock_enterprise_group)
+
+        # Verify that associations for group_uuid_1 are deleted
+        self.assertFalse(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_2).exists(),
+            "Associations for deleted group should be removed when signal is sent"
+        )
+
+        # Verify that associations for other groups are not affected
+        self.assertTrue(
+            PolicyGroupAssociation.objects.filter(enterprise_group_uuid=self.group_uuid_3).exists(),
+            "Associations for unrelated groups should not be affected when signal is sent"
+        )
+
+    def test_handle_enterprise_group_deleted_wrong_kwargs(self):
+        """
+        Test that the signal handler gracefully handles missing UUID.
+        """
+        # Initial count of associations
+        initial_count = PolicyGroupAssociation.objects.count()
+
+        # Call the signal handler with missing kwargs
+        with self.assertRaises(ValueError) as e:
+            handle_enterprise_group_deleted()
+            # Assert ValueError is raised for missing enterprise_group:
+        self.assertEqual(str(e.exception), 'Missing or invalid enterprise_group in signal')
+
+        # Call the signal handler with an invalid enterprise_group
+        with self.assertRaises(ValueError) as e:
+            handle_enterprise_group_deleted(enterprise_group="invalid_group")
+            # Assert ValueError is raised for invalid enterprise_group:
+        self.assertEqual(str(e.exception), 'Missing or invalid enterprise_group in signal')
+
+        # Verify no associations were deleted
+        self.assertEqual(
+            PolicyGroupAssociation.objects.count(),
+            initial_count,
+            "No associations should be deleted when signal is called with wrong kwargs"
+        )
diff --git a/enterprise_access/settings/base.py b/enterprise_access/settings/base.py
index 457d7c289..1ebc70430 100644
--- a/enterprise_access/settings/base.py
+++ b/enterprise_access/settings/base.py
@@ -466,9 +466,11 @@ def root(*path_fragments):
 LICENSE_REQUEST_TOPIC_NAME = "license-request"
 ACCESS_POLICY_TOPIC_NAME = "access-policy"
 SUBSIDY_REDEMPTION_TOPIC_NAME = "subsidy-redemption"
+ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME = "enterprise-groups-lifecycle"
 KAFKA_TOPICS = [
     COUPON_CODE_REQUEST_TOPIC_NAME,
     LICENSE_REQUEST_TOPIC_NAME,
+    ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME,
 
     # Access policy events
     ACCESS_POLICY_TOPIC_NAME,
diff --git a/enterprise_access/settings/devstack.py b/enterprise_access/settings/devstack.py
index 70e13441c..778912d86 100644
--- a/enterprise_access/settings/devstack.py
+++ b/enterprise_access/settings/devstack.py
@@ -132,9 +132,11 @@
 LICENSE_REQUEST_TOPIC_NAME = "license-request-dev"
 ACCESS_POLICY_TOPIC_NAME = "access-policy-dev"
 SUBSIDY_REDEMPTION_TOPIC_NAME = "subsidy-redemption-dev"
+ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME = "enterprise-groups-lifecycle-dev"
 KAFKA_TOPICS = [
     COUPON_CODE_REQUEST_TOPIC_NAME,
     LICENSE_REQUEST_TOPIC_NAME,
+    ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME,
 
     # Access policy events
     ACCESS_POLICY_TOPIC_NAME,
diff --git a/enterprise_access/settings/test.py b/enterprise_access/settings/test.py
index 060759580..2bffe0f0e 100644
--- a/enterprise_access/settings/test.py
+++ b/enterprise_access/settings/test.py
@@ -66,9 +66,11 @@
 LICENSE_REQUEST_TOPIC_NAME = "license-request-test"
 ACCESS_POLICY_TOPIC_NAME = "access-policy-test"
 SUBSIDY_REDEMPTION_TOPIC_NAME = "subsidy-redemption-test"
+ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME = "enterprise-groups-lifecycle-test"
 KAFKA_TOPICS = [
     COUPON_CODE_REQUEST_TOPIC_NAME,
     LICENSE_REQUEST_TOPIC_NAME,
+    ENTERPRISE_GROUPS_LIFECYCLE_TOPIC_NAME,
 
     # Access policy events
     ACCESS_POLICY_TOPIC_NAME,