From 2cac8246d36164acbcfd9987feff596570b74cac Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Thu, 9 Jul 2026 12:06:54 -0700 Subject: [PATCH 01/11] fix(algorithms): include a project's post-processing algorithms in its algorithm list The project-scoped algorithm list only returned algorithms attached to an enabled pipeline config. Post-processing algorithms such as class masking are created standalone with no pipeline, so they were hidden from the project's algorithm filter even though they produce determinations in the project. The list now also includes any algorithm that produced classifications in the project, so an operator can filter occurrences by a masked result. The detail endpoint was already unscoped; this brings the list in line with it. Co-Authored-By: Claude --- ami/ml/tests.py | 36 ++++++++++++++++++++++++++++++++++-- ami/ml/views.py | 21 ++++++++++++++++----- 2 files changed, 50 insertions(+), 7 deletions(-) diff --git a/ami/ml/tests.py b/ami/ml/tests.py index cb88ee6fd..dd3cc1b68 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2037,8 +2037,10 @@ def test_deployment_counts_refresh_after_save_results(self): class TestAlgorithmViewSetProjectFilter(APITestCase): """ - The algorithm list endpoint is scoped to algorithms belonging to - pipelines enabled for the active project. + The algorithm list endpoint is scoped to algorithms relevant to the active + project: those belonging to an enabled pipeline, plus any that produced + classifications in the project (e.g. post-processing algorithms like class + masking, which are created standalone with no pipeline). """ def setUp(self): @@ -2103,3 +2105,33 @@ def test_detail_endpoint_unscoped_even_with_project_id(self): response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["name"], "Algo Disabled") + + def _classify_in_project(self, algorithm, project): + """Give ``algorithm`` a terminal classification whose capture is in ``project``.""" + source_image = SourceImage.objects.create(project=project) + detection = Detection.objects.create(source_image=source_image) + return Classification.objects.create( + detection=detection, + algorithm=algorithm, + timestamp=datetime.datetime.now(datetime.timezone.utc), + ) + + def test_lists_post_processing_algorithm_with_classifications_in_project(self): + """A post-processing algorithm has no pipeline but produces determinations in + the project, so the list must include it — otherwise the user cannot filter + occurrences by the masked result.""" + masked_algo = Algorithm.objects.create(name="Class Masked Classifier", version=1) + self._classify_in_project(masked_algo, self.project) + + names = self._list_algorithm_names(project_id=self.project.pk) + self.assertIn("Class Masked Classifier", names) + self.assertIn("Algo Enabled", names, "Enabled-pipeline algorithms still appear") + self.assertNotIn("Algo Disabled", names, "A disabled pipeline with no classifications stays hidden") + + def test_classifications_in_other_project_do_not_leak(self): + """An algorithm whose classifications live in another project must not appear.""" + other_masked_algo = Algorithm.objects.create(name="Other Project Masked", version=1) + self._classify_in_project(other_masked_algo, self.other_project) + + names = self._list_algorithm_names(project_id=self.project.pk) + self.assertNotIn("Other Project Masked", names) diff --git a/ami/ml/views.py b/ami/ml/views.py index 7de502f4a..47a4c33b3 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -1,7 +1,7 @@ import logging from django.db import transaction -from django.db.models import Prefetch +from django.db.models import Prefetch, Q from django.db.models.query import QuerySet from django.utils.text import slugify from drf_spectacular.utils import extend_schema @@ -15,7 +15,7 @@ from ami.base.views import ProjectMixin from ami.main.api.schemas import project_id_doc_param from ami.main.api.views import DefaultViewSet -from ami.main.models import Project, SourceImage +from ami.main.models import Classification, Project, SourceImage from ami.ml.schemas import PipelineRegistrationResponse from .models.algorithm import Algorithm, AlgorithmCategoryMap @@ -56,14 +56,25 @@ class AlgorithmViewSet(DefaultViewSet, ProjectMixin): def get_queryset(self) -> QuerySet["Algorithm"]: qs: QuerySet["Algorithm"] = super().get_queryset() qs = qs.with_category_count() # type: ignore[union-attr] # Custom queryset method - # Only scope list by project. Detail stays unscoped so links from historical + # Only scope the list by project. Detail stays unscoped so links from historical # classifications whose pipeline is no longer enabled still resolve. if getattr(self, "action", None) == "list": project = self.get_active_project() if project: + # An algorithm is relevant to the project if it is configured via an + # enabled pipeline OR it produced classifications in the project. + # Post-processing algorithms (e.g. class masking) are created standalone + # with no pipeline, so the pipeline join alone would hide them even though + # they own determinations the user needs to filter occurrences by. + classified_in_project = Classification.objects.filter(detection__source_image__project=project).values( + "algorithm" + ) qs = qs.filter( - pipelines__project_pipeline_configs__project=project, - pipelines__project_pipeline_configs__enabled=True, + Q( + pipelines__project_pipeline_configs__project=project, + pipelines__project_pipeline_configs__enabled=True, + ) + | Q(pk__in=classified_in_project) ).distinct() return qs From 7a7b69bd746923fc23bf5b2c6757cd5bae918242 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Thu, 9 Jul 2026 12:06:56 -0700 Subject: [PATCH 02/11] fix(post-processing): make class masking idempotent across re-runs Class masking previously relied only on the terminal flag it flips on the source classification to avoid re-scoring it again. If a source became terminal again (after a dedup or re-classification pass), or a partially completed run was retried, the source could be masked a second time and gain a duplicate masked classification. The scope now also excludes sources that already have a masked child for the same masking algorithm, keyed on the applied_to lineage, so a source is masked at most once per masking algorithm. This makes finishing an interrupted run safe. Co-Authored-By: Claude --- ami/ml/post_processing/class_masking.py | 28 +++++++---- .../tests/test_class_masking.py | 47 +++++++++++++++++++ 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/ami/ml/post_processing/class_masking.py b/ami/ml/post_processing/class_masking.py index ffc77f346..f7c18d224 100644 --- a/ami/ml/post_processing/class_masking.py +++ b/ami/ml/post_processing/class_masking.py @@ -263,19 +263,31 @@ 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. ``config_schema`` guarantees exactly one scope id is set, so the single ``else`` branch is sound. + + Sources already re-scored by ``masking_algorithm`` are excluded via the + ``applied_to`` lineage so the mask is idempotent: a source is masked at most + once per masking algorithm. This keeps re-runs safe — finishing a partially + completed run (e.g. one the health-check reaper revoked) processes only the + remainder, and a source that became terminal again (after a dedup or + re-classification pass) is not masked a second time. The guard is on the + lineage rather than the terminal flag, which is why it survives that churn. """ - 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, + ) + .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(): @@ -312,7 +324,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_batch(m: dict) -> None: diff --git a/ami/ml/post_processing/tests/test_class_masking.py b/ami/ml/post_processing/tests/test_class_masking.py index b0578272c..458018ca7 100644 --- a/ami/ml/post_processing/tests/test_class_masking.py +++ b/ami/ml/post_processing/tests/test_class_masking.py @@ -253,6 +253,53 @@ def test_task_run_collection_scope_persists_masking_algorithm(self): occ.refresh_from_db() self.assertEqual(occ.determination, self.species_taxa[1], "Occurrence determination follows the masked result") + def test_rerun_does_not_duplicate_masked_classifications(self): + """Re-running the same mask must not create a second masked classification for + a source already re-scored, even if that source became terminal again in between. + + Idempotency is keyed on the ``applied_to`` lineage — a source is masked at most + once per masking algorithm — not on the terminal flag. This makes it safe to + finish a partially completed run (e.g. one the health-check reaper revoked) or + to re-run after a re-classification / dedup pass re-terminalized a source. + """ + logits = [0.5, 3.0, 3.5] # excluded index 2 is top; index 1 is the in-list winner + taxa_list = TaxaList.objects.create(name="Idempotency list") + taxa_list.taxa.set(self.species_taxa[:2]) + + det, _ = self._detection_with_occurrence() + original = self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits) + + def run(): + ClassMaskingTask( + source_image_collection_id=self.collection.pk, + taxa_list_id=taxa_list.pk, + algorithm_id=self.algorithm.pk, + ).run() + + run() + masking_algo = Algorithm.objects.get( + key=f"{self.algorithm.key}_filtered_by_taxa_list_{taxa_list.pk}_reweighted" + ) + self.assertEqual( + Classification.objects.filter(algorithm=masking_algo, applied_to=original).count(), + 1, + "First run masks the source exactly once", + ) + + # Simulate the source becoming terminal again (a dedup or re-classification pass) + # while its masked child still exists. The terminal filter alone would re-select + # it; the applied_to guard must still skip it. + original.refresh_from_db() + original.terminal = True + original.save(update_fields=["terminal"]) + + run() + self.assertEqual( + Classification.objects.filter(algorithm=masking_algo, applied_to=original).count(), + 1, + "Re-run must not create a duplicate masked classification for an already-masked source", + ) + def test_reweight_modes_get_distinct_masking_algorithms(self): """The reweight mode is part of the masking algorithm's identity. From 87a68ada5e13e80df49133ca4044d9c6cf058ff9 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:23:28 -0700 Subject: [PATCH 03/11] perf(algorithms): scope the project algorithm list to algorithms that ran List an algorithm for a project when it produced classifications there, and drop the enabled-pipeline branch entirely. A configured but never-run algorithm has no results to filter or inspect, so listing it only adds dead entries to the filter. Removing the branch also removes the OR across the pipeline join and the SELECT DISTINCT it forced. Measured against a local copy of production data (roughly 834k classifications), on one of the larger projects the paginated endpoint drops from 2206 ms to 556 ms, most of it in the pagination COUNT: 1722 ms to 275 ms. Both remaining shapes still sequentially scan the classification and detection tables, because no index reaches a project from a classification. The cost therefore tracks total table size rather than project size, and improving it further needs a denormalised project on Classification or a precomputed per-project list. Co-Authored-By: Claude --- ami/ml/tests.py | 61 ++++++++++++++++++++++++++++++------------------- ami/ml/views.py | 21 +++++++---------- 2 files changed, 45 insertions(+), 37 deletions(-) diff --git a/ami/ml/tests.py b/ami/ml/tests.py index dd3cc1b68..c465276e5 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2037,10 +2037,14 @@ def test_deployment_counts_refresh_after_save_results(self): class TestAlgorithmViewSetProjectFilter(APITestCase): """ - The algorithm list endpoint is scoped to algorithms relevant to the active - project: those belonging to an enabled pipeline, plus any that produced - classifications in the project (e.g. post-processing algorithms like class - masking, which are created standalone with no pipeline). + The algorithm list endpoint is scoped to the algorithms that actually produced + classifications in the active project. + + Pipeline configuration does not grant an algorithm a place in the list. An + algorithm wired to an enabled pipeline but never run has no results to filter or + inspect, and post-processing algorithms such as class masking are created + standalone with no pipeline at all, so pipeline membership is neither necessary + nor sufficient. """ def setUp(self): @@ -2050,16 +2054,18 @@ def setUp(self): self.project = Project.objects.create(name="Algo Project A", create_defaults=False) self.other_project = Project.objects.create(name="Algo Project B", create_defaults=False) - # Project A: one enabled pipeline, one disabled pipeline - self.algo_enabled = Algorithm.objects.create(name="Algo Enabled", version=1) + # Project A: an algorithm that has run, and one configured via a disabled pipeline + self.algo_used = Algorithm.objects.create(name="Algo Used", version=1) self.algo_disabled = Algorithm.objects.create(name="Algo Disabled", version=1) - # Project B: a different pipeline/algorithm + # Project A: enabled pipeline, but the algorithm never produced anything + self.algo_configured_unused = Algorithm.objects.create(name="Algo Configured Unused", version=1) + # Project B: a different algorithm, also used self.algo_other_project = Algorithm.objects.create(name="Algo Other Project", version=1) - # Unrelated algorithm not attached to any pipeline + # Unrelated algorithm not attached to any pipeline and never run self.algo_orphan = Algorithm.objects.create(name="Algo Orphan", version=1) enabled_pipeline = Pipeline.objects.create(name="Enabled Pipeline") - enabled_pipeline.algorithms.add(self.algo_enabled) + enabled_pipeline.algorithms.add(self.algo_used, self.algo_configured_unused) ProjectPipelineConfig.objects.create(project=self.project, pipeline=enabled_pipeline, enabled=True) disabled_pipeline = Pipeline.objects.create(name="Disabled Pipeline") @@ -2070,8 +2076,21 @@ def setUp(self): other_pipeline.algorithms.add(self.algo_other_project) ProjectPipelineConfig.objects.create(project=self.other_project, pipeline=other_pipeline, enabled=True) + self._classify_in_project(self.algo_used, self.project) + self._classify_in_project(self.algo_other_project, self.other_project) + self.client.force_authenticate(user=self.user) + def _classify_in_project(self, algorithm, project): + """Give ``algorithm`` a classification whose capture belongs to ``project``.""" + source_image = SourceImage.objects.create(project=project) + detection = Detection.objects.create(source_image=source_image) + return Classification.objects.create( + detection=detection, + algorithm=algorithm, + timestamp=datetime.datetime.now(datetime.timezone.utc), + ) + def _list_algorithm_names(self, project_id=None): params = {"project_id": project_id} if project_id is not None else {} url = reverse_with_params("api:algorithm-list", params=params) @@ -2079,9 +2098,15 @@ def _list_algorithm_names(self, project_id=None): self.assertEqual(response.status_code, 200) return {row["name"] for row in response.json()["results"]} - def test_lists_only_enabled_pipeline_algorithms_for_project(self): + def test_lists_only_algorithms_used_in_project(self): + """Having run in the project is what puts an algorithm in the list. + + ``Algo Configured Unused`` shares an enabled pipeline with ``Algo Used`` and is + excluded purely because it never classified anything, which pins the positive and + negative sides of the rule against an otherwise identical pair. + """ names = self._list_algorithm_names(project_id=self.project.pk) - self.assertEqual(names, {"Algo Enabled"}) + self.assertEqual(names, {"Algo Used"}) def test_other_project_only_sees_its_own_algorithms(self): names = self._list_algorithm_names(project_id=self.other_project.pk) @@ -2090,7 +2115,7 @@ def test_other_project_only_sees_its_own_algorithms(self): def test_unscoped_request_returns_all_algorithms(self): """Without project_id, current behavior lists all algorithms (unchanged).""" names = self._list_algorithm_names() - self.assertIn("Algo Enabled", names) + self.assertIn("Algo Used", names) self.assertIn("Algo Disabled", names) self.assertIn("Algo Other Project", names) self.assertIn("Algo Orphan", names) @@ -2106,16 +2131,6 @@ def test_detail_endpoint_unscoped_even_with_project_id(self): self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["name"], "Algo Disabled") - def _classify_in_project(self, algorithm, project): - """Give ``algorithm`` a terminal classification whose capture is in ``project``.""" - source_image = SourceImage.objects.create(project=project) - detection = Detection.objects.create(source_image=source_image) - return Classification.objects.create( - detection=detection, - algorithm=algorithm, - timestamp=datetime.datetime.now(datetime.timezone.utc), - ) - def test_lists_post_processing_algorithm_with_classifications_in_project(self): """A post-processing algorithm has no pipeline but produces determinations in the project, so the list must include it — otherwise the user cannot filter @@ -2125,8 +2140,6 @@ def test_lists_post_processing_algorithm_with_classifications_in_project(self): names = self._list_algorithm_names(project_id=self.project.pk) self.assertIn("Class Masked Classifier", names) - self.assertIn("Algo Enabled", names, "Enabled-pipeline algorithms still appear") - self.assertNotIn("Algo Disabled", names, "A disabled pipeline with no classifications stays hidden") def test_classifications_in_other_project_do_not_leak(self): """An algorithm whose classifications live in another project must not appear.""" diff --git a/ami/ml/views.py b/ami/ml/views.py index 47a4c33b3..e70768417 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -1,7 +1,7 @@ import logging from django.db import transaction -from django.db.models import Prefetch, Q +from django.db.models import Prefetch from django.db.models.query import QuerySet from django.utils.text import slugify from drf_spectacular.utils import extend_schema @@ -61,21 +61,16 @@ def get_queryset(self) -> QuerySet["Algorithm"]: if getattr(self, "action", None) == "list": project = self.get_active_project() if project: - # An algorithm is relevant to the project if it is configured via an - # enabled pipeline OR it produced classifications in the project. - # Post-processing algorithms (e.g. class masking) are created standalone - # with no pipeline, so the pipeline join alone would hide them even though - # they own determinations the user needs to filter occurrences by. + # An algorithm belongs to the project if it actually produced classifications + # there. Pipeline configuration is deliberately not consulted: a configured but + # never-run algorithm has no results to filter or inspect, and post-processing + # algorithms (e.g. class masking) are created standalone with no pipeline at all, + # so a pipeline join would both list algorithms with nothing behind them and hide + # ones that own live determinations. classified_in_project = Classification.objects.filter(detection__source_image__project=project).values( "algorithm" ) - qs = qs.filter( - Q( - pipelines__project_pipeline_configs__project=project, - pipelines__project_pipeline_configs__enabled=True, - ) - | Q(pk__in=classified_in_project) - ).distinct() + qs = qs.filter(pk__in=classified_in_project) return qs @extend_schema(parameters=[project_id_doc_param]) From 46d7c67ea97a6d9fcd8e3c38e297b07823df5eb0 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:26:17 -0700 Subject: [PATCH 04/11] docs(post-processing): state what the masking idempotency guard actually covers The docstring credited the lineage guard with resuming interrupted runs. It does not: demoting a source and writing its masked child share one transaction, so a resumed run already skips completed sources on the terminal flag alone. The guard covers only the case the flag cannot, a source restored to terminal by a later dedup or re-classification pass. Also record that a taxa list is identified by primary key rather than by contents, so editing a list in place and re-running the same mask leaves already-masked sources scored against the earlier membership. Co-Authored-By: Claude --- ami/ml/post_processing/class_masking.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/ami/ml/post_processing/class_masking.py b/ami/ml/post_processing/class_masking.py index f7c18d224..512d2833c 100644 --- a/ami/ml/post_processing/class_masking.py +++ b/ami/ml/post_processing/class_masking.py @@ -271,12 +271,19 @@ def _scoped_classifications( ``else`` branch is sound. Sources already re-scored by ``masking_algorithm`` are excluded via the - ``applied_to`` lineage so the mask is idempotent: a source is masked at most - once per masking algorithm. This keeps re-runs safe — finishing a partially - completed run (e.g. one the health-check reaper revoked) processes only the - remainder, and a source that became terminal again (after a dedup or - re-classification pass) is not masked a second time. The guard is on the - lineage rather than the terminal flag, which is why it survives that churn. + ``applied_to`` lineage, so a source carries at most one masked classification + per masking algorithm however many times the mask is re-run. + + The ``terminal=True`` filter alone already handles an interrupted run, because + demoting a source and writing its masked child share one transaction: a resumed + run sees completed sources as non-terminal and skips them. The lineage guard + covers the case the flag cannot — a source restored to terminal by a later dedup + or re-classification pass, which the flag would re-select and mask a second time. + + A taxa list is identified here by primary key, not by its contents. Editing a + list in place and re-running the same mask therefore skips the sources it + already masked, leaving them scored against the list's earlier membership. + Re-masking against changed taxa needs a new taxa list. """ base = ( Classification.objects.filter( From 80b80c2db46370805174f6d15a3f8e108fe46e82 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:33:18 -0700 Subject: [PATCH 05/11] refactor(algorithms): deduplicate the project algorithm subquery The subquery returned one row per classification, so a project with a million classifications fed a million rows into the IN clause to identify at most a few dozen algorithms. Selecting distinct algorithm ids says what the query means. Measured effect is small and close to run-to-run noise (median 610 ms -> 567 ms over four runs against a local copy of production data); the change is for clarity. Co-Authored-By: Claude --- ami/ml/views.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/ami/ml/views.py b/ami/ml/views.py index e70768417..f1dedb72d 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -67,8 +67,10 @@ def get_queryset(self) -> QuerySet["Algorithm"]: # algorithms (e.g. class masking) are created standalone with no pipeline at all, # so a pipeline join would both list algorithms with nothing behind them and hide # ones that own live determinations. - classified_in_project = Classification.objects.filter(detection__source_image__project=project).values( - "algorithm" + classified_in_project = ( + Classification.objects.filter(detection__source_image__project=project) + .values_list("algorithm", flat=True) + .distinct() ) qs = qs.filter(pk__in=classified_in_project) return qs From 0a6dfa0f9298789028a231fffc4161410bec002b Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:42:33 -0700 Subject: [PATCH 06/11] fix(algorithms): keep pipeline-configured algorithms in the project list Scoping the list purely by classification authorship dropped every detector. Localizers set Detection.detection_algorithm and never write a Classification, so the detector behind every detection in a project disappeared from the project's algorithm list, which renders a task_type column precisely to show detectors next to classifiers. Restore the enabled-pipeline join as the primary rule and admit post-processing algorithms through a second query, restricted to algorithms with no pipeline so the classification lookup stays off the full table. Collect the two separately rather than OR-ing them: an OR across the pipeline join forces a SELECT DISTINCT whose COUNT costs more than both queries together. Measured against a local copy of production data, on the largest project the list returns in 46 ms, versus 8 ms for the pipeline-only rule on main and 595 ms for the authorship-only rule this replaces. The result is a superset of main: nothing main listed is lost, and the post-processing algorithms are added. An algorithm attached to no pipeline that only ever produced detections is still absent, as it is on main. Catching it would mean scanning the detection table. Co-Authored-By: Claude --- ami/ml/tests.py | 42 ++++++++++++++++++++++++++++-------------- ami/ml/views.py | 37 ++++++++++++++++++++++++------------- 2 files changed, 52 insertions(+), 27 deletions(-) diff --git a/ami/ml/tests.py b/ami/ml/tests.py index c465276e5..706f8f6b7 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2037,14 +2037,13 @@ def test_deployment_counts_refresh_after_save_results(self): class TestAlgorithmViewSetProjectFilter(APITestCase): """ - The algorithm list endpoint is scoped to the algorithms that actually produced - classifications in the active project. - - Pipeline configuration does not grant an algorithm a place in the list. An - algorithm wired to an enabled pipeline but never run has no results to filter or - inspect, and post-processing algorithms such as class masking are created - standalone with no pipeline at all, so pipeline membership is neither necessary - nor sufficient. + The algorithm list endpoint is scoped to the algorithms relevant to the active + project, which they reach two ways. + + An enabled pipeline configures most of them. That covers detectors, which never + author a Classification and so could not be found by their results. Post-processing + algorithms such as class masking have no pipeline at all, so they are admitted by + having classified something in the project. """ def setUp(self): @@ -2098,15 +2097,30 @@ def _list_algorithm_names(self, project_id=None): self.assertEqual(response.status_code, 200) return {row["name"] for row in response.json()["results"]} - def test_lists_only_algorithms_used_in_project(self): - """Having run in the project is what puts an algorithm in the list. + def test_lists_enabled_pipeline_algorithms_for_project(self): + """An enabled pipeline admits its algorithms whether or not they have run. - ``Algo Configured Unused`` shares an enabled pipeline with ``Algo Used`` and is - excluded purely because it never classified anything, which pins the positive and - negative sides of the rule against an otherwise identical pair. + Running is not required because detectors never author a Classification, so a + results-only rule would drop the detector behind every detection in the project. """ names = self._list_algorithm_names(project_id=self.project.pk) - self.assertEqual(names, {"Algo Used"}) + self.assertEqual(names, {"Algo Used", "Algo Configured Unused"}) + + def test_detector_that_ran_is_listed_although_it_never_classified(self): + """Detectors set ``Detection.detection_algorithm`` and never write a + Classification, so they are reachable only through their pipeline. This pins the + regression where scoping the list purely by classification authorship dropped + every localizer from the project's algorithm list.""" + detector = Algorithm.objects.create(name="Algo Detector", version=1, task_type="localization") + pipeline = Pipeline.objects.get(name="Enabled Pipeline") + pipeline.algorithms.add(detector) + + source_image = SourceImage.objects.create(project=self.project) + Detection.objects.create(source_image=source_image, detection_algorithm=detector) + self.assertFalse(Classification.objects.filter(algorithm=detector).exists()) + + names = self._list_algorithm_names(project_id=self.project.pk) + self.assertIn("Algo Detector", names) def test_other_project_only_sees_its_own_algorithms(self): names = self._list_algorithm_names(project_id=self.other_project.pk) diff --git a/ami/ml/views.py b/ami/ml/views.py index f1dedb72d..d41213ad0 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -15,7 +15,7 @@ from ami.base.views import ProjectMixin from ami.main.api.schemas import project_id_doc_param from ami.main.api.views import DefaultViewSet -from ami.main.models import Classification, Project, SourceImage +from ami.main.models import Project, SourceImage from ami.ml.schemas import PipelineRegistrationResponse from .models.algorithm import Algorithm, AlgorithmCategoryMap @@ -61,18 +61,29 @@ def get_queryset(self) -> QuerySet["Algorithm"]: if getattr(self, "action", None) == "list": project = self.get_active_project() if project: - # An algorithm belongs to the project if it actually produced classifications - # there. Pipeline configuration is deliberately not consulted: a configured but - # never-run algorithm has no results to filter or inspect, and post-processing - # algorithms (e.g. class masking) are created standalone with no pipeline at all, - # so a pipeline join would both list algorithms with nothing behind them and hide - # ones that own live determinations. - classified_in_project = ( - Classification.objects.filter(detection__source_image__project=project) - .values_list("algorithm", flat=True) - .distinct() - ) - qs = qs.filter(pk__in=classified_in_project) + # Algorithms reach the list two ways. An enabled pipeline configures most of + # them, and that join alone covers detectors, which never author a + # Classification and so cannot be found by their results at all. + # + # Post-processing algorithms (e.g. class masking) are created standalone with + # no pipeline, so they are found by their classifications instead. That lookup + # is restricted to unpipelined algorithms first, which is what keeps it cheap: + # Classification has no project column and reaches one only through + # detection -> source_image, so an unrestricted version scans the whole table. + # + # The two are collected separately rather than OR'd into one filter. An OR + # across the pipeline join forces a SELECT DISTINCT whose COUNT costs more + # than both queries together. + configured_for_project = Algorithm.objects.filter( + pipelines__project_pipeline_configs__project=project, + pipelines__project_pipeline_configs__enabled=True, + ).values_list("pk", flat=True) + post_processing_used_in_project = Algorithm.objects.filter( + pipelines__isnull=True, + classifications__detection__source_image__project=project, + ).values_list("pk", flat=True) + # Materialising is safe here: algorithms number in the dozens platform-wide. + qs = qs.filter(pk__in=set(configured_for_project) | set(post_processing_used_in_project)) return qs @extend_schema(parameters=[project_id_doc_param]) From 48e7c2e7d8b400c7abbb4fedddf63509513dbb44 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:48:30 -0700 Subject: [PATCH 07/11] fix(algorithms): sort the algorithm id list so the query is cache-stable The ids came from a set, whose iteration order is not a documented guarantee. cachalot keys its cache on the generated query string, so a varying order of the same ids would produce cache misses for identical results. Also replaces the bound claimed for materialising the ids. "Dozens platform-wide" was true when written but class masking creates an algorithm per source algorithm, taxa list and reweight mode, so the count grows with use; the comment now says where the growth comes from and when to switch to a subquery. Co-Authored-By: Claude --- ami/ml/views.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/ami/ml/views.py b/ami/ml/views.py index d41213ad0..b8df25088 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -82,8 +82,16 @@ def get_queryset(self) -> QuerySet["Algorithm"]: pipelines__isnull=True, classifications__detection__source_image__project=project, ).values_list("pk", flat=True) - # Materialising is safe here: algorithms number in the dozens platform-wide. - qs = qs.filter(pk__in=set(configured_for_project) | set(post_processing_used_in_project)) + # Sorted so the generated SQL is stable for a given result: cachalot keys its + # cache on the query string, and an unordered set would vary it. + # + # Materialising the ids is fine at this cardinality. Algorithms are created per + # model rather than per run, and the one path that adds them over time is class + # masking, which creates a single algorithm per source algorithm, taxa list and + # reweight mode. Should that ever reach the thousands, this wants to become a + # subquery rather than an IN list. + relevant_ids = set(configured_for_project) | set(post_processing_used_in_project) + qs = qs.filter(pk__in=sorted(relevant_ids)) return qs @extend_schema(parameters=[project_id_doc_param]) From 7d67f5be25aa041d006a655d67383af6cdf6d8b6 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 14:56:04 -0700 Subject: [PATCH 08/11] fix(algorithms): deduplicate the post-processing lookup in the database MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The join from an algorithm to a project through its classifications emits one row per matching classification, so the lookup shipped one id per masked classification in the project — hundreds of thousands after a real masking run — to identify a handful of algorithms. Adding DISTINCT collapses that to one id per algorithm in the database. Measured effect is small on current data (the projects checked hold hundreds to low thousands of masked classifications), but the row count was linear in classification volume, which is exactly what this feature produces at scale. Also document that an algorithm on a pipeline not enabled for the project stays absent even when it has determinations there. Widening the second query to cover that case measured roughly five times the runtime, so it is left as a known gap matching the pipeline-join behaviour rather than folded into this change. Co-Authored-By: Claude --- ami/ml/tests.py | 22 ++++++++++++++++++++++ ami/ml/views.py | 24 ++++++++++++++++++------ 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 706f8f6b7..3fc03cb17 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2155,6 +2155,28 @@ def test_lists_post_processing_algorithm_with_classifications_in_project(self): names = self._list_algorithm_names(project_id=self.project.pk) self.assertIn("Class Masked Classifier", names) + def test_post_processing_lookup_is_deduplicated_in_the_database(self): + """The post-processing join emits one row per matching classification, so it must + deduplicate in SQL rather than in Python. + + Without that, the rows fetched grow with a project's masked classification count — + hundreds of thousands on a real masking run — to identify a handful of algorithms. + The listed names are correct either way, so this asserts the row count of the + underlying lookup rather than the endpoint's output. + """ + masked_algo = Algorithm.objects.create(name="Chatty Masked Classifier", version=1) + for _ in range(5): + self._classify_in_project(masked_algo, self.project) + + lookup = Algorithm.objects.filter( + pipelines__isnull=True, + classifications__detection__source_image__project=self.project, + ).values_list("pk", flat=True) + + self.assertEqual(len(list(lookup)), 5, "Undeduplicated join emits one row per classification") + self.assertEqual(len(list(lookup.distinct())), 1, "Deduplicating collapses them to the one algorithm") + self.assertIn("Chatty Masked Classifier", self._list_algorithm_names(project_id=self.project.pk)) + def test_classifications_in_other_project_do_not_leak(self): """An algorithm whose classifications live in another project must not appear.""" other_masked_algo = Algorithm.objects.create(name="Other Project Masked", version=1) diff --git a/ami/ml/views.py b/ami/ml/views.py index b8df25088..44f7896b7 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -66,11 +66,16 @@ def get_queryset(self) -> QuerySet["Algorithm"]: # Classification and so cannot be found by their results at all. # # Post-processing algorithms (e.g. class masking) are created standalone with - # no pipeline, so they are found by their classifications instead. That lookup - # is restricted to unpipelined algorithms first, which is what keeps it cheap: + # no pipeline, so they are found by their classifications instead. Restricting + # that lookup to unpipelined algorithms bounds which rows are scanned: # Classification has no project column and reaches one only through # detection -> source_image, so an unrestricted version scans the whole table. # + # An algorithm attached to a pipeline that is not enabled for this project is + # therefore absent even when it owns determinations here. That matches the + # behaviour on the pipeline join alone, and widening the second query to cover + # it costs roughly five times the runtime. + # # The two are collected separately rather than OR'd into one filter. An OR # across the pipeline join forces a SELECT DISTINCT whose COUNT costs more # than both queries together. @@ -78,10 +83,17 @@ def get_queryset(self) -> QuerySet["Algorithm"]: pipelines__project_pipeline_configs__project=project, pipelines__project_pipeline_configs__enabled=True, ).values_list("pk", flat=True) - post_processing_used_in_project = Algorithm.objects.filter( - pipelines__isnull=True, - classifications__detection__source_image__project=project, - ).values_list("pk", flat=True) + # Deduplicated in the database: the join emits one row per matching + # classification, so without this the result grows with a project's masked + # classification count rather than with its handful of algorithms. + post_processing_used_in_project = ( + Algorithm.objects.filter( + pipelines__isnull=True, + classifications__detection__source_image__project=project, + ) + .values_list("pk", flat=True) + .distinct() + ) # Sorted so the generated SQL is stable for a given result: cachalot keys its # cache on the query string, and an unordered set would vary it. # From e691c03534bda81df55505ed18f597c5f23cc4a1 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 15:00:47 -0700 Subject: [PATCH 09/11] perf(algorithms): look up post-processing algorithms via an indexed IN list The single-join form (pipelines__isnull=True combined with the classification relation) hides the candidate algorithm ids behind an anti-join at plan time, so Postgres sequentially scans the whole classification table to answer it. Its cost therefore grows with the platform's total classification count, and it is scanned cold on every cachalot invalidation, which happens on every classification write. Split it into two steps. Collect the unpipelined algorithm ids first, as an explicit list, then filter classifications on algorithm_id IN (...). The planner answers that from the algorithm_id index and touches only the rows for those algorithms. EXPLAIN on a local copy of production data confirms the plan change: a parallel sequential scan of main_classification (about 48 ms) becomes an index scan (about 6 ms), and the whole endpoint drops from roughly 46 ms to 12 ms. The first query is also cheap to cache, changing only when algorithms or pipeline links change rather than on every classification. Clear Classification's default ordering with .order_by() before .distinct(), otherwise the ordering columns widen the DISTINCT back to one row per classification. The dedup test now covers this. Co-Authored-By: Claude --- ami/ml/tests.py | 20 ++++++++++------- ami/ml/views.py | 58 ++++++++++++++++++++++++------------------------- 2 files changed, 41 insertions(+), 37 deletions(-) diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 3fc03cb17..73d91ee1d 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2156,25 +2156,29 @@ def test_lists_post_processing_algorithm_with_classifications_in_project(self): self.assertIn("Class Masked Classifier", names) def test_post_processing_lookup_is_deduplicated_in_the_database(self): - """The post-processing join emits one row per matching classification, so it must + """The post-processing lookup matches one row per classification, so it must deduplicate in SQL rather than in Python. Without that, the rows fetched grow with a project's masked classification count — hundreds of thousands on a real masking run — to identify a handful of algorithms. The listed names are correct either way, so this asserts the row count of the - underlying lookup rather than the endpoint's output. + underlying lookup rather than the endpoint's output. The ``.order_by()`` matters: + Classification's default ordering would otherwise widen the DISTINCT back to one + row per classification. """ masked_algo = Algorithm.objects.create(name="Chatty Masked Classifier", version=1) for _ in range(5): self._classify_in_project(masked_algo, self.project) - lookup = Algorithm.objects.filter( - pipelines__isnull=True, - classifications__detection__source_image__project=self.project, - ).values_list("pk", flat=True) + lookup = Classification.objects.filter( + algorithm_id=masked_algo.pk, + detection__source_image__project=self.project, + ).values_list("algorithm_id", flat=True) - self.assertEqual(len(list(lookup)), 5, "Undeduplicated join emits one row per classification") - self.assertEqual(len(list(lookup.distinct())), 1, "Deduplicating collapses them to the one algorithm") + self.assertEqual(len(list(lookup)), 5, "The lookup matches one row per classification") + self.assertEqual( + len(list(lookup.order_by().distinct())), 1, "Deduplicating collapses them to the one algorithm" + ) self.assertIn("Chatty Masked Classifier", self._list_algorithm_names(project_id=self.project.pk)) def test_classifications_in_other_project_do_not_leak(self): diff --git a/ami/ml/views.py b/ami/ml/views.py index 44f7896b7..9236b53be 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -15,7 +15,7 @@ from ami.base.views import ProjectMixin from ami.main.api.schemas import project_id_doc_param from ami.main.api.views import DefaultViewSet -from ami.main.models import Project, SourceImage +from ami.main.models import Classification, Project, SourceImage from ami.ml.schemas import PipelineRegistrationResponse from .models.algorithm import Algorithm, AlgorithmCategoryMap @@ -64,44 +64,44 @@ def get_queryset(self) -> QuerySet["Algorithm"]: # Algorithms reach the list two ways. An enabled pipeline configures most of # them, and that join alone covers detectors, which never author a # Classification and so cannot be found by their results at all. - # - # Post-processing algorithms (e.g. class masking) are created standalone with - # no pipeline, so they are found by their classifications instead. Restricting - # that lookup to unpipelined algorithms bounds which rows are scanned: - # Classification has no project column and reaches one only through - # detection -> source_image, so an unrestricted version scans the whole table. - # - # An algorithm attached to a pipeline that is not enabled for this project is - # therefore absent even when it owns determinations here. That matches the - # behaviour on the pipeline join alone, and widening the second query to cover - # it costs roughly five times the runtime. - # - # The two are collected separately rather than OR'd into one filter. An OR - # across the pipeline join forces a SELECT DISTINCT whose COUNT costs more - # than both queries together. configured_for_project = Algorithm.objects.filter( pipelines__project_pipeline_configs__project=project, pipelines__project_pipeline_configs__enabled=True, ).values_list("pk", flat=True) - # Deduplicated in the database: the join emits one row per matching - # classification, so without this the result grows with a project's masked - # classification count rather than with its handful of algorithms. + + # Post-processing algorithms (e.g. class masking) are created standalone with + # no pipeline, so they are found by their classifications instead. This is done + # in two steps on purpose. Collecting the unpipelined algorithm ids first, as an + # explicit list, lets the classification lookup filter on `algorithm_id IN (...)`, + # which the planner answers from the algorithm_id index. Expressing the same thing + # as a single join (`pipelines__isnull=True, classifications__...`) hides those ids + # behind an anti-join at plan time, so the planner sequentially scans the whole + # classification table instead — a cost that grows with the platform's total + # classification count rather than with this project. The first query is also + # cheap to cache: it only changes when algorithms or pipeline links change, not on + # every classification write. + # + # An algorithm attached to a pipeline that is not enabled for this project is + # absent even when it owns determinations here. That matches the behaviour on the + # pipeline join alone; widening the second lookup to cover it costs several times + # the runtime and is left as a follow-up. + unpipelined_algorithm_ids = list( + Algorithm.objects.filter(pipelines__isnull=True).values_list("pk", flat=True) + ) + # `.order_by()` clears Classification's default ordering before `.distinct()`. + # Without it the ordering columns join the SELECT and widen the DISTINCT, so it + # would return one row per classification rather than one per algorithm. post_processing_used_in_project = ( - Algorithm.objects.filter( - pipelines__isnull=True, - classifications__detection__source_image__project=project, + Classification.objects.filter( + algorithm_id__in=unpipelined_algorithm_ids, + detection__source_image__project=project, ) - .values_list("pk", flat=True) + .order_by() + .values_list("algorithm_id", flat=True) .distinct() ) # Sorted so the generated SQL is stable for a given result: cachalot keys its # cache on the query string, and an unordered set would vary it. - # - # Materialising the ids is fine at this cardinality. Algorithms are created per - # model rather than per run, and the one path that adds them over time is class - # masking, which creates a single algorithm per source algorithm, taxa list and - # reweight mode. Should that ever reach the thousands, this wants to become a - # subquery rather than an IN list. relevant_ids = set(configured_for_project) | set(post_processing_used_in_project) qs = qs.filter(pk__in=sorted(relevant_ids)) return qs From 71dbd65e45d1848902a26f4ad90300261c556724 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 15:06:32 -0700 Subject: [PATCH 10/11] fix(algorithms): sort the inner algorithm id list for a stable cache key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unpipelined algorithm ids are rendered verbatim into the classification lookup's IN clause. Algorithm orders by (name, version), so the ids came out in name order, and identical membership could produce a different id order — and so a different SQL string and cachalot cache key — across requests. Sorting makes the inner query cache-stable, matching the outer id list, which was already sorted for the same reason. Co-Authored-By: Claude --- ami/ml/views.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/ami/ml/views.py b/ami/ml/views.py index 9236b53be..87ff46933 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -85,7 +85,11 @@ def get_queryset(self) -> QuerySet["Algorithm"]: # absent even when it owns determinations here. That matches the behaviour on the # pipeline join alone; widening the second lookup to cover it costs several times # the runtime and is left as a follow-up. - unpipelined_algorithm_ids = list( + # Sorted for the same reason as the outer id list below: this list is rendered + # verbatim into the IN clause, and Algorithm's (name, version) ordering would + # otherwise vary the id order — and so the SQL string and its cachalot key — + # for identical membership. + unpipelined_algorithm_ids = sorted( Algorithm.objects.filter(pipelines__isnull=True).values_list("pk", flat=True) ) # `.order_by()` clears Classification's default ordering before `.distinct()`. From 996277d09ea4c259b03044a4b05445b198d03875 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Tue, 21 Jul 2026 16:05:18 -0700 Subject: [PATCH 11/11] refactor(algorithms): scope project list to algorithms that actually ran MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the pipeline-config + unpipelined-classification hybrid in AlgorithmViewSet with a single reusable queryset method, Algorithm.objects.used_in_project(). The list now contains exactly the algorithms that produced results in the project — detectors via their detections, classifiers and post-processing algorithms via their classifications — regardless of whether the owning pipeline is still enabled. A superseded model version stays listed as long as its determinations survive; an algorithm configured on an enabled pipeline that never ran no longer appears. Add matching OccurrenceQuerySet methods, detected_or_classified_by() and not_detected_or_classified_by(), and route OccurrenceAlgorithmFilter through them. The filter previously joined detections__classifications, which returned one row per matching classification and inflated the paginator's COUNT, and it matched only classifiers — a detector produced no match at all. The EXISTS form counts each occurrence once and matches both roles, so every algorithm the list shows can now filter occurrences. Document in AGENTS.md the two query-work rules this change was built on: run EXPLAIN (ANALYZE) before claiming a query's plan, and measure on the largest project (surveying several sizes) rather than a single small one. Cost note: used_in_project scans classification and detection because neither table has a project column, so it runs ~0.1-0.6s cold and scales with total table size until a denormalized project column exists — the honest cost of showing exactly what ran. Co-Authored-By: Claude --- .agents/AGENTS.md | 2 + ami/main/api/views.py | 14 +++-- ami/main/models.py | 27 ++++++++++ ami/main/tests.py | 102 +++++++++++++++++++++++++++++++++++++ ami/ml/models/algorithm.py | 40 +++++++++++++++ ami/ml/tests.py | 77 +++++++++++++++++----------- ami/ml/views.py | 55 +++----------------- 7 files changed, 233 insertions(+), 84 deletions(-) diff --git a/.agents/AGENTS.md b/.agents/AGENTS.md index 458ae513b..710d72148 100644 --- a/.agents/AGENTS.md +++ b/.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 diff --git a/ami/main/api/views.py b/ami/main/api/views.py index e245044a8..b499002b9 100644 --- a/ami/main/api/views.py +++ b/ami/main/api/views.py @@ -1222,11 +1222,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 +1241,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.detected_or_classified_by(algorithm_ids) if algorithm_ids_exclusive: - queryset = queryset.exclude(detections__classifications__algorithm__in=algorithm_ids_exclusive) + queryset = queryset.not_detected_or_classified_by(algorithm_ids_exclusive) return queryset diff --git a/ami/main/models.py b/ami/main/models.py index 3662b4107..017b51263 100644 --- a/ami/main/models.py +++ b/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 _machine_results_by(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).""" + 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 detected_or_classified_by(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._machine_results_by(algorithm_ids)) + + def not_detected_or_classified_by(self, algorithm_ids) -> "OccurrenceQuerySet": + """Occurrences with no result from any of the given algorithms.""" + return self.exclude(self._machine_results_by(algorithm_ids)) + def with_timestamps(self): """ These are timestamps used for filtering and ordering in the UI. diff --git a/ami/main/tests.py b/ami/main/tests.py index a0d0253a1..fb8bb771f 100644 --- a/ami/main/tests.py +++ b/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.detected_or_classified_by / not_detected_or_classified_by, + 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).detected_or_classified_by([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).detected_or_classified_by([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).detected_or_classified_by([self.classifier.pk]) + self.assertEqual(matched.count(), 2) + + def test_exclude_removes_matching_occurrences(self): + remaining = Occurrence.objects.filter(project=self.project).not_detected_or_classified_by([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.detected_or_classified_by(ids)) + excluded = self._pks(base.not_detected_or_classified_by(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. diff --git a/ami/ml/models/algorithm.py b/ami/ml/models/algorithm.py index 0e9df4609..605f90861 100644 --- a/ami/ml/models/algorithm.py +++ b/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): diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 73d91ee1d..bc5f65a9e 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -2037,13 +2037,15 @@ def test_deployment_counts_refresh_after_save_results(self): class TestAlgorithmViewSetProjectFilter(APITestCase): """ - The algorithm list endpoint is scoped to the algorithms relevant to the active - project, which they reach two ways. - - An enabled pipeline configures most of them. That covers detectors, which never - author a Classification and so could not be found by their results. Post-processing - algorithms such as class masking have no pipeline at all, so they are admitted by - having classified something in the project. + The algorithm list endpoint is scoped to the algorithms that actually produced + results in the active project, regardless of pipeline configuration. + + An algorithm qualifies by owning output rows: a detection made by it (detectors, + which never author a Classification) or a classification from it (classifiers and + standalone post-processing algorithms such as class masking). A superseded pipeline + version stays listed as long as its results survive, and an algorithm configured on + an enabled pipeline that has never run does not appear at all — the list reflects + what happened, not what is set up. """ def setUp(self): @@ -2053,14 +2055,14 @@ def setUp(self): self.project = Project.objects.create(name="Algo Project A", create_defaults=False) self.other_project = Project.objects.create(name="Algo Project B", create_defaults=False) - # Project A: an algorithm that has run, and one configured via a disabled pipeline + # Project A: an enabled-pipeline algorithm that has run, and one that never did. self.algo_used = Algorithm.objects.create(name="Algo Used", version=1) - self.algo_disabled = Algorithm.objects.create(name="Algo Disabled", version=1) - # Project A: enabled pipeline, but the algorithm never produced anything self.algo_configured_unused = Algorithm.objects.create(name="Algo Configured Unused", version=1) - # Project B: a different algorithm, also used + # Project A: an old version on a now-disabled pipeline, whose determinations survive. + self.algo_superseded = Algorithm.objects.create(name="Algo Superseded", version=1) + # Project B: a different algorithm, also used. self.algo_other_project = Algorithm.objects.create(name="Algo Other Project", version=1) - # Unrelated algorithm not attached to any pipeline and never run + # Unrelated algorithm attached to no pipeline and never run. self.algo_orphan = Algorithm.objects.create(name="Algo Orphan", version=1) enabled_pipeline = Pipeline.objects.create(name="Enabled Pipeline") @@ -2068,7 +2070,7 @@ def setUp(self): ProjectPipelineConfig.objects.create(project=self.project, pipeline=enabled_pipeline, enabled=True) disabled_pipeline = Pipeline.objects.create(name="Disabled Pipeline") - disabled_pipeline.algorithms.add(self.algo_disabled) + disabled_pipeline.algorithms.add(self.algo_superseded) ProjectPipelineConfig.objects.create(project=self.project, pipeline=disabled_pipeline, enabled=False) other_pipeline = Pipeline.objects.create(name="Other Project Pipeline") @@ -2076,6 +2078,7 @@ def setUp(self): ProjectPipelineConfig.objects.create(project=self.other_project, pipeline=other_pipeline, enabled=True) self._classify_in_project(self.algo_used, self.project) + self._classify_in_project(self.algo_superseded, self.project) self._classify_in_project(self.algo_other_project, self.other_project) self.client.force_authenticate(user=self.user) @@ -2097,23 +2100,33 @@ def _list_algorithm_names(self, project_id=None): self.assertEqual(response.status_code, 200) return {row["name"] for row in response.json()["results"]} - def test_lists_enabled_pipeline_algorithms_for_project(self): - """An enabled pipeline admits its algorithms whether or not they have run. + def test_lists_only_algorithms_that_produced_results(self): + """The project list is exactly the algorithms with output here: the classifier + that ran and the superseded version whose determinations survive. The enabled + pipeline's never-run algorithm is absent, proving configuration alone does not + admit an algorithm.""" + names = self._list_algorithm_names(project_id=self.project.pk) + self.assertEqual(names, {"Algo Used", "Algo Superseded"}) - Running is not required because detectors never author a Classification, so a - results-only rule would drop the detector behind every detection in the project. - """ + def test_configured_but_never_run_algorithm_is_hidden(self): + """An algorithm on an enabled pipeline that never produced a result does not + appear. This pins the semantic that the list follows results, not setup.""" + names = self._list_algorithm_names(project_id=self.project.pk) + self.assertNotIn("Algo Configured Unused", names) + + def test_superseded_pipeline_version_with_results_is_listed(self): + """An algorithm whose pipeline is disabled for the project — an older model + version, for instance — stays listed as long as its determinations exist, so the + user can still filter occurrences by what an earlier run produced.""" names = self._list_algorithm_names(project_id=self.project.pk) - self.assertEqual(names, {"Algo Used", "Algo Configured Unused"}) + self.assertIn("Algo Superseded", names) def test_detector_that_ran_is_listed_although_it_never_classified(self): """Detectors set ``Detection.detection_algorithm`` and never write a - Classification, so they are reachable only through their pipeline. This pins the - regression where scoping the list purely by classification authorship dropped + Classification, so they are reachable only through their detections. This pins + the regression where scoping the list purely by classification authorship dropped every localizer from the project's algorithm list.""" detector = Algorithm.objects.create(name="Algo Detector", version=1, task_type="localization") - pipeline = Pipeline.objects.get(name="Enabled Pipeline") - pipeline.algorithms.add(detector) source_image = SourceImage.objects.create(project=self.project) Detection.objects.create(source_image=source_image, detection_algorithm=detector) @@ -2130,20 +2143,22 @@ def test_unscoped_request_returns_all_algorithms(self): """Without project_id, current behavior lists all algorithms (unchanged).""" names = self._list_algorithm_names() self.assertIn("Algo Used", names) - self.assertIn("Algo Disabled", names) + self.assertIn("Algo Superseded", names) + self.assertIn("Algo Configured Unused", names) self.assertIn("Algo Other Project", names) self.assertIn("Algo Orphan", names) def test_detail_endpoint_unscoped_even_with_project_id(self): - """Detail stays unscoped so historical classification links still resolve.""" + """Detail stays unscoped so links to an algorithm outside the project's used + set — here an orphan that never ran — still resolve.""" url = reverse_with_params( "api:algorithm-detail", - kwargs={"pk": self.algo_disabled.pk}, + kwargs={"pk": self.algo_orphan.pk}, params={"project_id": self.project.pk}, ) response = self.client.get(url) self.assertEqual(response.status_code, 200) - self.assertEqual(response.json()["name"], "Algo Disabled") + self.assertEqual(response.json()["name"], "Algo Orphan") def test_lists_post_processing_algorithm_with_classifications_in_project(self): """A post-processing algorithm has no pipeline but produces determinations in @@ -2155,11 +2170,11 @@ def test_lists_post_processing_algorithm_with_classifications_in_project(self): names = self._list_algorithm_names(project_id=self.project.pk) self.assertIn("Class Masked Classifier", names) - def test_post_processing_lookup_is_deduplicated_in_the_database(self): - """The post-processing lookup matches one row per classification, so it must - deduplicate in SQL rather than in Python. + def test_used_lookup_is_deduplicated_in_the_database(self): + """``used_in_project`` matches one classification row per determination, so it + must deduplicate in SQL rather than in Python. - Without that, the rows fetched grow with a project's masked classification count — + Without that, the rows fetched grow with a project's classification count — hundreds of thousands on a real masking run — to identify a handful of algorithms. The listed names are correct either way, so this asserts the row count of the underlying lookup rather than the endpoint's output. The ``.order_by()`` matters: diff --git a/ami/ml/views.py b/ami/ml/views.py index 87ff46933..c0f626277 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -15,7 +15,7 @@ from ami.base.views import ProjectMixin from ami.main.api.schemas import project_id_doc_param from ami.main.api.views import DefaultViewSet -from ami.main.models import Classification, Project, SourceImage +from ami.main.models import Project, SourceImage from ami.ml.schemas import PipelineRegistrationResponse from .models.algorithm import Algorithm, AlgorithmCategoryMap @@ -61,53 +61,12 @@ def get_queryset(self) -> QuerySet["Algorithm"]: if getattr(self, "action", None) == "list": project = self.get_active_project() if project: - # Algorithms reach the list two ways. An enabled pipeline configures most of - # them, and that join alone covers detectors, which never author a - # Classification and so cannot be found by their results at all. - configured_for_project = Algorithm.objects.filter( - pipelines__project_pipeline_configs__project=project, - pipelines__project_pipeline_configs__enabled=True, - ).values_list("pk", flat=True) - - # Post-processing algorithms (e.g. class masking) are created standalone with - # no pipeline, so they are found by their classifications instead. This is done - # in two steps on purpose. Collecting the unpipelined algorithm ids first, as an - # explicit list, lets the classification lookup filter on `algorithm_id IN (...)`, - # which the planner answers from the algorithm_id index. Expressing the same thing - # as a single join (`pipelines__isnull=True, classifications__...`) hides those ids - # behind an anti-join at plan time, so the planner sequentially scans the whole - # classification table instead — a cost that grows with the platform's total - # classification count rather than with this project. The first query is also - # cheap to cache: it only changes when algorithms or pipeline links change, not on - # every classification write. - # - # An algorithm attached to a pipeline that is not enabled for this project is - # absent even when it owns determinations here. That matches the behaviour on the - # pipeline join alone; widening the second lookup to cover it costs several times - # the runtime and is left as a follow-up. - # Sorted for the same reason as the outer id list below: this list is rendered - # verbatim into the IN clause, and Algorithm's (name, version) ordering would - # otherwise vary the id order — and so the SQL string and its cachalot key — - # for identical membership. - unpipelined_algorithm_ids = sorted( - Algorithm.objects.filter(pipelines__isnull=True).values_list("pk", flat=True) - ) - # `.order_by()` clears Classification's default ordering before `.distinct()`. - # Without it the ordering columns join the SELECT and widen the DISTINCT, so it - # would return one row per classification rather than one per algorithm. - post_processing_used_in_project = ( - Classification.objects.filter( - algorithm_id__in=unpipelined_algorithm_ids, - detection__source_image__project=project, - ) - .order_by() - .values_list("algorithm_id", flat=True) - .distinct() - ) - # Sorted so the generated SQL is stable for a given result: cachalot keys its - # cache on the query string, and an unordered set would vary it. - relevant_ids = set(configured_for_project) | set(post_processing_used_in_project) - qs = qs.filter(pk__in=sorted(relevant_ids)) + # Scope the list to algorithms that actually produced results in the project — + # any superseded pipeline version or standalone post-processing algorithm whose + # output still exists, so the user can filter occurrences by anything that ran. + # Pipeline configuration is not consulted; configured-but-never-run algorithms + # do not appear. See the method for cost characteristics. + qs = qs.used_in_project(project) # type: ignore[union-attr] # Custom queryset method return qs @extend_schema(parameters=[project_id_doc_param])