diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index bb18fbd21f..05d596f689 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing unreleased
 
+[6.6.8] - 2026-03-05
+---------------------
+* feat: moving retirement code to edx-enterprise
+
 [6.6.7] - 2026-03-04
 ---------------------
 * feat: expose admin list endpoint with search and pagination
diff --git a/consent/apps.py b/consent/apps.py
index 4ffe385b85..8098acd55c 100644
--- a/consent/apps.py
+++ b/consent/apps.py
@@ -13,3 +13,9 @@ class ConsentConfig(AppConfig):
 
     name = 'consent'
     verbose_name = "Enterprise Consent"
+
+    def ready(self):
+        """
+        Perform one-time initialization: connect signal handlers.
+        """
+        import consent.signals  # noqa: F401  pylint: disable=import-outside-toplevel,unused-import
diff --git a/consent/signals.py b/consent/signals.py
new file mode 100644
index 0000000000..dec9e1439f
--- /dev/null
+++ b/consent/signals.py
@@ -0,0 +1,36 @@
+"""
+Django signal handlers for the consent module.
+"""
+from logging import getLogger
+
+from consent.models import DataSharingConsent
+
+logger = getLogger(__name__)
+
+try:
+    from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_LMS_CRITICAL
+except ImportError:
+    USER_RETIRE_LMS_CRITICAL = None
+
+
+def retire_users_data_sharing_consent(sender, user, retired_username, **kwargs):  # pylint: disable=unused-argument
+    """
+    Handle USER_RETIRE_LMS_CRITICAL signal: retire DataSharingConsent username records.
+
+    Idempotent: only updates records where username hasn't already been changed to the retired value.
+    """
+    if user.username != retired_username:
+        logger.info(
+            "Updating username %s to retired username %s on DataSharingConsent records",
+            user.username,
+            retired_username,
+        )
+    DataSharingConsent.objects.filter(
+        username=user.username,
+    ).exclude(
+        username=retired_username,
+    ).update(username=retired_username)
+
+
+if USER_RETIRE_LMS_CRITICAL is not None:
+    USER_RETIRE_LMS_CRITICAL.connect(retire_users_data_sharing_consent)
diff --git a/enterprise/__init__.py b/enterprise/__init__.py
index e823a51817..d3581b2d05 100644
--- a/enterprise/__init__.py
+++ b/enterprise/__init__.py
@@ -2,4 +2,4 @@
 Your project description goes here.
 """
 
-__version__ = "6.6.7"
+__version__ = "6.6.8"
diff --git a/enterprise/signals.py b/enterprise/signals.py
index ed462576ae..472dc26edb 100644
--- a/enterprise/signals.py
+++ b/enterprise/signals.py
@@ -31,12 +31,14 @@
 
 try:
     from common.djangoapps.student.models import CourseEnrollment
+    from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_LMS_CRITICAL
     from openedx_events.learning.signals import COURSE_ENROLLMENT_CHANGED, COURSE_UNENROLLMENT_COMPLETED
 
 except ImportError:
     CourseEnrollment = None
     COURSE_ENROLLMENT_CHANGED = None
     COURSE_UNENROLLMENT_COMPLETED = None
+    USER_RETIRE_LMS_CRITICAL = None
 
 logger = getLogger(__name__)
 _UNSAVED_FILEFIELD = 'unsaved_filefield'
@@ -451,3 +453,26 @@ def generate_default_orchestration_record_display_name(sender, instance, **kwarg
 
 if COURSE_ENROLLMENT_CHANGED is not None:
     COURSE_ENROLLMENT_CHANGED.connect(course_enrollment_changed_receiver)
+
+
+def retire_user_from_pending_enterprise_customer_user(sender, user, retired_email, **kwargs):  # pylint: disable=unused-argument
+    """
+    Handle USER_RETIRE_LMS_CRITICAL signal: retire PendingEnterpriseCustomerUser email address.
+
+    Idempotent: only updates records where user_email hasn't already been changed to the retired value.
+    """
+    if user.email != retired_email:
+        logger.info(
+            "Updating PendingEnterpriseCustomerUser email record from %s to retired email %s",
+            user.email,
+            retired_email,
+        )
+    models.PendingEnterpriseCustomerUser.objects.filter(
+        user_email=user.email,
+    ).exclude(
+        user_email=retired_email,
+    ).update(user_email=retired_email)
+
+
+if USER_RETIRE_LMS_CRITICAL is not None:
+    USER_RETIRE_LMS_CRITICAL.connect(retire_user_from_pending_enterprise_customer_user)
diff --git a/integrated_channels/integrated_channel/apps.py b/integrated_channels/integrated_channel/apps.py
index f719ec6382..9947746a75 100644
--- a/integrated_channels/integrated_channel/apps.py
+++ b/integrated_channels/integrated_channel/apps.py
@@ -11,3 +11,11 @@ class IntegratedChannelConfig(AppConfig):
     """
     name = 'integrated_channels.integrated_channel'
     verbose_name = "Enterprise Integrated Channels"
+
+    # TODO: We should move these integrated-channel-specific retirement handlers to the edx-integrated-channels repo
+    # and use `channel_integrations.integrated_channel` instead of `integrated_channels.integrated_channel`
+    def ready(self):
+        """
+        Perform one-time initialization: connect signal handlers.
+        """
+        import integrated_channels.integrated_channel.signals  # noqa: F401  pylint: disable=import-outside-toplevel,unused-import
diff --git a/integrated_channels/integrated_channel/signals.py b/integrated_channels/integrated_channel/signals.py
new file mode 100644
index 0000000000..9e8f91f41d
--- /dev/null
+++ b/integrated_channels/integrated_channel/signals.py
@@ -0,0 +1,54 @@
+"""
+Django signal handlers for integrated channels user retirement.
+"""
+from logging import getLogger
+
+from enterprise.models import EnterpriseCourseEnrollment, EnterpriseCustomerUser
+from integrated_channels.degreed.models import DegreedLearnerDataTransmissionAudit
+from integrated_channels.sap_success_factors.models import SapSuccessFactorsLearnerDataTransmissionAudit
+
+logger = getLogger(__name__)
+
+try:
+    from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_LMS_CRITICAL
+except ImportError:
+    USER_RETIRE_LMS_CRITICAL = None
+
+
+def retire_sapsf_data_transmission(sender, user, **kwargs):  # pylint: disable=unused-argument
+    """
+    Handle USER_RETIRE_LMS_CRITICAL signal: clear sapsf_user_id on audit records.
+
+    Idempotent: only updates records where sapsf_user_id is not already empty.
+    """
+    logger.info(f'Retiring sapsf_user_id {user.id} on audit records')
+    for ent_user in EnterpriseCustomerUser.objects.filter(user_id=user.id):
+        for enrollment in EnterpriseCourseEnrollment.objects.filter(enterprise_customer_user=ent_user):
+            logger.info(f'Updating enrollment {enrollment.id} to clear sapsf_user_id {user.id}')
+            SapSuccessFactorsLearnerDataTransmissionAudit.objects.filter(
+                enterprise_course_enrollment_id=enrollment.id,
+            ).exclude(
+                sapsf_user_id="",
+            ).update(sapsf_user_id="")
+
+
+def retire_degreed_data_transmission(sender, user, **kwargs):  # pylint: disable=unused-argument
+    """
+    Handle USER_RETIRE_LMS_CRITICAL signal: clear degreed_user_email on audit records.
+
+    Idempotent: only updates records where degreed_user_email is not already empty.
+    """
+    logger.info(f'Retiring degreed_user_email on audit records for user {user.id}')
+    for ent_user in EnterpriseCustomerUser.objects.filter(user_id=user.id):
+        for enrollment in EnterpriseCourseEnrollment.objects.filter(enterprise_customer_user=ent_user):
+            logger.info(f'Updating enrollment {enrollment.id} to clear degreed_user_email for ECU {ent_user}')
+            DegreedLearnerDataTransmissionAudit.objects.filter(
+                enterprise_course_enrollment_id=enrollment.id,
+            ).exclude(
+                degreed_user_email="",
+            ).update(degreed_user_email="")
+
+
+if USER_RETIRE_LMS_CRITICAL is not None:
+    USER_RETIRE_LMS_CRITICAL.connect(retire_sapsf_data_transmission)
+    USER_RETIRE_LMS_CRITICAL.connect(retire_degreed_data_transmission)
diff --git a/requirements/edx-platform-constraints.txt b/requirements/edx-platform-constraints.txt
index 74a902c6d1..7a4e99772e 100644
--- a/requirements/edx-platform-constraints.txt
+++ b/requirements/edx-platform-constraints.txt
@@ -461,7 +461,7 @@ edx-drf-extensions==10.6.0
     #   enterprise-integrated-channels
     #   openedx-authz
     #   openedx-core
-edx-enterprise==6.6.5
+edx-enterprise==6.6.8
     # via
     #   -c requirements/constraints.txt
     #   -r requirements/edx/kernel.in
diff --git a/tests/test_consent/test_signals.py b/tests/test_consent/test_signals.py
new file mode 100644
index 0000000000..2314d8849e
--- /dev/null
+++ b/tests/test_consent/test_signals.py
@@ -0,0 +1,56 @@
+"""
+Tests for consent signal handlers.
+"""
+import unittest
+
+from pytest import mark
+
+from consent.models import DataSharingConsent
+from consent.signals import retire_users_data_sharing_consent
+from test_utils.factories import DataSharingConsentFactory, EnterpriseCustomerFactory, UserFactory
+
+
+@mark.django_db
+class TestRetireUsersDataSharingConsent(unittest.TestCase):
+    """
+    Tests for the retire_users_data_sharing_consent signal handler.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.retired_username = f'retired__{self.user.username}'
+        self.enterprise_customer = EnterpriseCustomerFactory()
+        self.consent = DataSharingConsentFactory(
+            username=self.user.username,
+            enterprise_customer=self.enterprise_customer,
+        )
+
+    def _send_signal(self):
+        """Helper to invoke the handler as the signal would."""
+        retire_users_data_sharing_consent(
+            sender=None,
+            user=self.user,
+            retired_username=self.retired_username,
+            retired_email=f'retired__{self.user.email}',
+        )
+
+    def test_retires_username_on_consent_records(self):
+        self._send_signal()
+
+        self.consent.refresh_from_db()
+        assert self.consent.username == self.retired_username
+
+    def test_no_consent_records_with_original_username_remain(self):
+        original_username = self.user.username
+        self._send_signal()
+
+        assert not DataSharingConsent.objects.filter(username=original_username).exists()
+
+    def test_idempotent_when_already_retired(self):
+        """Calling the handler twice does not raise and leaves data in the correct state."""
+        self._send_signal()
+        self._send_signal()
+
+        self.consent.refresh_from_db()
+        assert self.consent.username == self.retired_username
diff --git a/tests/test_enterprise/test_signals.py b/tests/test_enterprise/test_signals.py
index 615f5b7f1e..f9a526b603 100644
--- a/tests/test_enterprise/test_signals.py
+++ b/tests/test_enterprise/test_signals.py
@@ -31,6 +31,7 @@
     create_enterprise_enrollment_receiver,
     enterprise_unenrollment_receiver,
     handle_user_post_save,
+    retire_user_from_pending_enterprise_customer_user,
 )
 from integrated_channels.integrated_channel.models import OrphanedContentTransmissions
 from test_utils import EmptyCacheMixin
@@ -1174,3 +1175,46 @@ def test_update_enterprise_catalog_query(self, api_client_mock):
             include_exec_ed_2u_courses=test_query.include_exec_ed_2u_courses,
         )
         api_client_mock.return_value.refresh_catalogs.assert_called_with([enterprise_catalog_2])
+
+
+@mark.django_db
+class TestRetireUserFromPendingEnterpriseCustomerUser(unittest.TestCase):
+    """
+    Tests for the retire_user_from_pending_enterprise_customer_user signal handler.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.retired_email = f'retired__{self.user.email}'
+        self.pending_enterprise_user = PendingEnterpriseCustomerUserFactory(
+            user_email=self.user.email,
+        )
+
+    def _send_signal(self):
+        retire_user_from_pending_enterprise_customer_user(
+            sender=None,
+            user=self.user,
+            retired_username=f'retired__{self.user.username}',
+            retired_email=self.retired_email,
+        )
+
+    def test_retires_user_email_on_pending_records(self):
+        self._send_signal()
+
+        self.pending_enterprise_user.refresh_from_db()
+        assert self.pending_enterprise_user.user_email == self.retired_email
+
+    def test_no_records_with_original_email_remain(self):
+        original_email = self.user.email
+        self._send_signal()
+
+        assert not PendingEnterpriseCustomerUser.objects.filter(user_email=original_email).exists()
+
+    def test_idempotent_when_already_retired(self):
+        """Calling the handler twice does not raise and leaves data in the correct state."""
+        self._send_signal()
+        self._send_signal()
+
+        self.pending_enterprise_user.refresh_from_db()
+        assert self.pending_enterprise_user.user_email == self.retired_email
diff --git a/tests/test_integrated_channels/test_integrated_channel/test_signals.py b/tests/test_integrated_channels/test_integrated_channel/test_signals.py
new file mode 100644
index 0000000000..2c562fd1ae
--- /dev/null
+++ b/tests/test_integrated_channels/test_integrated_channel/test_signals.py
@@ -0,0 +1,106 @@
+"""
+Tests for integrated_channel retirement signal handlers.
+"""
+import unittest
+
+from pytest import mark
+
+from integrated_channels.degreed.models import DegreedLearnerDataTransmissionAudit
+from integrated_channels.integrated_channel.signals import (
+    retire_degreed_data_transmission,
+    retire_sapsf_data_transmission,
+)
+from integrated_channels.sap_success_factors.models import SapSuccessFactorsLearnerDataTransmissionAudit
+from test_utils.factories import (
+    DegreedLearnerDataTransmissionAuditFactory,
+    EnterpriseCourseEnrollmentFactory,
+    EnterpriseCustomerUserFactory,
+    SapSuccessFactorsLearnerDataTransmissionAuditFactory,
+    UserFactory,
+)
+
+
+@mark.django_db
+class TestRetireSapsfDataTransmission(unittest.TestCase):
+    """
+    Tests for the retire_sapsf_data_transmission signal handler.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.enterprise_customer_user = EnterpriseCustomerUserFactory(user_id=self.user.id)
+        self.enrollment = EnterpriseCourseEnrollmentFactory(
+            enterprise_customer_user=self.enterprise_customer_user,
+        )
+        self.audit = SapSuccessFactorsLearnerDataTransmissionAuditFactory(
+            enterprise_course_enrollment_id=self.enrollment.id,
+            sapsf_user_id='sapsf-abc123',
+        )
+
+    def _send_signal(self):
+        retire_sapsf_data_transmission(sender=None, user=self.user)
+
+    def test_clears_sapsf_user_id(self):
+        self._send_signal()
+
+        self.audit.refresh_from_db()
+        assert self.audit.sapsf_user_id == ''
+
+    def test_no_audits_with_original_sapsf_user_id_remain(self):
+        self._send_signal()
+
+        assert not SapSuccessFactorsLearnerDataTransmissionAudit.objects.filter(
+            enterprise_course_enrollment_id=self.enrollment.id,
+            sapsf_user_id='sapsf-abc123',
+        ).exists()
+
+    def test_idempotent_when_already_retired(self):
+        self._send_signal()
+        self._send_signal()
+
+        self.audit.refresh_from_db()
+        assert self.audit.sapsf_user_id == ''
+
+
+@mark.django_db
+class TestRetireDegreedDataTransmission(unittest.TestCase):
+    """
+    Tests for the retire_degreed_data_transmission signal handler.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.enterprise_customer_user = EnterpriseCustomerUserFactory(user_id=self.user.id)
+        self.enrollment = EnterpriseCourseEnrollmentFactory(
+            enterprise_customer_user=self.enterprise_customer_user,
+        )
+        self.audit = DegreedLearnerDataTransmissionAuditFactory(
+            enterprise_course_enrollment_id=self.enrollment.id,
+            degreed_user_email='original@degreed.example.com',
+        )
+
+    def _send_signal(self):
+        retire_degreed_data_transmission(sender=None, user=self.user)
+
+    def test_clears_degreed_user_email(self):
+        self._send_signal()
+
+        self.audit.refresh_from_db()
+        assert self.audit.degreed_user_email == ''
+
+    def test_no_audits_with_original_email_remain(self):
+        self._send_signal()
+
+        assert not DegreedLearnerDataTransmissionAudit.objects.filter(
+            enterprise_course_enrollment_id=self.enrollment.id,
+            degreed_user_email='original@degreed.example.com',
+        ).exists()
+
+    def test_idempotent_when_already_retired(self):
+        self._send_signal()
+        self._send_signal()
+
+        self.audit.refresh_from_db()
+        assert self.audit.degreed_user_email == ''