AAP-89067: Fix stale image purge when EE image reference form changes (#16664)

* AAP-89067: Fix stale image purge when EE image reference form changes

The _handle_image_cleanup signal compared images by string only, so two
references to the same underlying image (e.g. tag vs digest) were treated
as different. This caused podman rmi to strip all names from a still-needed
image, breaking environments especially in disconnected installs.

Add layered guards: skip cleanup for managed images (settings and
managed=True EE objects), detect tag-vs-digest reference form changes
for the same repository, and check other EEs for cross-reference form
conflicts on delete. Genuinely different versions (v1→v2) still get
cleaned up.

* Reduce cognitive complexity of _handle_image_cleanup

Extract the delete-path loop into _other_ee_shares_repo_with_different_ref
to flatten nesting and bring cognitive complexity under the SonarQube
threshold of 15.

* Extend managed-image guard to match by repo when ref forms differ

_is_managed_image previously only did exact string comparison against
managed settings, so repo@sha256:... would not match repo:tag in
CONTROL_PLANE_EXECUTION_ENVIRONMENT or GLOBAL_JOB_EXECUTION_ENVIRONMENTS.
Now also compares by repository + reference form (tag vs digest).

* Add docstring to _handle_image_cleanup

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

awx/main/signals.py

@@ -593,9 +593,117 @@ def deny_orphaned_approvals(sender, instance, **kwargs):
         approval.deny()
 
 
-def _handle_image_cleanup(removed_image, pk):
+def _extract_image_repo(image_ref):
-    if (not removed_image) or ExecutionEnvironment.objects.filter(image=removed_image).exclude(pk=pk).exists():
+    """Return the repository part of an image reference, without tag or digest."""
-        return  # if other EE objects reference the tag, then do not purge it
+    if not image_ref:
+        return None
+
+    if '@' in image_ref:
+        return image_ref.split('@')[0]
+
+    if ':' in image_ref:
+        parts = image_ref.rsplit(':', 1)
+        # If the part after ':' contains '/' it's part of a path, not a tag
+        if '/' in parts[1]:
+            return image_ref
+        return parts[0]
+
+    return image_ref
+
+
+def _managed_setting_images():
+    """Collect all image references from managed EE settings."""
+    images = []
+    for ee_config in getattr(settings, 'GLOBAL_JOB_EXECUTION_ENVIRONMENTS', []):
+        image = ee_config.get('image')
+        if image:
+            images.append(image)
+    cp_image = getattr(settings, 'CONTROL_PLANE_EXECUTION_ENVIRONMENT', None)
+    if cp_image:
+        images.append(cp_image)
+    return images
+
+
+def _is_managed_image(image_ref, exclude_pk=None):
+    """Check if an image is referenced by managed EE settings or managed EE objects.
+
+    Matches both exact strings and same-repo different-reference-form (tag vs digest).
+    """
+    ref_repo = _extract_image_repo(image_ref)
+
+    for managed_image in _managed_setting_images():
+        if managed_image == image_ref:
+            logger.info("Skipping cleanup of %s - referenced in managed EE settings", image_ref)
+            return True
+        if ref_repo and _extract_image_repo(managed_image) == ref_repo and _different_ref_forms(image_ref, managed_image):
+            logger.info("Skipping cleanup of %s - same repository as managed setting image %s", image_ref, managed_image)
+            return True
+
+    qs = ExecutionEnvironment.objects.filter(managed=True, image=image_ref)
+    if exclude_pk is not None:
+        qs = qs.exclude(pk=exclude_pk)
+    if qs.exists():
+        logger.info("Skipping cleanup of %s - used by a managed EE", image_ref)
+        return True
+
+    return False
+
+
+def _has_digest(image_ref):
+    return '@' in image_ref
+
+
+def _different_ref_forms(ref_a, ref_b):
+    """True when one ref uses a digest and the other uses a tag (or bare name)."""
+    return _has_digest(ref_a) != _has_digest(ref_b)
+
+
+def _other_ee_shares_repo_with_different_ref(image_ref, repo, pk):
+    """Check if another EE uses the same repo with a different reference form (tag vs digest)."""
+    for ee in ExecutionEnvironment.objects.exclude(pk=pk).only('image', 'name'):
+        if not ee.image:
+            continue
+        if _extract_image_repo(ee.image) == repo and _different_ref_forms(image_ref, ee.image):
+            logger.info(
+                "Skipping cleanup of %s - EE '%s' uses %s from the same repository with a different reference form",
+                image_ref,
+                ee.name,
+                ee.image,
+            )
+            return True
+    return False
+
+
+def _handle_image_cleanup(removed_image, pk, new_image=None):
+    """Skip podman rmi when the image is managed, in use, or likely the same image under a different ref form."""
+    if not removed_image:
+        return
+
+    if _is_managed_image(removed_image, exclude_pk=pk):
+        return
+
+    if ExecutionEnvironment.objects.filter(image=removed_image).exclude(pk=pk).exists():
+        return
+
+    removed_repo = _extract_image_repo(removed_image)
+
+    # Update path: old and new share the same repo but differ in reference form
+    # (tag vs digest) — they likely point to the same underlying image.
+    if new_image and removed_repo:
+        new_repo = _extract_image_repo(new_image)
+        if removed_repo == new_repo and _different_ref_forms(removed_image, new_image):
+            logger.info(
+                "Skipping cleanup of %s - new image %s appears to be a different reference form for the same image",
+                removed_image,
+                new_image,
+            )
+            return
+
+    # Delete path: another EE references the same repo with a different reference
+    # form, so purging could strip names from a still-needed image.
+    if removed_repo and _other_ee_shares_repo_with_different_ref(removed_image, removed_repo, pk):
+        return
+
     handle_removed_image.delay(remove_images=[removed_image])
 
 
@@ -612,7 +720,7 @@ def remove_stale_image(sender, instance, created, **kwargs):
         return
     removed_image = instance._prior_values_store.get('image')
     if removed_image and removed_image != instance.image:
-        _handle_image_cleanup(removed_image, instance.pk)
+        _handle_image_cleanup(removed_image, instance.pk, new_image=instance.image)
 
 
 @receiver(post_save, sender=Session)

awx/main/tests/functional/models/test_execution_environment.py

@@ -44,3 +44,114 @@ def test_image_deletion_creates_delete_task(cleanup_patch):
     execution_environment.delete()
 
     cleanup_patch.delay.assert_called_once_with(remove_images=['quay.io/foo/bar'])
+
+
+@pytest.mark.django_db
+def test_managed_image_in_global_ee_settings_not_cleaned(cleanup_patch, settings):
+    """Images in GLOBAL_JOB_EXECUTION_ENVIRONMENTS should not be cleaned up"""
+    settings.GLOBAL_JOB_EXECUTION_ENVIRONMENTS = [{'name': 'Managed EE', 'image': 'quay.io/managed/ee:latest'}]
+
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/managed/ee:latest')
+    ee.image = 'quay.io/new/image:v1'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_managed_image_matching_control_plane_not_cleaned(cleanup_patch, settings):
+    """Images matching CONTROL_PLANE_EXECUTION_ENVIRONMENT should not be cleaned up"""
+    settings.CONTROL_PLANE_EXECUTION_ENVIRONMENT = 'quay.io/control/plane:latest'
+
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/control/plane:latest')
+    ee.image = 'quay.io/new/image:v1'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_managed_setting_different_ref_form_not_cleaned(cleanup_patch, settings):
+    """Images sharing the same repo as a managed setting but with different ref form should not be cleaned up"""
+    settings.CONTROL_PLANE_EXECUTION_ENVIRONMENT = 'quay.io/control/plane:latest'
+
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/control/plane@sha256:abc123')
+    ee.image = 'quay.io/new/image:v1'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_managed_setting_different_tag_still_cleaned(cleanup_patch, settings):
+    """Images with a different tag from a managed setting should still be cleaned up"""
+    settings.GLOBAL_JOB_EXECUTION_ENVIRONMENTS = [{'name': 'Managed EE', 'image': 'quay.io/managed/ee:latest'}]
+    settings.CONTROL_PLANE_EXECUTION_ENVIRONMENT = ''
+
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/managed/ee:v1')
+    ee.image = 'quay.io/managed/ee:v2'
+    ee.save()
+
+    cleanup_patch.delay.assert_called_once_with(remove_images=['quay.io/managed/ee:v1'])
+
+
+@pytest.mark.django_db
+def test_managed_ee_object_not_cleaned(cleanup_patch):
+    """Images belonging to a managed=True EE object should not be cleaned up"""
+    ExecutionEnvironment.objects.create(name='managed-ee', image='quay.io/foo/bar:latest', managed=True)
+
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar:latest')
+    ee.image = 'quay.io/new/image:v1'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_tag_to_digest_same_repo_not_cleaned(cleanup_patch):
+    """When updating from tag to digest for the same repo, skip cleanup"""
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar:latest')
+    ee.image = 'quay.io/foo/bar@sha256:abc123def456'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_digest_to_tag_same_repo_not_cleaned(cleanup_patch):
+    """When updating from digest to tag for the same repo, skip cleanup"""
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar@sha256:abc123def456')
+    ee.image = 'quay.io/foo/bar:latest'
+    ee.save()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_different_tags_same_repo_is_cleaned(cleanup_patch):
+    """Different tags for the same repo are genuinely different images and should be cleaned up"""
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar:v1')
+    ee.image = 'quay.io/foo/bar:v2'
+    ee.save()
+
+    cleanup_patch.delay.assert_called_once_with(remove_images=['quay.io/foo/bar:v1'])
+
+
+@pytest.mark.django_db
+def test_delete_ee_with_other_ee_sharing_repo_different_ref(cleanup_patch):
+    """On delete, skip cleanup if another EE uses the same repo with a different ref form"""
+    ExecutionEnvironment.objects.create(name='other-ee', image='quay.io/foo/bar@sha256:abc123')
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar:latest')
+    ee.delete()
+
+    cleanup_patch.delay.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_delete_ee_no_shared_repo_is_cleaned(cleanup_patch):
+    """On delete, clean up when no other EE shares the repo"""
+    ExecutionEnvironment.objects.create(name='other-ee', image='quay.io/different/repo:latest')
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar:latest')
+    ee.delete()
+
+    cleanup_patch.delay.assert_called_once_with(remove_images=['quay.io/foo/bar:latest'])

awx/main/tests/unit/test_signals.py

@@ -1,6 +1,47 @@
+import pytest
+
 from awx.main import signals
 
 
+class TestExtractImageRepo:
+    @pytest.mark.parametrize(
+        'image_ref,expected',
+        [
+            ('quay.io/ansible/awx-ee:latest', 'quay.io/ansible/awx-ee'),
+            ('registry.io/repo:v1.2.3', 'registry.io/repo'),
+            ('repo:tag', 'repo'),
+            ('quay.io/ansible/awx-ee@sha256:abc123def456', 'quay.io/ansible/awx-ee'),
+            ('registry.io/repo@sha256:xyz789', 'registry.io/repo'),
+            ('quay.io/ansible/awx-ee', 'quay.io/ansible/awx-ee'),
+            ('registry.io/repo', 'registry.io/repo'),
+            ('repo', 'repo'),
+            ('localhost:5000/repo:tag', 'localhost:5000/repo'),
+            ('registry.example.com:443/path/to/repo:latest', 'registry.example.com:443/path/to/repo'),
+            (None, None),
+            ('', None),
+        ],
+    )
+    def test_extract_image_repo(self, image_ref, expected):
+        assert signals._extract_image_repo(image_ref) == expected
+
+
+class TestDifferentRefForms:
+    def test_tag_vs_digest(self):
+        assert signals._different_ref_forms('repo:latest', 'repo@sha256:abc') is True
+
+    def test_digest_vs_tag(self):
+        assert signals._different_ref_forms('repo@sha256:abc', 'repo:latest') is True
+
+    def test_both_tags(self):
+        assert signals._different_ref_forms('repo:v1', 'repo:v2') is False
+
+    def test_both_digests(self):
+        assert signals._different_ref_forms('repo@sha256:aaa', 'repo@sha256:bbb') is False
+
+    def test_bare_vs_digest(self):
+        assert signals._different_ref_forms('repo', 'repo@sha256:abc') is True
+
+
 class TestCleanupDetachedLabels:
     def test_cleanup_detached_labels_on_deleted_parent(self, mocker):
         mock_labels = [mocker.MagicMock(), mocker.MagicMock()]