Filter occurrences by the algorithms that actually ran, and make class masking safe to re-run

.agents/AGENTS.md

@@ -19,6 +19,8 @@ Every call to the AI model API incurs a cost and requires electricity. Be smart
 - Focus on optimizing cold queries first before adding caching
 - When ordering by annotated fields, pagination COUNT queries include those annotations - use `.values('pk')` to strip them
 - For large tables (>10k rows), consider fuzzy counting using PostgreSQL's pg_class.reltuples
+- **Run `EXPLAIN (ANALYZE)` before claiming anything about a query plan.** Do not describe a query as "uses the index", "stays off the table", or "only touches N rows" from reading the ORM expression — verify it. A filter that looks bounded can still trigger a `Seq Scan`: an anti-join such as `pipelines__isnull=True` hides the candidate ids at plan time, so Postgres scans the whole table despite an index being available. Materializing the ids first and filtering on `field__in=[literal ids]` is what lets the planner use the index.
+- **Measure on the largest project in the local database, and survey several project sizes.** The local DB is a copy of production — use it. A small or empty project routinely hides the defect. If a query's cost does not move with project size, it is bound by total table size and will degrade as the platform grows regardless of tenant. Wrap timing loops in `cachalot_disabled()` and rebuild the queryset each iteration, or a reused queryset serves from `_result_cache` and reports a fake ~0 ms.
 
 **Git Commit Guidelines:**
 - Do NOT include "Generated with Claude Code" in commit messages

ami/main/api/views.py

@@ -35,6 +35,8 @@
 from ami.main.api.schemas import limit_doc_param, project_id_doc_param
 from ami.main.api.serializers import TagSerializer
 from ami.main.models_future.occurrence import model_agreement_for_project, top_identifiers_for_project
+from ami.ml.models.algorithm import Algorithm
+from ami.ml.serializers import AlgorithmSerializer
 from ami.utils.requests import get_default_classification_threshold
 from ami.utils.storages import ConnectionTestResult
 
@@ -1222,11 +1224,15 @@ def filter_queryset(self, request, queryset, view):
 
 class OccurrenceAlgorithmFilter(filters.BaseFilterBackend):
     """
-    Filter occurrences by the detection algorithm that detected them.
+    Filter occurrences by any algorithm that produced a result on them.
 
-    Accepts a list of algorithm ids to filter by or exclude by.
+    Matches an occurrence when one of its detections was made by the given
+    algorithm (detectors) or one of its classifications came from it
+    (classifiers and post-processing algorithms), so every algorithm listed
+    by ``Algorithm.objects.used_in_project()`` can be filtered here.
 
-    This filter can be both inclusive and exclusive.
+    Accepts a list of algorithm ids to filter by (``algorithm``) or exclude
+    by (``not_algorithm``). Both are supported and may be combined.
     """
 
     query_param = "algorithm"
@@ -1237,9 +1243,9 @@ def filter_queryset(self, request, queryset, view):
         algorithm_ids_exclusive = request.query_params.getlist(self.query_param_exclusive)
 
         if algorithm_ids:
-            queryset = queryset.filter(detections__classifications__algorithm__in=algorithm_ids)
+            queryset = queryset.processed_by_algorithm(algorithm_ids)
         if algorithm_ids_exclusive:
-            queryset = queryset.exclude(detections__classifications__algorithm__in=algorithm_ids_exclusive)
+            queryset = queryset.not_processed_by_algorithm(algorithm_ids_exclusive)
 
         return queryset
 
@@ -1467,6 +1473,33 @@ def get_queryset(self) -> QuerySet["Occurrence"]:
     def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)
 
+    @extend_schema(parameters=[project_id_doc_param], responses=AlgorithmSerializer(many=True))
+    @action(detail=False, methods=["get"], name="algorithms")
+    def algorithms(self, request: Request) -> Response:
+        """Choices for the occurrence algorithm filter: every algorithm that produced a result in the project.
+
+        Includes superseded pipeline versions and standalone post-processing algorithms
+        (their output is still filterable), so it differs from the project's algorithm
+        list at ``/ml/algorithms/``, which shows what the project can run — its
+        enabled-pipeline algorithms. Serving choices from what actually ran means the
+        filter never offers a value with zero results.
+        """
+        project = self.get_active_project()
+        if project is None:
+            raise api_exceptions.ValidationError({"project_id": "This parameter is required."})
+        if not Project.objects.visible_for_user(request.user).filter(pk=project.pk).exists():
+            raise NotFound("Project not found.")
+
+        qs = (
+            Algorithm.objects.all()
+            .with_category_count()  # type: ignore[union-attr] # Custom queryset method
+            .used_in_project(project)
+            .order_by("name")
+        )
+        page = self.paginate_queryset(qs)
+        serializer = AlgorithmSerializer(page, many=True, context=self.get_serializer_context())
+        return self.get_paginated_response(serializer.data)
+
 
 class OccurrenceStatsViewSet(viewsets.GenericViewSet, ProjectMixin):
     """Aggregate stats over Occurrences. Each @action == one stats kind.

ami/main/models.py

@@ -3353,6 +3353,33 @@ def valid(self):
     def with_detections_count(self):
         return self.annotate(detections_count=models.Count("detections", distinct=True))
 
+    def _processed_by_algorithm_q(self, algorithm_ids) -> Exists:
+        """Subquery matching occurrences with any result from the given algorithms —
+        a detection made by one (detectors) or a classification from one (classifiers
+        and post-processing algorithms such as class masking or a size filter)."""
+        return Exists(
+            Detection.objects.filter(occurrence_id=OuterRef("pk")).filter(
+                models.Q(detection_algorithm__in=algorithm_ids)
+                | models.Q(classifications__algorithm__in=algorithm_ids)
+            )
+        )
+
+    def processed_by_algorithm(self, algorithm_ids) -> "OccurrenceQuerySet":
+        """Occurrences with at least one result from the given algorithms.
+
+        Matches detectors through Detection.detection_algorithm and classifiers or
+        post-processing algorithms through their classifications, so every algorithm
+        listed by ``Algorithm.objects.used_in_project()`` can match here. The EXISTS
+        form returns each occurrence once; a join through
+        ``detections__classifications`` returns one row per matching result, which
+        inflates pagination counts and duplicates rows across pages.
+        """
+        return self.filter(self._processed_by_algorithm_q(algorithm_ids))
+
+    def not_processed_by_algorithm(self, algorithm_ids) -> "OccurrenceQuerySet":
+        """Occurrences with no result from any of the given algorithms."""
+        return self.exclude(self._processed_by_algorithm_q(algorithm_ids))
+
     def with_timestamps(self):
         """
         These are timestamps used for filtering and ordering in the UI.

ami/main/tests.py

@@ -6510,6 +6510,108 @@ def test_valid_returns_only_real_with_determination(self):
         self.assertNotIn(no_determination.pk, valid_pks)
 
 
+class TestOccurrenceAlgorithmFilterQuerySet(TestCase):
+    """
+    Covers OccurrenceQuerySet.processed_by_algorithm / not_processed_by_algorithm,
+    which back the ?algorithm= and ?not_algorithm= occurrence filters (PR #1368).
+
+    The filter must match an algorithm by either role it can play: the detector that
+    made a detection, or the classifier / post-processing algorithm that authored a
+    classification. It must also count each occurrence once — the previous join form
+    (``detections__classifications__algorithm__in``) returned one row per matching
+    classification, inflating the paginator's COUNT and duplicating occurrences across
+    pages.
+    """
+
+    def setUp(self):
+        from ami.main.models import Taxon
+        from ami.ml.models.algorithm import Algorithm
+
+        self.project = Project.objects.create(name="Occurrence Algo Filter Project")
+        self.deployment = Deployment.objects.create(project=self.project, name="dep")
+        self.event = Event.objects.create(
+            project=self.project,
+            deployment=self.deployment,
+            group_by="2024-01-01",
+            start=datetime.datetime(2024, 1, 1, 0, 0),
+        )
+        self.source_image = SourceImage.objects.create(
+            deployment=self.deployment,
+            project=self.project,
+            event=self.event,
+            path="occ-algo-filter.jpg",
+        )
+        self.taxon = Taxon.objects.create(name="Occurrence Algo Filter Taxon")
+
+        self.detector = Algorithm.objects.create(name="Filter Detector", version=1, task_type="localization")
+        self.classifier = Algorithm.objects.create(name="Filter Classifier", version=1, task_type="classification")
+        self.other = Algorithm.objects.create(name="Filter Other", version=1, task_type="classification")
+
+        # Detected by `detector`, classified once by `classifier`.
+        self.occ_classified = self._make_occurrence(detector=self.detector, classifiers=[self.classifier])
+        # Same pair, but classified three times — the double-count trap.
+        self.occ_multi = self._make_occurrence(
+            detector=self.detector, classifiers=[self.classifier, self.classifier, self.classifier]
+        )
+        # A different algorithm entirely, to prove filters are exclusive.
+        self.occ_other = self._make_occurrence(detector=self.other, classifiers=[self.other])
+
+    def _make_occurrence(self, detector, classifiers) -> Occurrence:
+        occ = Occurrence.objects.create(
+            project=self.project,
+            event=self.event,
+            deployment=self.deployment,
+            determination=self.taxon,
+        )
+        detection = Detection.objects.create(
+            source_image=self.source_image,
+            bbox=[0.0, 0.0, 1.0, 1.0],
+            detection_algorithm=detector,
+            occurrence=occ,
+        )
+        for classifier in classifiers:
+            detection.classifications.create(
+                taxon=self.taxon,
+                algorithm=classifier,
+                score=0.9,
+                timestamp=datetime.datetime.now(),
+            )
+        return occ
+
+    def _pks(self, queryset):
+        return set(queryset.values_list("pk", flat=True))
+
+    def test_filter_by_classifier_matches_its_occurrences(self):
+        matched = Occurrence.objects.filter(project=self.project).processed_by_algorithm([self.classifier.pk])
+        self.assertEqual(self._pks(matched), {self.occ_classified.pk, self.occ_multi.pk})
+
+    def test_filter_by_detector_matches_occurrences_it_detected(self):
+        """A detector authors no Classification, so the old classification-join form
+        returned nothing for it. Matching through Detection.detection_algorithm is what
+        lets the user filter by a localizer at all."""
+        matched = Occurrence.objects.filter(project=self.project).processed_by_algorithm([self.detector.pk])
+        self.assertEqual(self._pks(matched), {self.occ_classified.pk, self.occ_multi.pk})
+
+    def test_count_not_inflated_by_multiple_classifications(self):
+        """The occurrence with three classifications by the same algorithm must count
+        once. The paginator calls this same COUNT, so an inflated value would report
+        four occurrences where there are two and repeat rows across pages."""
+        matched = Occurrence.objects.filter(project=self.project).processed_by_algorithm([self.classifier.pk])
+        self.assertEqual(matched.count(), 2)
+
+    def test_exclude_removes_matching_occurrences(self):
+        remaining = Occurrence.objects.filter(project=self.project).not_processed_by_algorithm([self.other.pk])
+        self.assertEqual(self._pks(remaining), {self.occ_classified.pk, self.occ_multi.pk})
+
+    def test_exclude_is_the_complement_of_include(self):
+        base = Occurrence.objects.filter(project=self.project)
+        ids = [self.classifier.pk]
+        included = self._pks(base.processed_by_algorithm(ids))
+        excluded = self._pks(base.not_processed_by_algorithm(ids))
+        self.assertEqual(included | excluded, self._pks(base))
+        self.assertEqual(included & excluded, set())
+
+
 class TestCleanupNullOnlyOccurrencesCommand(TestCase):
     """
     Covers ami/main/management/commands/cleanup_null_only_occurrences.py.

ami/ml/models/algorithm.py

@@ -157,6 +157,46 @@ def with_category_count(self):
         """
         return self.annotate(category_count=ArrayLength("category_map__labels"))
 
+    def used_in_project(self, project) -> AlgorithmQuerySet:
+        """Algorithms that produced results in the project, whether or not their
+        pipeline is still enabled there.
+
+        "Used" means owning actual output rows: classifiers and post-processing
+        algorithms are found through their classifications, detectors through their
+        detections (detectors never author a Classification). Superseded pipeline
+        versions and standalone post-processing algorithms therefore stay listed as
+        long as their results exist, while a configured-but-never-run algorithm does
+        not appear.
+
+        Cost note (EXPLAIN-verified against a production copy): neither lookup can be
+        answered from an index, because neither table has a project column — project is
+        reachable only through source_image. Both sides scan, so the call costs roughly
+        0.1-0.6 s cold regardless of project size, growing with total table size.
+        Executes the two lookups immediately rather than lazily; the id lists are tiny
+        (one row per algorithm) and sorted so the SQL string, and therefore cachalot's
+        cache key, is stable. Making this index-fast requires a denormalised project
+        column on Classification/Detection.
+        """
+        from ami.main.models import Classification, Detection
+
+        # ``order_by()`` clears each model's default ordering before ``distinct()``;
+        # otherwise the ordering columns widen the DISTINCT back to one row per result.
+        classifier_ids = (
+            Classification.objects.filter(detection__source_image__project=project)
+            .order_by()
+            .values_list("algorithm_id", flat=True)
+            .distinct()
+        )
+        detector_ids = (
+            Detection.objects.filter(source_image__project=project)
+            .order_by()
+            .values_list("detection_algorithm", flat=True)
+            .distinct()
+        )
+        ids = set(classifier_ids) | set(detector_ids)
+        ids.discard(None)  # detections without a detection_algorithm
+        return self.filter(pk__in=sorted(ids))
+
 
 # Task types enum for better type checking
 class AlgorithmTaskType(str, enum.Enum):

ami/ml/post_processing/class_masking.py

@@ -272,7 +272,7 @@ def _get_or_create_masking_algorithm(
         return algorithm
 
     def _scoped_classifications(
-        self, config: ClassMaskingConfig, source_algorithm: Algorithm
+        self, config: ClassMaskingConfig, source_algorithm: Algorithm, masking_algorithm: Algorithm
     ) -> tuple[QuerySet[Classification], str]:
         """Resolve the terminal classifications to re-score from the config's scope.
 
@@ -284,13 +284,27 @@ def _scoped_classifications(
         arrays, which measured 208s versus 0.5s on a 55,530-row scope. Broadening
         either scope to match several collections or a whole project would fan out
         and need de-duplication again — restrict the columns first. See #1376.
+
+        Sources already re-scored by ``masking_algorithm`` are excluded via the
+        ``applied_to`` lineage, so a source is masked at most once per masking
+        algorithm even if it becomes terminal again later. A taxa list is identified
+        by primary key, not contents — re-masking against an edited list needs a new
+        list. See #1368.
         """
-        base = Classification.objects.filter(
-            terminal=True,
-            algorithm=source_algorithm,
-            scores__isnull=False,
-            logits__isnull=False,
-        ).select_related("detection", "detection__occurrence")
+        base = (
+            Classification.objects.filter(
+                terminal=True,
+                algorithm=source_algorithm,
+                scores__isnull=False,
+                logits__isnull=False,
+            )
+            # If another task needs this replay guard, hoist it to a
+            # ClassificationQuerySet method (e.g. not_derived_by(algorithm))
+            # rather than copying the exclude.
+            .exclude(derived_classifications__algorithm=masking_algorithm).select_related(
+                "detection", "detection__occurrence"
+            )
+        )
 
         if config.occurrence_id is not None:
             if not Occurrence.objects.filter(pk=config.occurrence_id).exists():
@@ -327,7 +341,7 @@ def run(self) -> None:
         masking_algorithm = self._get_or_create_masking_algorithm(
             source_algorithm, taxa_list, reweight=config.reweight
         )
-        classifications, scope_desc = self._scoped_classifications(config, source_algorithm)
+        classifications, scope_desc = self._scoped_classifications(config, source_algorithm, masking_algorithm)
         self.logger.info(f"Applying class masking on {scope_desc} using taxa list {taxa_list.pk}")
 
         def _on_setup(total: int) -> None: