diff --git a/ami/exports/format_types.py b/ami/exports/format_types.py
index 9c57b5c25..a3f4c82d0 100644
--- a/ami/exports/format_types.py
+++ b/ami/exports/format_types.py
@@ -44,6 +44,10 @@ def get_serializer_class(self):
         return get_export_serializer()
 
     def get_queryset(self):
+        # OccurrenceSerializer (the export base class) calls strict prefetch
+        # helpers in `_best_prediction` / `_best_identification` /
+        # `get_detection_images`. Apply detail prefetches so those helpers
+        # find detections + nested classifications in the cache.
         return (
             Occurrence.objects.valid()  # type: ignore[union-attr]  Custom manager method
             .filter(project=self.project)
@@ -55,6 +59,7 @@ def get_queryset(self):
             .with_timestamps()  # type: ignore[union-attr]  Custom queryset method
             .with_detections_count()
             .with_identifications()
+            .with_detail_prefetches()  # type: ignore[union-attr]  Custom queryset method
         )
 
     def export(self):
diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py
index ec79603e7..354df5459 100644
--- a/ami/main/api/serializers.py
+++ b/ami/main/api/serializers.py
@@ -591,7 +591,7 @@ def get_taxa(self, obj):
 class TaxonListSerializer(DefaultSerializer):
     # latest_detection = DetectionNestedSerializer(read_only=True)
     occurrences = serializers.SerializerMethodField()
-    parents = TaxonNestedSerializer(read_only=True)
+    parents = TaxonParentSerializer(many=True, read_only=True, source="parents_json")
     parent_id = serializers.PrimaryKeyRelatedField(queryset=Taxon.objects.all(), source="parent")
     tags = serializers.SerializerMethodField()
 
@@ -1314,10 +1314,14 @@ class Meta:
 
 
 class OccurrenceListSerializer(DefaultSerializer):
+    # List cards render one cover image; detail subclass raises this to 100.
+    detection_images_limit: int | None = 1
+
     determination = CaptureTaxonSerializer(read_only=True)
     deployment = DeploymentNestedSerializer(read_only=True)
     event = EventNestedSerializer(read_only=True)
     # first_appearance = TaxonSourceImageNestedSerializer(read_only=True)
+    detection_images = serializers.SerializerMethodField()
     determination_details = serializers.SerializerMethodField()
     best_machine_prediction = serializers.SerializerMethodField()
     identifications = OccurrenceIdentificationSerializer(many=True, read_only=True)
@@ -1364,27 +1368,32 @@ class Meta:
             "updated_at",
         ]
 
+    def get_detection_images(self, obj: Occurrence) -> list[str]:
+        from ami.main.models_future.occurrence import detection_image_urls_from_prefetch
+
+        return detection_image_urls_from_prefetch(obj, limit=self.detection_images_limit)
+
     def get_determination_details(self, obj: Occurrence):
-        # @TODO convert this to query methods to avoid N+1 queries.
-        # Currently at 100+ queries per page of 10 occurrences.
-        # Add a reusable method to the OccurrenceQuerySet class and call it from the ViewSet.
+        from ami.main.models_future.occurrence import best_identification_from_prefetch, best_prediction_from_prefetch
 
         context = self.context
-
         # Add this occurrence to the context so that the nested serializers can access it
         # the `parent` attribute is not available since we are manually instantiating the serializers
         context["occurrence"] = obj
 
         taxon = TaxonNestedSerializer(obj.determination, context=context).data if obj.determination else None
-        if obj.best_identification:
-            identification = OccurrenceIdentificationSerializer(obj.best_identification, context=context).data
+
+        best_ident = best_identification_from_prefetch(obj)
+        if best_ident:
+            identification = OccurrenceIdentificationSerializer(best_ident, context=context).data
         else:
             identification = None
 
-        if identification or not obj.best_prediction:
+        if identification:
             prediction = None
         else:
-            prediction = ClassificationNestedSerializer(obj.best_prediction, context=context).data
+            best_pred = best_prediction_from_prefetch(obj)
+            prediction = ClassificationNestedSerializer(best_pred, context=context).data if best_pred else None
 
         return dict(
             taxon=taxon,
@@ -1399,10 +1408,12 @@ def get_best_machine_prediction(self, obj: Occurrence) -> dict | None:
         Populated regardless of human verification status, so clients can always show
         the ML result alongside a human-set determination.
         """
+        from ami.main.models_future.occurrence import best_prediction_from_prefetch
+
         context = self.context
         context["occurrence"] = obj
 
-        prediction = obj.best_prediction
+        prediction = best_prediction_from_prefetch(obj)
         if not prediction:
             return None
 
@@ -1421,6 +1432,8 @@ def get_best_machine_prediction(self, obj: Occurrence) -> dict | None:
 
 
 class OccurrenceSerializer(OccurrenceListSerializer):
+    detection_images_limit: int | None = 100
+
     determination = CaptureTaxonSerializer(read_only=True)
     detections = DetectionNestedSerializer(many=True, read_only=True)
     identifications = OccurrenceIdentificationSerializer(many=True, read_only=True)
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index 9f38a970b..57654dde8 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -1227,12 +1227,10 @@ def get_queryset(self) -> QuerySet["Occurrence"]:
         qs = qs.with_detections_count().with_timestamps()  # type: ignore
         qs = qs.with_identifications()  # type: ignore
         qs = qs.apply_default_filters(project, self.request)  # type: ignore
-        if self.action != "list":
-            qs = qs.prefetch_related(
-                Prefetch(
-                    "detections", queryset=Detection.objects.order_by("-timestamp").select_related("source_image")
-                )
-            )
+        if self.action == "list":
+            qs = qs.with_list_prefetches()  # type: ignore
+        else:
+            qs = qs.with_detail_prefetches()  # type: ignore
 
         return qs
 
diff --git a/ami/main/models.py b/ami/main/models.py
index e91b395b7..b30b4e645 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -60,6 +60,11 @@
 # over non-terminal, then highest score, with pk as the deterministic tiebreaker.
 BEST_MACHINE_PREDICTION_ORDER: Final = ("-terminal", "-score", "-pk")
 
+# Ordering for "best identification" selection used by Occurrence.best_identification,
+# OccurrenceQuerySet.with_verification_info(), and best_identification_from_prefetch().
+# Most recent non-withdrawn identification wins, with pk as the deterministic tiebreaker.
+BEST_IDENTIFICATION_ORDER: Final = ("-created_at", "-pk")
+
 
 class TaxonRank(OrderedEnum):
     KINGDOM = "KINGDOM"
@@ -2932,6 +2937,18 @@ def with_identifications(self):
             "identifications__user",
         )
 
+    def with_list_prefetches(self):
+        """Add prefetches the list serializer needs (detection paths, classifications)."""
+        from ami.main.models_future.occurrence import prefetch_detections_for_list
+
+        return self.prefetch_related(prefetch_detections_for_list())
+
+    def with_detail_prefetches(self):
+        """Add prefetches the detail serializer needs (detections + source_image + classifications)."""
+        from ami.main.models_future.occurrence import prefetch_detections_for_detail
+
+        return self.prefetch_related(prefetch_detections_for_detail())
+
     def with_best_detection(self):
         """
         Annotate the queryset with fields from the best detection.
@@ -3014,7 +3031,7 @@ def with_verification_info(self):
         """
         best_identification_subquery = Identification.objects.filter(
             occurrence=OuterRef("pk"), withdrawn=False
-        ).order_by("-created_at")
+        ).order_by(*BEST_IDENTIFICATION_ORDER)
 
         return self.annotate(
             verified_by_name=models.Subquery(best_identification_subquery.values("user__name")[:1]),
@@ -3240,7 +3257,11 @@ def best_identification(self):
 
         @TODO this could use a confidence level chosen manually by the users/experts.
         """
-        return Identification.objects.filter(occurrence=self, withdrawn=False).order_by("-created_at").first()
+        return (
+            Identification.objects.filter(occurrence=self, withdrawn=False)
+            .order_by(*BEST_IDENTIFICATION_ORDER)
+            .first()
+        )
 
     def get_determination_score(self) -> float | None:
         if not self.determination:
diff --git a/ami/main/models_future/occurrence.py b/ami/main/models_future/occurrence.py
new file mode 100644
index 000000000..4592ec8c9
--- /dev/null
+++ b/ami/main/models_future/occurrence.py
@@ -0,0 +1,131 @@
+"""
+Reusable Prefetch factories for Occurrence list/detail rendering.
+
+The serializer trusts the prefetch contract — the viewset is the single place
+that wires it up. Don't gate serializer methods on `_prefetched_objects_cache`
+membership; require the prefetch.
+
+Tracking issue: https://github.com/RolnickLab/antenna/issues/1271
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from django.db.models import Prefetch
+
+if TYPE_CHECKING:
+    from ami.main.models import Classification, Identification, Occurrence
+
+
+def _detections_prefetch(*, ordering: tuple[str, ...], with_source_image: bool) -> Prefetch:
+    from ami.main.models import Classification, Detection
+
+    qs = Detection.objects.prefetch_related(
+        Prefetch(
+            "classifications",
+            queryset=Classification.objects.select_related("taxon", "algorithm"),
+        )
+    ).order_by(*ordering)
+    if with_source_image:
+        qs = qs.select_related("source_image")
+    return Prefetch("detections", queryset=qs)
+
+
+def prefetch_detections_for_list() -> Prefetch:
+    """Detections + nested classifications, ordered for stable list image galleries."""
+    return _detections_prefetch(ordering=("frame_num", "timestamp"), with_source_image=False)
+
+
+def prefetch_detections_for_detail() -> Prefetch:
+    """Detections + nested classifications + source_image, ordered most-recent-first.
+
+    Detail responses serialize each detection via `DetectionNestedSerializer`,
+    which dereferences `source_image` (as `capture`).
+    """
+    return _detections_prefetch(ordering=("-timestamp",), with_source_image=True)
+
+
+def _require_prefetch(occurrence: Occurrence, *relations: str) -> None:
+    """Raise if any required top-level relation is missing from the prefetch cache.
+
+    Only checks top-level relations; nested prefetch (e.g. `detections__classifications`)
+    is required by callers but not enforced here. Use the list/detail factories above
+    to keep the pairing correct. Tighter depth-checking is deferred to django-zen-queries
+    (#1271 follow-up).
+    """
+    cache = getattr(occurrence, "_prefetched_objects_cache", {})
+    missing = [r for r in relations if r not in cache]
+    if missing:
+        raise RuntimeError(
+            f"Occurrence {occurrence.pk} is missing prefetched relations {missing!r}. "
+            "Apply OccurrenceQuerySet.with_list_prefetches() / with_detail_prefetches() / "
+            "with_identifications() in the viewset's get_queryset()."
+        )
+
+
+def best_prediction_from_prefetch(occurrence: Occurrence) -> Classification | None:
+    """Pick the best machine prediction from a prefetched occurrence in Python.
+
+    Mirrors `Occurrence.best_prediction` (per-algorithm max-score, then `-terminal, -score`).
+    Skips `score=None` to match SQL semantics of `score__in=Subquery(...)`.
+
+    Strict: requires `detections` (and each detection's `classifications`) prefetched.
+    """
+    _require_prefetch(occurrence, "detections")
+    classifications = [
+        c for det in occurrence.detections.all() for c in det.classifications.all() if c.score is not None
+    ]
+    if not classifications:
+        return None
+
+    max_score_per_algo: dict[object, float] = {}
+    for c in classifications:
+        existing = max_score_per_algo.get(c.algorithm_id)
+        if existing is None or c.score > existing:
+            max_score_per_algo[c.algorithm_id] = c.score
+
+    candidates = [c for c in classifications if c.score == max_score_per_algo[c.algorithm_id]]
+    candidates.sort(
+        key=lambda c: (
+            0 if getattr(c, "terminal", False) else 1,
+            -c.score,
+            -c.pk,
+        )
+    )
+    return candidates[0]
+
+
+def best_identification_from_prefetch(occurrence: Occurrence) -> Identification | None:
+    """Pick the most recent non-withdrawn identification from prefetched data.
+
+    Mirrors `Occurrence.best_identification` (BEST_IDENTIFICATION_ORDER = -created_at, -pk).
+
+    Strict: requires `identifications` prefetched (via `OccurrenceQuerySet.with_identifications()`).
+    """
+    _require_prefetch(occurrence, "identifications")
+    best: Identification | None = None
+    best_key: tuple[bool, object, int] | None = None
+    for ident in occurrence.identifications.all():
+        if ident.withdrawn:
+            continue
+        ident_key: tuple[bool, object, int] = (ident.created_at is not None, ident.created_at, ident.pk)
+        if best_key is None or ident_key > best_key:
+            best = ident
+            best_key = ident_key
+    return best
+
+
+def detection_image_urls_from_prefetch(occurrence: Occurrence, limit: int | None = None) -> list[str]:
+    """Return media URLs for the prefetched detections (filtering out `path=None`).
+
+    Strict: requires `detections` prefetched. Pass `limit` to bound output.
+    """
+    _require_prefetch(occurrence, "detections")
+
+    from ami.main.models import get_media_url
+
+    detections = [det for det in occurrence.detections.all() if det.path]
+    if limit is not None:
+        detections = detections[:limit]
+    return [get_media_url(det.path) for det in detections]
diff --git a/ami/main/tests.py b/ami/main/tests.py
index 0e7672e1c..360384853 100644
--- a/ami/main/tests.py
+++ b/ami/main/tests.py
@@ -2927,6 +2927,326 @@ def test_taxa_include_occurrence_determinations_not_directly_linked(self):
         )
 
 
+class TestOccurrenceListQueryCount(APITestCase):
+    """Guard against N+1 regressions in OccurrenceViewSet.list (issue #1271).
+
+    The number of queries should be roughly constant regardless of page size:
+    a small handful of pagination/auth/permission queries plus a fixed set of
+    prefetches (occurrences, detections, classifications, identifications, etc.).
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=10)
+
+        # Use a low threshold so all occurrences are returned
+        self.project.default_filters_score_threshold = 0.0
+        self.project.save()
+
+        self.user = User.objects.create_user(email="qcount@insectai.org", is_staff=False, is_superuser=False)
+        self.client.force_authenticate(user=self.user)
+
+    def _list_occurrences(self, limit: int) -> int:
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url = f"/api/v2/occurrences/?project_id={self.project.pk}&limit={limit}"
+        # Clear cachalot cache between runs so query counts reflect cold state.
+        # Failures must propagate — a warm cache hides query-scaling regressions.
+        caches["default"].clear()
+
+        with CaptureQueriesContext(connection) as ctx:
+            res = self.client.get(url)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(res.data["results"]), min(limit, res.data["count"]))
+        # List responses cap detection_images per row (1 cover image; prevents unbounded payload).
+        for row in res.data["results"]:
+            self.assertLessEqual(len(row["detection_images"]), 1)
+        return len(ctx.captured_queries)
+
+    def test_list_query_count_does_not_scale_with_page_size(self):
+        """Query count for a page of 25 should be very close to a page of 5."""
+        create_occurrences(deployment=self.deployment, num=25, determination_score=0.9)
+
+        small_count = self._list_occurrences(limit=5)
+        large_count = self._list_occurrences(limit=25)
+
+        # Allow a small constant overhead (e.g. cachalot bookkeeping) but
+        # nowhere near 5x scaling.
+        self.assertLessEqual(
+            large_count,
+            small_count + 5,
+            f"Query count grew with page size: {small_count} -> {large_count} (likely N+1 regression)",
+        )
+
+
+class TestOccurrenceDetailQueryCount(APITestCase):
+    """Guard against N+1 regressions on the detail endpoint.
+
+    `OccurrenceSerializer` extends `OccurrenceListSerializer`, so any
+    cache-key gating in the shared base can silently regress the detail
+    path. This guard pins detail-endpoint query count.
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=10)
+
+        self.project.default_filters_score_threshold = 0.0
+        self.project.save()
+
+        self.user = User.objects.create_user(email="qcount-detail@insectai.org", is_staff=False, is_superuser=False)
+        self.client.force_authenticate(user=self.user)
+
+    def _detail_query_count(self, occurrence_id: int) -> int:
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url = f"/api/v2/occurrences/{occurrence_id}/?project_id={self.project.pk}"
+        caches["default"].clear()
+        with CaptureQueriesContext(connection) as ctx:
+            res = self.client.get(url)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        return len(ctx.captured_queries)
+
+    def test_detail_query_count_is_bounded(self):
+        """Detail endpoint query count must not scale with detections per occurrence."""
+        create_occurrences(deployment=self.deployment, num=5, determination_score=0.9)
+        occurrence_id = Occurrence.objects.filter(project=self.project).values_list("pk", flat=True).first()
+        self.assertIsNotNone(occurrence_id)
+
+        count = self._detail_query_count(occurrence_id)
+        # 7 list-path queries + a small allowance for detail-only (full detections,
+        # nested predictions, source_image select_related). Hard ceiling catches
+        # silent reintroduction of N+1 in the shared serializer base.
+        self.assertLess(count, 20, f"Detail endpoint took {count} queries (likely N+1 regression)")
+
+    def test_detail_query_count_does_not_scale_with_detections(self):
+        """Detail with many detections per occurrence must not multiply queries.
+
+        `create_occurrences` produces 1 detection/classification per occurrence.
+        Inflate one occurrence to N detections × N classifications and confirm
+        query count stays in the same envelope as the single-detection case
+        (i.e. bounded by the ceiling above, not detection_count).
+        """
+        import datetime
+
+        from ami.main.models import Classification, Detection
+
+        create_occurrences(deployment=self.deployment, num=3, determination_score=0.9)
+        occurrence = Occurrence.objects.filter(project=self.project).first()
+        self.assertIsNotNone(occurrence)
+
+        seed_detection = occurrence.detections.first()
+        self.assertIsNotNone(seed_detection)
+        seed_classification = seed_detection.classifications.first()
+        self.assertIsNotNone(seed_classification)
+        source_image = seed_detection.source_image
+
+        for i in range(8):
+            det = Detection.objects.create(
+                source_image=source_image,
+                occurrence=occurrence,
+                timestamp=source_image.timestamp,
+                bbox=[10, 10, 20, 20],
+                path=f"detections/inflated_{i}.jpg",
+            )
+            for j in range(3):
+                Classification.objects.create(
+                    detection=det,
+                    taxon=seed_classification.taxon,
+                    algorithm=seed_classification.algorithm,
+                    score=0.8 + (j * 0.01),
+                    timestamp=datetime.datetime.now(),
+                )
+
+        count = self._detail_query_count(occurrence.pk)
+        self.assertLess(
+            count,
+            20,
+            f"Detail with 9 detections × 4 classifications took {count} queries — N+1 regression",
+        )
+
+
+class TestOccurrencePrefetchHelpersEdgeCases(APITestCase):
+    """Helpers in `models_future/occurrence.py` must handle empty prefetch caches gracefully.
+
+    `.valid()` excludes zero-detection occurrences from the list endpoint, but
+    helpers can still be called on objects whose detections were deleted, or in
+    non-API contexts (admin, exports). Empty must return None / [] without 500.
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=2)
+        create_occurrences(deployment=self.deployment, num=2, determination_score=0.9)
+
+    def _empty_occurrence(self) -> "Occurrence":
+        """An occurrence with prefetch caches populated but all relations empty."""
+        from ami.main.models_future.occurrence import prefetch_detections_for_list
+
+        occurrence = (
+            Occurrence.objects.filter(project=self.project)
+            .prefetch_related(prefetch_detections_for_list(), "identifications")
+            .first()
+        )
+        assert occurrence is not None
+        # Wipe relations through DB to leave the prefetch cache empty.
+        occurrence.detections.all().delete()
+        occurrence.identifications.all().delete()
+        # Re-fetch with prefetches so the cache reflects the empty state.
+        return (
+            Occurrence.objects.filter(pk=occurrence.pk)
+            .prefetch_related(prefetch_detections_for_list(), "identifications")
+            .get()
+        )
+
+    def test_helpers_return_safe_defaults_on_empty(self):
+        from ami.main.models_future.occurrence import (
+            best_identification_from_prefetch,
+            best_prediction_from_prefetch,
+            detection_image_urls_from_prefetch,
+        )
+
+        occurrence = self._empty_occurrence()
+        self.assertIsNone(best_prediction_from_prefetch(occurrence))
+        self.assertIsNone(best_identification_from_prefetch(occurrence))
+        self.assertEqual(detection_image_urls_from_prefetch(occurrence), [])
+        self.assertEqual(detection_image_urls_from_prefetch(occurrence, limit=3), [])
+
+    def test_helpers_raise_when_prefetch_missing(self):
+        """Strict contract: missing prefetch must raise, not silently slow-path."""
+        from ami.main.models_future.occurrence import (
+            best_identification_from_prefetch,
+            best_prediction_from_prefetch,
+            detection_image_urls_from_prefetch,
+        )
+
+        # No prefetch applied
+        occurrence = Occurrence.objects.filter(project=self.project).first()
+        assert occurrence is not None
+        with self.assertRaises(RuntimeError):
+            best_prediction_from_prefetch(occurrence)
+        with self.assertRaises(RuntimeError):
+            best_identification_from_prefetch(occurrence)
+        with self.assertRaises(RuntimeError):
+            detection_image_urls_from_prefetch(occurrence)
+
+
+@override_settings(CACHALOT_ENABLED=False)
+class TestSourceImageListQueryCount(APITestCase):
+    """Audit SourceImageViewSet.list for N+1 (follow-up to PR #1274).
+
+    Cachalot disabled so we measure cold query count, not warm cache.
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=25)
+        create_occurrences(deployment=self.deployment, num=25, determination_score=0.9)
+
+        self.project.default_filters_score_threshold = 0.0
+        self.project.save()
+
+        self.user = User.objects.create_user(
+            email="qcount-sourceimage@insectai.org", is_staff=False, is_superuser=False
+        )
+        self.client.force_authenticate(user=self.user)
+
+    def _list_query_count(self, limit: int, with_detections: bool) -> int:
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url = (
+            f"/api/v2/captures/?project_id={self.project.pk}&limit={limit}"
+            f"&with_detections={'true' if with_detections else 'false'}"
+        )
+        caches["default"].clear()
+        with CaptureQueriesContext(connection) as ctx:
+            res = self.client.get(url)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        return len(ctx.captured_queries)
+
+    def test_list_query_count_does_not_scale_without_detections(self):
+        small = self._list_query_count(limit=5, with_detections=False)
+        large = self._list_query_count(limit=25, with_detections=False)
+        print(f"\n[AUDIT] SourceImage list (no detections): limit=5 -> {small}q, limit=25 -> {large}q")
+        self.assertLessEqual(large, small + 5, f"SourceImage list scaling: {small} -> {large} (likely N+1)")
+
+    def test_list_query_count_does_not_scale_with_detections(self):
+        small = self._list_query_count(limit=5, with_detections=True)
+        large = self._list_query_count(limit=25, with_detections=True)
+        print(f"\n[AUDIT] SourceImage list (with_detections=true): limit=5 -> {small}q, limit=25 -> {large}q")
+        self.assertLessEqual(
+            large, small + 5, f"SourceImage list scaling with detections: {small} -> {large} (likely N+1)"
+        )
+
+    def test_list_query_with_counts_and_detections(self):
+        """Realistic UI call: with_counts=true & with_detections=true at limit=25."""
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url_base = f"/api/v2/captures/?project_id={self.project.pk}&with_detections=true&with_counts=true"
+        # Two warmups: first request does session/auth bootstrap, second equalises pool state.
+        self.client.get(url_base + "&limit=1")
+        self.client.get(url_base + "&limit=1")
+        caches["default"].clear()
+        with CaptureQueriesContext(connection) as ctx_small:
+            self.client.get(url_base + "&limit=5")
+        caches["default"].clear()
+        with CaptureQueriesContext(connection) as ctx_large:
+            self.client.get(url_base + "&limit=25")
+        small, large = len(ctx_small.captured_queries), len(ctx_large.captured_queries)
+        print(
+            f"\n[AUDIT] SourceImage list (with_detections+with_counts): " f"limit=5 -> {small}q, limit=25 -> {large}q"
+        )
+        self.assertLessEqual(large, small + 5, f"SourceImage list scaling: {small} -> {large} (likely N+1)")
+
+
+@override_settings(CACHALOT_ENABLED=False)
+class TestTaxonListQueryCount(APITestCase):
+    """Audit TaxonViewSet.list for N+1 (follow-up to PR #1274, task #9/#15).
+
+    Cachalot disabled so we measure cold query count, not warm cache.
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=25)
+        # Spread occurrences across many taxa so the taxon list has 25+ rows.
+        taxa = list(Taxon.objects.filter(projects=self.project))
+        for taxon in taxa[:25]:
+            create_occurrences(deployment=self.deployment, num=2, taxon=taxon, determination_score=0.9)
+
+        self.project.default_filters_score_threshold = 0.0
+        self.project.save()
+
+        self.user = User.objects.create_user(email="qcount-taxon@insectai.org", is_staff=False, is_superuser=False)
+        self.client.force_authenticate(user=self.user)
+
+    def _list_query_count(self, limit: int) -> int:
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url = f"/api/v2/taxa/?project_id={self.project.pk}&limit={limit}"
+        caches["default"].clear()
+        with CaptureQueriesContext(connection) as ctx:
+            res = self.client.get(url)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        return len(ctx.captured_queries)
+
+    def test_list_query_count_does_not_scale_with_page_size(self):
+        small = self._list_query_count(limit=5)
+        large = self._list_query_count(limit=25)
+        print(f"\n[AUDIT] Taxon list: limit=5 -> {small}q, limit=25 -> {large}q")
+        self.assertLessEqual(large, small + 5, f"Taxon list scaling: {small} -> {large} (likely N+1)")
+
+
 class TestProjectDefaultTaxaFilter(APITestCase):
     """
     Tests for project default taxa filtering (include/exclude lists).
diff --git a/scripts/benchmark_occurrences_list.sh b/scripts/benchmark_occurrences_list.sh
new file mode 100755
index 000000000..9373d9e53
--- /dev/null
+++ b/scripts/benchmark_occurrences_list.sh
@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# Benchmark the /api/v2/occurrences/ list endpoint under concurrent load.
+#
+# Used to A/B compare a deployment with PR #1274 (occurrence list N+1 fix)
+# against a baseline. Each request uses a randomized offset to bypass
+# django-cachalot. Reports avg / p50 / p95 / p99 / max plus error counts.
+#
+# Usage:
+#   ./scripts/benchmark_occurrences_list.sh <base_url> <project_id> <concurrency> <total> <limit>
+#
+# Example (prod):
+#   ./scripts/benchmark_occurrences_list.sh https://api.antenna.insectai.org 18 10 30 25
+#
+# Example (staging A/B):
+#   ./scripts/benchmark_occurrences_list.sh https://arctia.dev.antenna.insectai.org 5 10 30 25
+#   ./scripts/benchmark_occurrences_list.sh https://serbia.dev.antenna.insectai.org 5 10 30 25
+
+set -euo pipefail
+
+BASE_URL="${1:?base_url required, e.g. https://api.antenna.insectai.org}"
+PROJECT_ID="${2:?project_id required}"
+CONCURRENCY="${3:-10}"
+TOTAL="${4:-30}"
+LIMIT="${5:-25}"
+
+# Probe project size so the random offset stays within the dataset
+COUNT=$(curl -s "${BASE_URL}/api/v2/occurrences/?project_id=${PROJECT_ID}&limit=1" | jq -r '.count')
+if [[ -z "${COUNT}" || "${COUNT}" == "null" ]]; then
+  echo "Failed to probe occurrence count at ${BASE_URL} project ${PROJECT_ID}" >&2
+  exit 1
+fi
+MAX_OFFSET=$(( COUNT > LIMIT ? COUNT - LIMIT : 0 ))
+
+OUT=$(mktemp)
+trap 'rm -f "${OUT}"' EXIT
+
+echo "Benchmark: ${BASE_URL} project=${PROJECT_ID} count=${COUNT} concurrency=${CONCURRENCY} total=${TOTAL} limit=${LIMIT}"
+
+START=$(date +%s.%N)
+seq 1 "${TOTAL}" | xargs -n1 -P"${CONCURRENCY}" -I{} bash -c "
+  offset=\$(( ${MAX_OFFSET} > 0 ? RANDOM % ${MAX_OFFSET} : 0 ))
+  curl -s -o /dev/null -w '%{http_code} %{time_total}\n' \
+    '${BASE_URL}/api/v2/occurrences/?project_id=${PROJECT_ID}&limit=${LIMIT}&offset='\${offset}
+" > "${OUT}"
+END=$(date +%s.%N)
+WALL=$(awk "BEGIN {printf \"%.3f\", ${END} - ${START}}")
+
+ERRS=$(awk '$1!=200{print}' "${OUT}" | wc -l | tr -d ' ')
+ERR_CODES=$(awk '$1!=200{print $1}' "${OUT}" | sort | uniq -c | tr '\n' ' ')
+
+OK_TIMES=$(awk '$1==200{print $2}' "${OUT}" | sort -n)
+OK_COUNT=$(echo "${OK_TIMES}" | grep -c . || echo 0)
+
+if [[ "${OK_COUNT}" -eq 0 ]]; then
+  echo "wall=${WALL}s errors=${ERRS}/${TOTAL} (${ERR_CODES})"
+  exit 1
+fi
+
+P50=$(echo "${OK_TIMES}" | awk -v n="${OK_COUNT}" 'NR==int(n*0.5)+1 {printf "%.3f", $1}')
+P95=$(echo "${OK_TIMES}" | awk -v n="${OK_COUNT}" 'NR==int(n*0.95) {printf "%.3f", $1}')
+P99=$(echo "${OK_TIMES}" | awk -v n="${OK_COUNT}" 'NR==int(n*0.99) {printf "%.3f", $1}')
+AVG=$(echo "${OK_TIMES}" | awk '{s+=$1} END{printf "%.3f", s/NR}')
+MIN=$(echo "${OK_TIMES}" | head -1 | awk '{printf "%.3f", $1}')
+MAX=$(echo "${OK_TIMES}" | tail -1 | awk '{printf "%.3f", $1}')
+
+echo "wall=${WALL}s avg=${AVG}s min=${MIN}s p50=${P50}s p95=${P95}s p99=${P99}s max=${MAX}s errors=${ERRS}/${TOTAL} (${ERR_CODES:-none})"