diff --git a/.github/workflows/check-compact-connect.yml b/.github/workflows/check-compact-connect.yml
index 7902ac5a4..0b69755da 100644
--- a/.github/workflows/check-compact-connect.yml
+++ b/.github/workflows/check-compact-connect.yml
@@ -101,9 +101,11 @@ jobs:
 
       - name: Install dev dependencies
         run: "pip install -r backend/compact-connect/requirements-dev.txt"
-
+      # Note we are currently pinning the pip version to deal with compatibility issues released with pip 25.3
+      # see https://stackoverflow.com/a/79802727 If this issues is addressed in a later version, we can remove the
+      # extra pip install command so we stop pinning the pip version
       - name: Install all Python dependencies
-        run: "cd backend/compact-connect; bin/sync_deps.sh"
+        run: "cd backend/compact-connect; pip install -U 'pip<25.3'; bin/sync_deps.sh"
 
       - name: Test backend
         run: "cd backend/compact-connect; bin/run_tests.sh -l all -no"
diff --git a/backend/common-cdk/common_constructs/stack.py b/backend/common-cdk/common_constructs/stack.py
index c38dcd994..652b0383f 100644
--- a/backend/common-cdk/common_constructs/stack.py
+++ b/backend/common-cdk/common_constructs/stack.py
@@ -97,6 +97,10 @@ def __init__(self, *args, environment_context: dict, environment_name: str, **kw
 
         self.environment_context = environment_context
         self.environment_name = environment_name
+        # We only set the API_BASE_URL common env var if the API_DOMAIN_NAME is set
+        # The API_BASE_URL is used by the feature flag client to call the flag check endpoint
+        if self.api_domain_name:
+            self.common_env_vars.update({'API_BASE_URL': f'https://{self.api_domain_name}'})
 
     @cached_property
     def hosted_zone(self) -> IHostedZone | None:
diff --git a/backend/compact-connect/docs/README.md b/backend/compact-connect/docs/README.md
index ab0fdd159..ca34259a2 100644
--- a/backend/compact-connect/docs/README.md
+++ b/backend/compact-connect/docs/README.md
@@ -31,6 +31,7 @@ Export your license data to a CSV file, formatted as follows:
    - String lengths are enforced - exceeding them will cause validation errors
    - Some fields have a set list of allowed values. For those fields, make sure to enter the value exactly, including
      spacing and capitalization
+   - SSNs must be unique within a single CSV upload file. Do not include multiple rows with the same `ssn` in one file. If duplicate SSNs are sent within the same file, the first row will be processed, but all other duplicate rows will be rejected.
 
 #### Field Descriptions
 
diff --git a/backend/compact-connect/docs/it_staff_onboarding_instructions.md b/backend/compact-connect/docs/it_staff_onboarding_instructions.md
index 7dfcf6b1c..45fe9642d 100644
--- a/backend/compact-connect/docs/it_staff_onboarding_instructions.md
+++ b/backend/compact-connect/docs/it_staff_onboarding_instructions.md
@@ -240,6 +240,8 @@ For your convenience, use of this feature is included in the [Postman Collection
 - `licenseType` must match exactly with one of the valid types for the specified compact
 - All date fields must use the `YYYY-MM-DD` format
 - The API does not accept `null` values. For optional fields with no value, omit the field or leave it empty in CSV.
+- For CSV uploads, SSNs must be unique within a single file. Do not include multiple rows with the same `ssn` in one upload. If duplicate SSNs are sent within the same file, the first row will be processed, but all other duplicate rows will be rejected.
+- For JSON uploads, SSNs must be unique within a single request payload (array). Do not include duplicate `ssn` values in the same batch. Attempting to do so will cause the entire request to be rejected.
 
 ## Common Upload Strategies: JSON vs CSV
 
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py
index 84483a716..1156b72ab 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py
@@ -1428,9 +1428,9 @@ def encumber_privilege(self, adverse_action: AdverseActionData) -> None:
 
             now = config.current_standard_datetime
             # TODO - remove the flag as part of https://github.com/csg-org/CompactConnect/issues/1136 # noqa: FIX002
-            from cc_common.feature_flag_client import is_feature_enabled
+            from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled
 
-            if is_feature_enabled('encumbrance-multi-category-flag'):
+            if is_feature_enabled(FeatureFlagEnum.ENCUMBRANCE_MULTI_CATEGORY_FLAG):
                 encumbrance_details = {
                     'clinicalPrivilegeActionCategories': adverse_action.clinicalPrivilegeActionCategories,
                     'adverseActionId': adverse_action.adverseActionId,
@@ -2674,9 +2674,9 @@ def encumber_home_jurisdiction_license_privileges(
             'Found privileges to encumber', privilege_count=len(unencumbered_privileges_associated_with_license)
         )
         # TODO - remove the flag as part of https://github.com/csg-org/CompactConnect/issues/1136 # noqa: FIX002
-        from cc_common.feature_flag_client import is_feature_enabled
+        from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled
 
-        if is_feature_enabled('encumbrance-multi-category-flag'):
+        if is_feature_enabled(FeatureFlagEnum.ENCUMBRANCE_MULTI_CATEGORY_FLAG):
             encumbrance_details = {
                 'clinicalPrivilegeActionCategories': adverse_action.clinicalPrivilegeActionCategories,
                 'licenseJurisdiction': jurisdiction,
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py
index ded200d6f..5ce44325a 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py
@@ -375,9 +375,9 @@ def construct_simplified_privilege_history_object(
             ):
                 # TODO - remove the flag as part of https://github.com/csg-org/CompactConnect/issues/1136 # noqa: FIX002
                 # as well as check for deprecated field
-                from cc_common.feature_flag_client import is_feature_enabled
+                from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled
 
-                if is_feature_enabled('encumbrance-multi-category-flag'):
+                if is_feature_enabled(FeatureFlagEnum.ENCUMBRANCE_MULTI_CATEGORY_FLAG):
                     if 'clinicalPrivilegeActionCategory' in event['encumbranceDetails']:
                         event['note'] = event['encumbranceDetails'].get('clinicalPrivilegeActionCategory')
                     else:
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py
index c2f77a878..27d3fa4d4 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py
@@ -288,7 +288,6 @@ class UpdateCategory(CCEnum):
     DEACTIVATION = 'deactivation'
     EXPIRATION = 'expiration'
     ISSUANCE = 'issuance'
-    OTHER = 'other'
     RENEWAL = 'renewal'
     ENCUMBRANCE = 'encumbrance'
     HOME_JURISDICTION_CHANGE = 'homeJurisdictionChange'
@@ -297,6 +296,9 @@ class UpdateCategory(CCEnum):
     # this is specific to privileges that are deactivated due to a state license deactivation
     LICENSE_DEACTIVATION = 'licenseDeactivation'
     EMAIL_CHANGE = 'emailChange'
+    # NOTE: this value should explicitly be used for license upload updates, not anywhere else
+    # it is referenced in the event that an invalid license upload needs to be reverted.
+    LICENSE_UPLOAD_UPDATE_OTHER = 'other'
 
 
 class ActiveInactiveStatus(CCEnum):
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_client.py b/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_client.py
index 23e3565f2..0b9bcaa21 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_client.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_client.py
@@ -11,6 +11,7 @@
 import requests
 
 from cc_common.config import config, logger
+from cc_common.feature_flag_enum import FeatureFlagEnum
 
 
 @dataclass
@@ -43,14 +44,16 @@ def to_dict(self) -> dict[str, Any]:
         return result
 
 
-def is_feature_enabled(flag_name: str, context: FeatureFlagContext | None = None, fail_default: bool = False) -> bool:
+def is_feature_enabled(
+    flag_name: FeatureFlagEnum, context: FeatureFlagContext | None = None, fail_default: bool = False
+) -> bool:
     """
     Check if a feature flag is enabled.
 
     This function calls the internal feature flag API endpoint to determine
     if a feature flag is enabled for the given context.
 
-    :param flag_name: The name of the feature flag to check
+    :param flag_name: The name of the feature flag to check.
     :param context: Optional FeatureFlagContext for feature flag evaluation
     :param fail_default: If True, return True on errors; if False, return False on errors (default: False)
     :return: True if the feature flag is enabled, False otherwise (or fail_default value on error)
@@ -76,6 +79,7 @@ def is_feature_enabled(flag_name: str, context: FeatureFlagContext | None = None
         ):
     """
     try:
+        logger.info("checking status of feature flag", flag_name=flag_name)
         api_base_url = _get_api_base_url()
         endpoint_url = f'{api_base_url}/v1/flags/{flag_name}/check'
 
@@ -103,7 +107,8 @@ def is_feature_enabled(flag_name: str, context: FeatureFlagContext | None = None
             # Invalid response format - return fail_default value
             return fail_default
 
-        return bool(response_data['enabled'])
+        logger.info('Checked flag status successfully', flag_name=flag_name, enabled=response_data['enabled'])
+        return response_data['enabled']
 
     # We catch all exceptions to prevent a feature flag issue causing the system from operating
     except Exception as e:  # noqa: BLE001
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_enum.py b/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_enum.py
new file mode 100644
index 000000000..5180b9199
--- /dev/null
+++ b/backend/compact-connect/lambdas/python/common/cc_common/feature_flag_enum.py
@@ -0,0 +1,15 @@
+from enum import StrEnum
+
+
+class FeatureFlagEnum(StrEnum):
+    """
+    Central source for all feature flags currently referenced in the python code of the project.
+    Flags should be defined here when first added, and removed when the flag
+    is no longer in use.
+    """
+
+    # flag used by internal testing
+    TEST_FLAG = 'test-flag'
+    # runtime flags
+    ENCUMBRANCE_MULTI_CATEGORY_FLAG = 'encumbrance-multi-category-flag'
+    DUPLICATE_SSN_UPLOAD_CHECK_FLAG = 'duplicate-ssn-upload-check-flag'
diff --git a/backend/compact-connect/lambdas/python/common/tests/unit/test_feature_flag_client.py b/backend/compact-connect/lambdas/python/common/tests/unit/test_feature_flag_client.py
index d244cea68..b95300b13 100644
--- a/backend/compact-connect/lambdas/python/common/tests/unit/test_feature_flag_client.py
+++ b/backend/compact-connect/lambdas/python/common/tests/unit/test_feature_flag_client.py
@@ -1,5 +1,7 @@
 from unittest.mock import MagicMock, patch
 
+from cc_common.feature_flag_enum import FeatureFlagEnum
+
 from tests import TstLambdas
 
 
@@ -13,7 +15,7 @@ def test_is_feature_enabled_returns_true_when_flag_enabled(self):
         mock_response.json.return_value = {'enabled': True}
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response) as mock_post:
-            result = is_feature_enabled('test-flag')
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG)
 
             # Verify the result
             self.assertTrue(result)
@@ -35,7 +37,7 @@ def test_is_feature_enabled_returns_false_when_flag_disabled(self):
         mock_response.json.return_value = {'enabled': False}
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag')
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG)
 
             # Verify the result
             self.assertFalse(result)
@@ -51,7 +53,7 @@ def test_is_feature_enabled_with_context(self):
         context = FeatureFlagContext(user_id='user123', custom_attributes={'licenseType': 'lpc'})
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response) as mock_post:
-            result = is_feature_enabled('test-flag', context=context)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, context=context)
 
             # Verify the result
             self.assertTrue(result)
@@ -71,7 +73,7 @@ def test_is_feature_enabled_fail_closed_on_timeout(self):
         from cc_common.feature_flag_client import is_feature_enabled
 
         with patch('cc_common.feature_flag_client.requests.post', side_effect=Exception('Timeout')):
-            result = is_feature_enabled('test-flag', fail_default=False)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=False)
 
             # Verify it fails closed (returns False)
             self.assertFalse(result)
@@ -81,7 +83,7 @@ def test_is_feature_enabled_fail_open_on_timeout(self):
         from cc_common.feature_flag_client import is_feature_enabled
 
         with patch('cc_common.feature_flag_client.requests.post', side_effect=Exception('Timeout')):
-            result = is_feature_enabled('test-flag', fail_default=True)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=True)
 
             # Verify it fails open (returns True)
             self.assertTrue(result)
@@ -95,7 +97,7 @@ def test_is_feature_enabled_fail_closed_on_http_error(self):
         mock_response.raise_for_status.side_effect = Exception('500 Server Error')
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=False)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=False)
 
             # Verify it fails closed (returns False)
             self.assertFalse(result)
@@ -109,7 +111,7 @@ def test_is_feature_enabled_fail_open_on_http_error(self):
         mock_response.raise_for_status.side_effect = Exception('500 Server Error')
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=True)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=True)
 
             # Verify it fails open (returns True)
             self.assertTrue(result)
@@ -124,7 +126,7 @@ def test_is_feature_enabled_fail_closed_on_invalid_response(self):
         mock_response.raise_for_status = MagicMock()
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=False)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=False)
 
             # Verify it fails closed (returns False)
             self.assertFalse(result)
@@ -139,7 +141,7 @@ def test_is_feature_enabled_fail_open_on_invalid_response(self):
         mock_response.raise_for_status = MagicMock()
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=True)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=True)
 
             # Verify it fails open (returns True)
             self.assertTrue(result)
@@ -154,7 +156,7 @@ def test_is_feature_enabled_fail_closed_on_json_parse_error(self):
         mock_response.raise_for_status = MagicMock()
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=False)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=False)
 
             # Verify it fails closed (returns False)
             self.assertFalse(result)
@@ -168,7 +170,7 @@ def test_is_feature_enabled_fail_open_on_json_parse_error(self):
         mock_response.json.side_effect = ValueError('Invalid JSON')
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response):
-            result = is_feature_enabled('test-flag', fail_default=True)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, fail_default=True)
 
             # Verify it fails open (returns True)
             self.assertTrue(result)
@@ -220,7 +222,7 @@ def test_is_feature_enabled_with_context_user_id_only(self):
         context = FeatureFlagContext(user_id='user789')
 
         with patch('cc_common.feature_flag_client.requests.post', return_value=mock_response) as mock_post:
-            result = is_feature_enabled('test-flag', context=context)
+            result = is_feature_enabled(FeatureFlagEnum.TEST_FLAG, context=context)
 
             # Verify the result
             self.assertTrue(result)
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/bulk_upload.py b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/bulk_upload.py
index 1b9118d25..9aff60105 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/bulk_upload.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/bulk_upload.py
@@ -13,6 +13,9 @@
 )
 from cc_common.event_batch_writer import EventBatchWriter
 from cc_common.exceptions import CCInternalException
+
+# initialize flag outside of handler so the flag is cached for the lifecycle of the lambda execution environment
+from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled  # noqa: E402
 from cc_common.utils import (
     ResponseEncoder,
     api_handler,
@@ -22,6 +25,10 @@
 from license_csv_reader import LicenseCSVReader
 from marshmallow import ValidationError
 
+duplicate_ssn_check_flag_enabled = is_feature_enabled(
+    FeatureFlagEnum.DUPLICATE_SSN_UPLOAD_CHECK_FLAG, fail_default=True
+)
+
 
 @api_handler
 @authorize_compact_jurisdiction(action='write')
@@ -139,6 +146,9 @@ def process_bulk_upload_file(
     current_batch = []
     total_processed = 0
     failed_validation_count = 0
+    # track which ssns were included in this file to detect duplicates,
+    # which are not allowed within the same file upload
+    ssns_in_file_upload = {}
 
     with EventBatchWriter(config.events_client) as event_writer:
         for i, raw_license in enumerate(reader.licenses(stream)):
@@ -148,6 +158,17 @@ def process_bulk_upload_file(
                     # dict() here, because it prevents `compact` and `jurisdiction` from being allowed in the
                     # raw_license
                     validated_license = schema.load(dict(compact=compact, jurisdiction=jurisdiction, **raw_license))
+                    # verify that this ssn has not been used previously in the same batch
+                    license_ssn = validated_license['ssn']
+                    if duplicate_ssn_check_flag_enabled:
+                        matched_ssn_index = ssns_in_file_upload.get(license_ssn)
+                        if matched_ssn_index:
+                                raise ValidationError(
+                                    message=f'Duplicate License SSN detected. SSN matches with record '
+                                            f'{matched_ssn_index}. Every record must have a unique SSN within the same '
+                                            f'file.'
+                                )
+                        ssns_in_file_upload.update({license_ssn: i + 1})
                 except TypeError as e:
                     # This will be raised, if `raw_license` includes compact and/or jurisdiction fields
                     logger.error('License contains unsupported fields', fields=list(raw_license.keys()), exc_info=e)
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py
index 2a394b04f..652a861b4 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py
@@ -103,9 +103,9 @@ def _generate_adverse_action_for_record_type(
     adverse_action.actionAgainst = adverse_action_against_record_type
     adverse_action.encumbranceType = EncumbranceType(adverse_action_request['encumbranceType'])
     # TODO - remove the flag conditions as part of https://github.com/csg-org/CompactConnect/issues/1136 # noqa: FIX002
-    from cc_common.feature_flag_client import is_feature_enabled
+    from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled
 
-    if is_feature_enabled('encumbrance-multi-category-flag'):
+    if is_feature_enabled(FeatureFlagEnum.ENCUMBRANCE_MULTI_CATEGORY_FLAG):
         if 'clinicalPrivilegeActionCategory' in adverse_action_request:
             # replicate data to both the deprecated and new fields
             adverse_action.clinicalPrivilegeActionCategory = ClinicalPrivilegeActionCategory(
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/ingest.py b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/ingest.py
index 7021d1d5e..e84ace0b0 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/ingest.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/ingest.py
@@ -297,7 +297,7 @@ def _populate_update_record(*, existing_license: dict, updated_values: dict, rem
         update_type = UpdateCategory.DEACTIVATION
         logger.info('License deactivation detected')
     if update_type is None:
-        update_type = UpdateCategory.OTHER
+        update_type = UpdateCategory.LICENSE_UPLOAD_UPDATE_OTHER
         logger.info('License update detected')
 
     now = config.current_standard_datetime
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/licenses.py b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/licenses.py
index 17c80b9d8..405a22b5e 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/handlers/licenses.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/handlers/licenses.py
@@ -10,6 +10,14 @@
 
 schema = LicensePostRequestSchema()
 
+# initialize flag outside of handler so the flag is cached for the lifecycle of the execution environment
+from cc_common.feature_flag_client import FeatureFlagEnum, is_feature_enabled  # noqa: E402
+
+# low risk flag, so we default to enabled if failure detected
+duplicate_ssn_check_flag_enabled = is_feature_enabled(
+    FeatureFlagEnum.DUPLICATE_SSN_UPLOAD_CHECK_FLAG, fail_default=True
+)
+
 
 @api_handler
 @optional_signature_auth
@@ -63,6 +71,19 @@ def post_licenses(event: dict, context: LambdaContext):  # noqa: ARG001 unused-a
                 'errors': invalid_records,
             }
         )
+    if duplicate_ssn_check_flag_enabled:
+        # verify that none of the SSNs are repeats within the same batch
+        license_ssns = [license_record['ssn'] for license_record in licenses]
+        if len(set(license_ssns)) < len(license_ssns):
+            raise CCInvalidRequestCustomResponseException(
+                response_body={
+                    'message': 'Invalid license records in request. See errors for more detail.',
+                    'errors': {
+                        'SSN': 'Same SSN detected on multiple rows. '
+                        'Every record must have a unique SSN within the same request.'
+                    },
+                }
+            )
 
     event_time = config.current_standard_datetime
 
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_bulk_upload.py b/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_bulk_upload.py
index d5a90aa09..865f912a4 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_bulk_upload.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_bulk_upload.py
@@ -1,6 +1,6 @@
 import csv
 import json
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 from botocore.exceptions import ClientError
@@ -8,6 +8,9 @@
 
 from tests.function import TstFunction
 
+mock_flag_client = MagicMock()
+mock_flag_client.return_value = True
+
 
 @mock_aws
 class TestBulkUpload(TstFunction):
@@ -40,6 +43,7 @@ def test_get_bulk_upload_url_forbidden(self):
 
 
 @mock_aws
+@patch('cc_common.feature_flag_client.is_feature_enabled', mock_flag_client)
 class TestProcessObjects(TstFunction):
     def test_uploaded_csv(self):
         from handlers.bulk_upload import parse_bulk_upload_file
@@ -267,6 +271,88 @@ def test_bulk_upload_prevents_compact_jurisdiction_overwrites(self):
 
             self.assertEqual(expected_entry, call_args)
 
+    def test_bulk_upload_prevents_repeated_ssns_within_the_same_file_upload(self):
+        """Test that duplicate SSNs within a CSV upload are detected and rejected."""
+        from handlers.bulk_upload import parse_bulk_upload_file
+
+        # Create CSV content that includes duplicate SSNs
+        # Rows that duplicate the same SSN will be considered an error and not processed
+        csv_content = (
+            'ssn,npi,licenseNumber,givenName,middleName,familyName,suffix,dateOfBirth,dateOfIssuance'
+            ',dateOfRenewal,dateOfExpiration,licenseStatus,compactEligibility,homeAddressStreet1'
+            ',homeAddressStreet2,homeAddressCity,homeAddressState,homeAddressPostalCode'
+            ',emailAddress,phoneNumber,licenseType,licenseStatusName\n'
+            '123-45-6789,1234567890,LICENSE123,John,Middle,Doe,Jr.,1990-01-01,2020-01-01,2021-01-01,2023-01-01,active,'
+            'eligible,123 Main St,Apt 1,Columbus,OH,43215,test@example.com,+15551234567,audiologist,Active\n'
+            '123-45-6789,1234567890,LICENSE456,Jane,Middle,Smith,,1995-01-01,2023-01-01,2025-01-01,2026-01-01,active,'
+            'eligible,123 Main St,Apt 1,Columbus,OH,43215,test@example.com,+15551234567,audiologist,Active'
+        )
+
+        # Upload the CSV content directly to the mock S3 bucket
+        object_key = f'aslp/oh/{uuid4().hex}'
+        self._bucket.put_object(Key=object_key, Body=csv_content)
+
+        # Simulate the s3 bucket event
+        with open('../common/tests/resources/put-event.json') as f:
+            event = json.load(f)
+
+        event['Records'][0]['s3']['bucket'] = {
+            'name': self._bucket.name,
+            'arn': f'arn:aws:s3:::{self._bucket.name}',
+            'ownerIdentity': {'principalId': 'ASDFG123'},
+        }
+        event['Records'][0]['s3']['object']['key'] = object_key
+
+        # Mock EventBatchWriter to capture put_event calls
+        with patch('handlers.bulk_upload.EventBatchWriter') as mock_event_writer_class:
+            mock_event_writer = mock_event_writer_class.return_value.__enter__.return_value
+            # Mock the failed_entry_count attribute to return 0
+            mock_event_writer.failed_entry_count = 0
+
+            # Process the file - should not raise an exception
+            parse_bulk_upload_file(event, self.mock_context)
+
+            # Verify that put_event was called for the validation error
+            mock_event_writer.put_event.assert_called_once()
+
+            # Get the call arguments to verify the event details
+            call_args = mock_event_writer.put_event.call_args[1]['Entry']
+
+            # Verify the complete event structure
+            expected_entry = {
+                'Source': f'org.compactconnect.bulk-ingest.{object_key}',
+                'DetailType': 'license.validation-error',
+                'Detail': json.dumps(
+                    {
+                        'eventTime': '1970-01-01T00:00:00+00:00',
+                        'compact': 'aslp',
+                        'jurisdiction': 'oh',
+                        'recordNumber': 2,
+                        'validData': {
+                            'licenseType': 'audiologist',
+                            'licenseStatusName': 'Active',
+                            'licenseStatus': 'active',
+                            'compactEligibility': 'eligible',
+                            'npi': '1234567890',
+                            'licenseNumber': 'LICENSE456',
+                            'givenName': 'Jane',
+                            'middleName': 'Middle',
+                            'familyName': 'Smith',
+                            'dateOfIssuance': '2023-01-01',
+                            'dateOfRenewal': '2025-01-01',
+                            'dateOfExpiration': '2026-01-01',
+                        },
+                        'errors': [
+                            'Duplicate License SSN detected. SSN matches with record 1. '
+                            'Every record must have a unique SSN within the same file.'
+                        ],
+                    }
+                ),
+                'EventBusName': 'license-data-events',
+            }
+
+            self.assertEqual(expected_entry, call_args)
+
     def test_bulk_upload_handles_bom_character(self):
         """Test that CSV files with BOM characters are handled correctly."""
         from handlers.bulk_upload import parse_bulk_upload_file
diff --git a/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_licenses.py b/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_licenses.py
index 143b0bfcc..f338eb0e6 100644
--- a/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_licenses.py
+++ b/backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_licenses.py
@@ -1,6 +1,6 @@
 import json
 from datetime import datetime
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 from common_test.sign_request import sign_request
@@ -8,8 +8,12 @@
 
 from .. import TstFunction
 
+mock_flag_client = MagicMock()
+mock_flag_client.return_value = True
+
 
 @mock_aws
+@patch('cc_common.feature_flag_client.is_feature_enabled', mock_flag_client)
 @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-11-08T23:59:59+00:00'))
 class TestLicenses(TstFunction):
     def setUp(self):
@@ -365,6 +369,36 @@ def test_post_licenses_null_field_returns_error(self):
             json.loads(resp['body']),
         )
 
+    def test_post_licenses_returns_400_if_repeated_ssns_detected(self):
+        from handlers.licenses import post_licenses
+
+        with open('../common/tests/resources/api-event.json') as f:
+            event = json.load(f)
+
+        # The user has write permission for aslp/oh
+        event['requestContext']['authorizer']['claims']['scope'] = 'openid email aslp/readGeneral oh/aslp.write'
+        event['pathParameters'] = {'compact': 'aslp', 'jurisdiction': 'oh'}
+        with open('../common/tests/resources/api/license-post.json') as f:
+            license_data = json.load(f)
+        event['body'] = json.dumps([license_data, license_data])
+
+        # Add signature authentication headers
+        event = self._create_signed_event(event)
+
+        resp = post_licenses(event, self.mock_context)
+
+        self.assertEqual(400, resp['statusCode'])
+        self.assertEqual(
+            {
+                'message': 'Invalid license records in request. See errors for more detail.',
+                'errors': {
+                    'SSN': 'Same SSN detected on multiple rows. '
+                    'Every record must have a unique SSN within the same request.',
+                },
+            },
+            json.loads(resp['body']),
+        )
+
     def test_post_licenses_strips_whitespace_from_string_fields(self):
         """Test that whitespace is stripped from all string fields in license data."""
         from handlers.licenses import post_licenses
diff --git a/backend/compact-connect/stacks/feature_flag_stack/__init__.py b/backend/compact-connect/stacks/feature_flag_stack/__init__.py
index 301d9536d..3700c5696 100644
--- a/backend/compact-connect/stacks/feature_flag_stack/__init__.py
+++ b/backend/compact-connect/stacks/feature_flag_stack/__init__.py
@@ -124,6 +124,20 @@ def __init__(
             environment_name=environment_name,
         )
 
+        self.duplicate_ssn_upload_check_flag = FeatureFlagResource(
+            self,
+            'DuplicateSsnUploadCheckFlag',
+            provider=self.provider,  # Shared provider
+            flag_name='duplicate-ssn-upload-check-flag',
+            # Low risk update, we will automatically enable for every environment
+            auto_enable_envs=[
+                FeatureFlagEnvironmentName.TEST,
+                FeatureFlagEnvironmentName.BETA,
+                FeatureFlagEnvironmentName.PROD,
+            ],
+            environment_name=environment_name,
+        )
+
     def _create_common_provider(self, environment_name: str) -> Provider:
         # Create shared Lambda function for managing all feature flags
         # This function is reused across all FeatureFlagResource instances