diff --git a/CHANGES/plugin_api/7272.bugfix b/CHANGES/plugin_api/7272.bugfix
new file mode 100644
index 00000000000..7ff63330d5d
--- /dev/null
+++ b/CHANGES/plugin_api/7272.bugfix
@@ -0,0 +1 @@
+Harden RepositoryVersion memoization against bugs that could lead to incorrect counts and repository content.
\ No newline at end of file
diff --git a/pulpcore/app/models/repository.py b/pulpcore/app/models/repository.py
index b70cdcd4dba..8bf3dfe1166 100644
--- a/pulpcore/app/models/repository.py
+++ b/pulpcore/app/models/repository.py
@@ -1069,6 +1069,7 @@ def add_content(self, content):
                 complete RepositoryVersion
         """
 
+        assert issubclass(content.model, Content)
         if self.complete:
             raise ResourceImmutableError(self)
 
@@ -1077,6 +1078,7 @@ def add_content(self, content):
             .exclude(pulp_domain_id=get_domain_pk())
             .exists()
         )
+
         repo_content = []
         to_add = set(content.values_list("pk", flat=True)) - set(self._get_content_ids())
         with transaction.atomic():
@@ -1115,12 +1117,14 @@ def remove_content(self, content):
                 complete RepositoryVersion
         """
 
+        assert issubclass(content.model, Content)
         if self.complete:
             raise ResourceImmutableError(self)
 
         if not content or not content.count():
             return
 
+        # check that all content is within the current domain
         assert (
             not Content.objects.filter(pk__in=content)
             .exclude(pulp_domain_id=get_domain_pk())
@@ -1158,6 +1162,7 @@ def set_content(self, content):
             pulpcore.exception.ResourceImmutableError: if set_content is called on a
                 complete RepositoryVersion
         """
+        assert issubclass(content.model, Content)
         self.remove_content(self.content.exclude(pk__in=content))
         self.add_content(content.exclude(pk__in=self.content))
 
@@ -1307,6 +1312,9 @@ def _compute_counts(self):
         objects and makes new ones with each call.
         """
         with transaction.atomic():
+            # relatively inexpensive sanity check for memoization
+            assert len(self.content_ids) == self._content_relationships().count()
+            # delete existing content details and recompute them all
             RepositoryVersionContentDetails.objects.filter(repository_version=self).delete()
             counts_list = []
             for value, name in RepositoryVersionContentDetails.COUNT_TYPE_CHOICES:
diff --git a/pulpcore/tests/unit/models/test_repository.py b/pulpcore/tests/unit/models/test_repository.py
index 9fbc137b771..3caf360b999 100644
--- a/pulpcore/tests/unit/models/test_repository.py
+++ b/pulpcore/tests/unit/models/test_repository.py
@@ -3,6 +3,7 @@
 
 from itertools import compress
 
+from pulpcore.app.models import RepositoryVersionContentDetails
 from pulpcore.plugin.models import Artifact, Content, ContentArtifact, Repository
 from pulpcore.plugin.repo_version_utils import validate_version_paths
 
@@ -59,10 +60,44 @@ def _verify_content_sets(version, current, added, removed, base_version=None):
                 verify the difference to
 
         """
+        # assert that the memoization is set
+        assert version.content_ids is not None
+        # assert that content_ids list matches the RepositoryContent representation
+        repo_content_pks = set(
+            version._content_relationships().values_list("content_id", flat=True)
+        )
+        content_ids = set(version.content_ids)
+        assert repo_content_pks == content_ids
+
         current_pks = set(version.content.values_list("pk", flat=True))
         added_pks = set(version.added(base_version).values_list("pk", flat=True))
         removed_pks = set(version.removed(base_version).values_list("pk", flat=True))
 
+        # assert that the memoized content counts (distinct from content sets) are correct
+        # NOTE: RepositoryVersionContentDetails stores counts for what was added/removed
+        # BY this version (i.e., added()/removed() with base_version=None), not relative
+        # to an arbitrary base_version. So we only verify RVCD when base_version is None.
+
+        rvcd_qs = RepositoryVersionContentDetails.objects.filter(
+            repository_version=version, content_type="core.content"
+        )
+
+        if rvcd_present := rvcd_qs.filter(
+            count_type=RepositoryVersionContentDetails.PRESENT
+        ).first():
+            assert rvcd_present.count == len(current_pks)
+
+        if base_version is None:
+            if rvcd_added := rvcd_qs.filter(
+                count_type=RepositoryVersionContentDetails.ADDED
+            ).first():
+                assert rvcd_added.count == len(added_pks)
+
+            if rvcd_removed := rvcd_qs.filter(
+                count_type=RepositoryVersionContentDetails.REMOVED
+            ).first():
+                assert rvcd_removed.count == len(removed_pks)
+
         # There must never be content shown as added & removed
         assert added_pks.intersection(removed_pks) == set()
 
@@ -73,8 +108,7 @@ def _verify_content_sets(version, current, added, removed, base_version=None):
     return _verify_content_sets
 
 
-@pytest.mark.django_db
-def test_add_and_remove_content(repository, add_content, remove_content, verify_content_sets):
+def test_add_and_remove_content(db, repository, add_content, remove_content, verify_content_sets):
     version0 = repository.latest_version()
 
     with repository.new_version() as version1:
@@ -94,7 +128,7 @@ def test_add_and_remove_content(repository, add_content, remove_content, verify_
     verify_content_sets(version3, [1, 0, 1, 1, 0], [1, 0, 0, 0, 0], [0, 0, 0, 0, 0], version2)
 
 
-def test_add_remove(repository, add_content, remove_content, verify_content_sets):
+def test_add_remove(db, repository, add_content, remove_content, verify_content_sets):
     """Verify that adding and then removing content units is handled properly."""
     version0 = repository.latest_version()
 
@@ -108,7 +142,7 @@ def test_add_remove(repository, add_content, remove_content, verify_content_sets
     assert repository.latest_version() == version0, "Empty version1 must not exist."
 
 
-def test_remove_add(repository, add_content, remove_content, verify_content_sets):
+def test_remove_add(db, repository, add_content, remove_content, verify_content_sets):
     """Verify that removing and then adding content units is handled properly."""
     with repository.new_version() as version1:
         add_content(version1, [1, 1, 1, 1, 1])
@@ -123,7 +157,9 @@ def test_remove_add(repository, add_content, remove_content, verify_content_sets
     assert repository.latest_version() == version1, "Empty version2 must not exist."
 
 
-def test_multiple_adds_and_removes(repository, add_content, remove_content, verify_content_sets):
+def test_multiple_adds_and_removes(
+    db, repository, add_content, remove_content, verify_content_sets
+):
     """Verify that adding/removing content multiple times is handled properly.
 
     Additionally, verify that other content (untouched, simple add, simple
@@ -171,7 +207,7 @@ def test_multiple_adds_and_removes(repository, add_content, remove_content, veri
     verify_content_sets(version2, [1, 1, 0, 1, 0], [0, 1, 0, 1, 0], [0, 0, 1, 0, 1])
 
 
-def test_content_batch_qs(repository, content_pks, add_content):
+def test_content_batch_qs(db, repository, content_pks, add_content):
     """Verify content iteration using content_batch_qs()."""
     sorted_pks = content_pks[:4]
     with repository.new_version() as version1:
@@ -213,11 +249,7 @@ def test_content_batch_qs_using_filter(repository, content_pks, add_content):
         pks_of_next_qs(qs_generator)
 
 
-@pytest.mark.django_db
-def test_next_version_with_one_version():
-    repository = Repository.objects.create()
-    repository.CONTENT_TYPES = [Content]
-
+def test_next_version_with_one_version(db, repository):
     assert repository.next_version == 1
     assert repository.latest_version().number == 0
     content = Content.objects.create(pulp_type="core.content")
@@ -234,21 +266,14 @@ def test_next_version_with_one_version():
     assert repository.latest_version().number == 0
 
 
-@pytest.mark.django_db
-def test_next_version_with_multiple_versions():
-    repository = Repository.objects.create()
-    repository.CONTENT_TYPES = [Content]
-
+def test_next_version_with_multiple_versions(db, repository, content_pks):
     assert repository.next_version == 1
     assert repository.latest_version().number == 0
 
-    contents = [Content(pulp_type="core.content") for _ in range(0, 3)]
-    Content.objects.bulk_create(contents)
-
     versions = [repository.latest_version()]
-    for content in contents:
+    for pk in content_pks[:3]:
         with repository.new_version() as version:
-            version.add_content(Content.objects.filter(pk=content.pk))
+            version.add_content(Content.objects.filter(pk=pk))
             versions.append(version)
 
     assert repository.next_version == 4
@@ -261,8 +286,284 @@ def test_next_version_with_multiple_versions():
     assert repository.latest_version().number == 1
 
 
-@pytest.mark.django_db
-def test_shared_artifact_same_path_validation(tmp_path):
+def test_add_existing_content(db, repository, add_content, verify_content_sets):
+    """Verify that adding content that already exists in the repo is a no-op."""
+    version0 = repository.latest_version()
+
+    # Create version1 with some content
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Try to add content that's already present (plus some new content)
+    with repository.new_version() as version2:
+        # Add content that's already in version1 (0, 1) and new content (3, 4)
+        add_content(version2, [1, 1, 0, 1, 1])
+
+    # Verify that version2 has the union of content, but only shows new content as "added"
+    verify_content_sets(version1, [1, 1, 1, 0, 0], [1, 1, 1, 0, 0], [0, 0, 0, 0, 0], version0)
+    verify_content_sets(version2, [1, 1, 1, 1, 1], [0, 0, 0, 1, 1], [0, 0, 0, 0, 0], version1)
+    verify_content_sets(version2, [1, 1, 1, 1, 1], [1, 1, 1, 1, 1], [0, 0, 0, 0, 0], version0)
+
+
+def test_remove_absent_content(db, repository, add_content, remove_content, verify_content_sets):
+    """Verify that removing content that doesn't exist in the repo is a no-op."""
+    version0 = repository.latest_version()
+
+    # Create version1 with some content
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Try to remove content that's not present (3, 4) along with content that is (0, 1)
+    with repository.new_version() as version2:
+        remove_content(version2, [1, 1, 0, 1, 1])
+
+    # Verify that only content that was actually present is shown as "removed"
+    verify_content_sets(version2, [0, 0, 1, 0, 0], [0, 0, 0, 0, 0], [1, 1, 0, 0, 0], version1)
+    verify_content_sets(version2, [0, 0, 1, 0, 0], [0, 0, 1, 0, 0], [0, 0, 0, 0, 0], version0)
+
+
+def test_empty_version_no_operations(repository):
+    """Verify that creating a version with no operations doesn't create a new version."""
+    version0 = repository.latest_version()
+
+    # Open and close a version context without doing anything
+    with repository.new_version():
+        pass
+
+    # No new version should be created
+    assert repository.latest_version() == version0, "Empty version must not be created."
+
+
+def test_version_identical_to_previous(repository, add_content, remove_content):
+    """Verify that a version identical to the previous version is not created."""
+    # Create version1 with some content
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Create a version that adds and removes content, ending up identical to version1
+    with repository.new_version() as version2:
+        add_content(version2, [0, 0, 0, 1, 1])  # Add content 3, 4
+        remove_content(version2, [0, 0, 0, 1, 1])  # Remove content 3, 4
+
+    # Version2 should not be created since it's identical to version1
+    assert repository.latest_version() == version1, "Identical version must not be created."
+
+    # Create a version that removes and then adds content, ending up identical to version1
+    with repository.new_version() as version3:
+        remove_content(version3, [1, 1, 0, 0, 0])  # Remove content 3, 4
+        add_content(version3, [1, 1, 0, 0, 0])  # Add content 3, 4
+
+    # Version3 should not be created since it's identical to version1
+    assert repository.latest_version() == version1, "Identical version must not be created."
+
+
+def test_base_version_none(db, repository, add_content, verify_content_sets):
+    """Verify that added() and removed() work correctly when base_version is None."""
+    # When base_version is None, added() should return all content, removed() should be empty
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Explicitly pass None as base_version
+    verify_content_sets(
+        version1, [1, 1, 1, 0, 0], [1, 1, 1, 0, 0], [0, 0, 0, 0, 0], base_version=None
+    )
+
+
+def test_non_sequential_version_comparison(
+    repository, add_content, remove_content, verify_content_sets
+):
+    """Verify that comparing non-sequential versions works correctly."""
+    version0 = repository.latest_version()
+
+    # Create version1 with content 0, 1, 2
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Create version2 that removes content 0
+    with repository.new_version() as version2:
+        remove_content(version2, [1, 0, 0, 0, 0])
+
+    # Create version3 that adds content 3
+    with repository.new_version() as version3:
+        add_content(version3, [0, 0, 0, 1, 0])
+
+    # Create version4 that adds content 4 and removes content 1
+    with repository.new_version() as version4:
+        add_content(version4, [0, 0, 0, 0, 1])
+        remove_content(version4, [0, 1, 0, 0, 0])
+
+    # Compare version4 to version1 (skipping version2 and version3)
+    # version1: [1, 1, 1, 0, 0]
+    # version4: [0, 0, 1, 1, 1]
+    # added: 3, 4
+    # removed: 0, 1
+    verify_content_sets(version4, [0, 0, 1, 1, 1], [0, 0, 0, 1, 1], [1, 1, 0, 0, 0], version1)
+
+    # Compare version4 to version0 (skipping all intermediate versions)
+    verify_content_sets(version4, [0, 0, 1, 1, 1], [0, 0, 1, 1, 1], [0, 0, 0, 0, 0], version0)
+
+
+def test_version_compared_to_itself(repository, add_content, verify_content_sets):
+    """Verify that comparing a version to itself shows no additions or removals."""
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Compare version1 to itself
+    verify_content_sets(version1, [1, 1, 1, 0, 0], [0, 0, 0, 0, 0], [0, 0, 0, 0, 0], version1)
+
+
+def test_add_empty_queryset(db, repository):
+    """Verify that adding an empty queryset doesn't create a version."""
+    version0 = repository.latest_version()
+
+    # Add empty queryset
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.none())
+
+    # No version should be created
+    assert repository.latest_version() == version0
+
+
+def test_add_nonexistent_content(db, repository):
+    """Verify that attempting to add non-existent content is handled correctly."""
+    # Create a content unit
+    content = Content.objects.create(pulp_type="core.content")
+
+    # Create version with existing content
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk=content.pk))
+
+    assert version1.content.count() == 1
+
+    # Try to add content using a non-existent PK (empty queryset)
+    with repository.new_version() as version2:
+        # This should not fail, just add nothing (empty queryset)
+        version2.add_content(Content.objects.filter(pk=uuid4()))
+
+    # Version2 should not be created since no content was actually added
+    assert repository.latest_version() == version1
+
+
+def test_add_wrong_content_type(db, repository):
+    """Verify that adding content of wrong type is handled."""
+    # Create content with a different pulp_type
+    wrong_type_content = Content.objects.create(pulp_type="wrong.type")
+    correct_type_content = Content.objects.create(pulp_type="core.content")
+
+    # Creating a repository version containing a type disallowed by the repository
+    # should raise an error
+    with pytest.raises(ValueError):
+        with repository.new_version() as version1:
+            version1.add_content(Content.objects.filter(pk=wrong_type_content.pk))
+            version1.add_content(Content.objects.filter(pk=correct_type_content.pk))
+
+
+def test_add_remove_non_content_type(db, repository):
+    """Verify that adding a queryset of non-Content type raises AssertionError."""
+    # Create a real Content object first
+    content = Content.objects.create(pulp_type="core.content")
+
+    # Create a ContentArtifact which is NOT a Content subclass
+    content_artifact = ContentArtifact.objects.create(content=content, relative_path="test/path")
+
+    # Attempting to add a queryset of ContentArtifact (not Content) should raise AssertionError
+    with pytest.raises(AssertionError):
+        with repository.new_version() as version1:
+            version1.add_content(ContentArtifact.objects.filter(pk=content_artifact.pk))
+
+    # Attempting to add a queryset of ContentArtifact (not Content) should raise AssertionError
+    with pytest.raises(AssertionError):
+        with repository.new_version() as version1:
+            version1.remove_content(ContentArtifact.objects.filter(pk=content_artifact.pk))
+
+
+def test_remove_empty_queryset(db, repository):
+    """Verify that removing an empty queryset doesn't create a version."""
+    content = Content.objects.create(pulp_type="core.content")
+
+    # Create version with content
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk=content.pk))
+
+    # Remove empty queryset
+    with repository.new_version() as version2:
+        version2.remove_content(Content.objects.none())
+
+    # No new version should be created
+    assert repository.latest_version() == version1
+
+
+def test_remove_nonexistent_content_from_version(db, repository):
+    """Verify that removing non-existent content doesn't cause errors."""
+    contents = [Content(pulp_type="core.content") for _ in range(3)]
+    Content.objects.bulk_create(contents)
+    pks = [c.pk for c in contents]
+
+    # Create version with some content
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk__in=pks[:2]))
+
+    # Try to remove content that was never in the repository
+    with repository.new_version() as version2:
+        # This should not fail, just remove nothing
+        version2.remove_content(Content.objects.filter(pk=pks[2]))
+
+    # Version2 should not be created since nothing changed
+    assert repository.latest_version() == version1
+
+
+def test_mixed_add_remove_with_empty_result(db, repository, content_pks):
+    """Verify that mixed operations resulting in no change don't create a version."""
+    # Create version with content
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk__in=content_pks))
+
+    # Perform operations that cancel out
+    with repository.new_version() as version2:
+        # Remove content that doesn't exist (no-op)
+        version2.remove_content(Content.objects.filter(pk=9999999))
+        # Add content that already exists (no-op)
+        version2.add_content(Content.objects.filter(pk=content_pks[0]))
+
+    # No new version should be created
+    assert repository.latest_version() == version1
+
+
+def test_operations_on_completed_version(db, repository, content_pks):
+    """Verify that operations on a completed version are not allowed."""
+    # Create and complete a version
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk=content_pks[0]))
+
+    # Version is now complete. Trying to modify it should fail
+    # The version context manager sets the version as complete on exit
+    with pytest.raises(Exception):  # Could be various exceptions depending on implementation
+        version1.add_content(Content.objects.filter(pk=content_pks[1]))
+
+
+def test_transaction_rollback_on_error(db, repository):
+    """Verify that transaction rollback works correctly when version creation fails."""
+    content = Content.objects.create(pulp_type="core.content")
+    version0 = repository.latest_version()
+
+    # Try to create a version but force an error
+    try:
+        with repository.new_version() as version1:
+            version1.add_content(Content.objects.filter(pk=content.pk))
+            # Force an error by raising an exception
+            raise ValueError("Simulated error during version creation")
+    except ValueError:
+        pass
+
+    # The version should not have been created due to rollback
+    assert repository.latest_version() == version0
+    # Repository should still be in a valid state
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk=content.pk))
+    assert repository.latest_version().number == 1
+
+
+def test_shared_artifact_same_path_validation(db, tmp_path, repository):
     """
     Test that multiple content units can reference the same artifact with the same
     relative path without causing validation errors.
@@ -270,10 +571,6 @@ def test_shared_artifact_same_path_validation(tmp_path):
     This reproduces scenarios where different content units legitimately share
     the same artifact (e.g. upstream source files).
     """
-    # Create a repository
-    repository = Repository.objects.create(name=uuid4())
-    repository.CONTENT_TYPES = [Content]
-
     # Create a shared artifact using proper test pattern
     artifact_path = tmp_path / "shared_file.txt"
     artifact_path.write_text("Shared content data")
@@ -300,16 +597,11 @@ def test_shared_artifact_same_path_validation(tmp_path):
     validate_version_paths(new_version)
 
 
-@pytest.mark.django_db
-def test_different_artifacts_same_path_validation_fails(tmp_path):
+def test_different_artifacts_same_path_validation_fails(db, tmp_path, repository):
     """
     Test that different artifacts trying to use the same relative path
     still fail validation (this is a real conflict that should be caught).
     """
-    # Create a repository
-    repository = Repository.objects.create(name=uuid4())
-    repository.CONTENT_TYPES = [Content]
-
     # Create two different artifacts using proper test pattern
     artifact1_path = tmp_path / "artifact1.txt"
     artifact1_path.write_text("Content of first artifact")
@@ -341,3 +633,158 @@ def test_different_artifacts_same_path_validation_fails(tmp_path):
     # This should raise a validation error due to path conflict
     with pytest.raises(ValueError, match="Repository version errors"):
         validate_version_paths(new_version)
+
+
+def test_content_relationships_after_version_deletion(
+    repository, add_content, remove_content, verify_content_sets
+):
+    """Verify behavior when comparing to a deleted base version."""
+    version0 = repository.latest_version()
+
+    # Create version1 with content 0, 1, 2, 4
+    with repository.new_version() as version1:
+        add_content(version1, [1, 1, 1, 0, 0])
+
+    # Create version2 that adds content 3 and removes content 2
+    with repository.new_version() as version2:
+        add_content(version2, [0, 0, 0, 0, 1])
+        remove_content(version2, [0, 1, 0, 0, 0])
+
+    # Create version3 that removes content
+    with repository.new_version() as version3:
+        remove_content(version3, [0, 0, 1, 0, 0])
+
+    verify_content_sets(version2, [1, 0, 1, 0, 1], [0, 0, 0, 0, 1], [0, 1, 0, 0, 0])
+    verify_content_sets(
+        version2, [1, 0, 1, 0, 1], [1, 0, 1, 0, 1], [0, 0, 0, 0, 0], base_version=version0
+    )
+    verify_content_sets(version3, [1, 0, 0, 0, 1], [0, 0, 0, 0, 0], [0, 0, 1, 0, 0])
+
+    # Delete version1
+    version1.delete()
+
+    verify_content_sets(version2, [1, 0, 1, 0, 1], [1, 0, 1, 0, 1], [0, 0, 0, 0, 0])
+    verify_content_sets(version3, [1, 0, 0, 0, 1], [0, 0, 0, 0, 0], [0, 0, 1, 0, 0])
+
+    # Verify that comparing to version0 still works
+    added_pks = set(version2.added(base_version=version0).values_list("pk", flat=True))
+    removed_pks = set(version2.removed(base_version=version0).values_list("pk", flat=True))
+    assert len(added_pks) == 3
+    assert len(removed_pks) == 0
+
+    # Delete version2
+    version2.delete()
+
+    verify_content_sets(version3, [1, 0, 0, 0, 1], [1, 0, 0, 0, 1], [0, 0, 0, 0, 0])
+
+    # Verify that comparing to version0 still works
+    added_pks = set(version3.added(base_version=version0).values_list("pk", flat=True))
+    removed_pks = set(version3.removed(base_version=version0).values_list("pk", flat=True))
+    assert len(added_pks) == 2
+    assert len(removed_pks) == 0
+
+
+def test_comparing_distant_versions(repository, add_content, remove_content, verify_content_sets):
+    """Verify comparing versions that are many versions apart."""
+    version0 = repository.latest_version()
+
+    # Create a chain of versions, each modifying content
+    versions = [version0]
+
+    # Version 1: Add content 0, 1
+    with repository.new_version() as v:
+        add_content(v, [1, 1, 0, 0, 0])
+    versions.append(v)
+
+    # Version 2: Add content 2
+    with repository.new_version() as v:
+        add_content(v, [0, 0, 1, 0, 0])
+    versions.append(v)
+
+    # Version 3: Remove content 0
+    with repository.new_version() as v:
+        remove_content(v, [1, 0, 0, 0, 0])
+    versions.append(v)
+
+    # Version 4: Add content 3
+    with repository.new_version() as v:
+        add_content(v, [0, 0, 0, 1, 0])
+    versions.append(v)
+
+    # Version 5: Remove content 1
+    with repository.new_version() as v:
+        remove_content(v, [0, 1, 0, 0, 0])
+    versions.append(v)
+
+    # Version 6: Add content 4
+    with repository.new_version() as v:
+        add_content(v, [0, 0, 0, 0, 1])
+    versions.append(v)
+
+    # Version 7: Add content 0 back
+    with repository.new_version() as v:
+        add_content(v, [1, 0, 0, 0, 0])
+    versions.append(v)
+
+    # Compare version 7 to version 1 (6 versions apart)
+    # Version 1: [1, 1, 0, 0, 0]
+    # Version 7: [1, 0, 1, 1, 1]
+    # added: 2, 3, 4
+    # removed: 1
+    verify_content_sets(
+        versions[7], [1, 0, 1, 1, 1], [0, 0, 1, 1, 1], [0, 1, 0, 0, 0], base_version=versions[1]
+    )
+
+    # Compare version 7 to version 0 (start)
+    verify_content_sets(
+        versions[7], [1, 0, 1, 1, 1], [1, 0, 1, 1, 1], [0, 0, 0, 0, 0], base_version=version0
+    )
+
+
+def test_batch_operations_preserve_correctness(repository, db):
+    """Verify that batching content operations maintains correctness."""
+    # Create content in batches
+    batch1 = [Content(pulp_type="core.content") for _ in range(30)]
+    batch2 = [Content(pulp_type="core.content") for _ in range(30)]
+    batch3 = [Content(pulp_type="core.content") for _ in range(40)]
+
+    Content.objects.bulk_create(batch1)
+    Content.objects.bulk_create(batch2)
+    Content.objects.bulk_create(batch3)
+
+    batch1_pks = sorted([c.pk for c in batch1])
+    batch2_pks = sorted([c.pk for c in batch2])
+    batch3_pks = sorted([c.pk for c in batch3])
+
+    # Add content in batches within a single version
+    with repository.new_version() as version1:
+        version1.add_content(Content.objects.filter(pk__in=batch1_pks))
+        version1.add_content(Content.objects.filter(pk__in=batch2_pks))
+        version1.add_content(Content.objects.filter(pk__in=batch3_pks))
+
+    # Verify all content was added
+    assert version1.content.count() == 100
+
+    # Verify RepositoryVersionContentDetails
+    rvcd_qs = RepositoryVersionContentDetails.objects.filter(
+        repository_version=version1, content_type="core.content"
+    )
+    assert rvcd_qs.get(count_type=RepositoryVersionContentDetails.PRESENT).count == 100
+    assert rvcd_qs.get(count_type=RepositoryVersionContentDetails.ADDED).count == 100
+    assert rvcd_qs.filter(count_type=RepositoryVersionContentDetails.REMOVED).first() is None
+
+    # Remove content in batches
+    with repository.new_version() as version2:
+        version2.remove_content(Content.objects.filter(pk__in=batch1_pks))
+        version2.remove_content(Content.objects.filter(pk__in=batch2_pks))
+
+    # Verify correct content remains
+    assert version2.content.count() == 40
+
+    # Verify RepositoryVersionContentDetails
+    rvcd_qs = RepositoryVersionContentDetails.objects.filter(
+        repository_version=version2, content_type="core.content"
+    )
+    assert rvcd_qs.get(count_type=RepositoryVersionContentDetails.PRESENT).count == 40
+    assert rvcd_qs.filter(count_type=RepositoryVersionContentDetails.ADDED).first() is None
+    assert rvcd_qs.get(count_type=RepositoryVersionContentDetails.REMOVED).count == 60