diff --git a/ami/main/admin.py b/ami/main/admin.py
index 404325a93..b25860507 100644
--- a/ami/main/admin.py
+++ b/ami/main/admin.py
@@ -14,7 +14,9 @@
 from ami.jobs.models import Job
 from ami.ml.models.project_pipeline_config import ProjectPipelineConfig
 from ami.ml.post_processing.admin.actions import make_post_processing_action
+from ami.ml.post_processing.admin.class_masking_form import ClassMaskingActionForm
 from ami.ml.post_processing.admin.small_size_filter_form import SmallSizeFilterActionForm
+from ami.ml.post_processing.class_masking import ClassMaskingTask
 from ami.ml.post_processing.small_size_filter import SmallSizeFilterTask
 from ami.ml.tasks import remove_duplicate_classifications
 
@@ -552,6 +554,12 @@ def detections_count(self, obj) -> int:
         scope_resolver=lambda occurrence: {"occurrence_id": occurrence.pk},
         name_resolver=lambda task_cls, occurrence: (f"Post-processing: {task_cls.name} on Occurrence {occurrence.pk}"),
     )
+    run_class_masking = make_post_processing_action(
+        ClassMaskingTask,
+        ClassMaskingActionForm,
+        scope_resolver=lambda occurrence: {"occurrence_id": occurrence.pk},
+        name_resolver=lambda task_cls, occurrence: (f"Post-processing: {task_cls.name} on Occurrence {occurrence.pk}"),
+    )
 
     @admin.action(description="Recompute determination from current classifications and identifications")
     def recompute_determination(self, request: HttpRequest, queryset: QuerySet[Any]) -> None:
@@ -568,7 +576,7 @@ def recompute_determination(self, request: HttpRequest, queryset: QuerySet[Any])
             count += 1
         self.message_user(request, f"Recomputed determination for {count} occurrence(s).")
 
-    actions = [run_small_size_filter, recompute_determination]
+    actions = [run_small_size_filter, run_class_masking, recompute_determination]
 
     # Order by -id (the indexed primary key) rather than -created_at, which has no
     # index and would force a full sort of the table to find the newest page. id
@@ -850,11 +858,19 @@ def populate_collection_async(self, request: HttpRequest, queryset: QuerySet[Sou
             f"Post-processing: {task_cls.name} on Capture Set {collection.pk}"
         ),
     )
-
+    run_class_masking = make_post_processing_action(
+        ClassMaskingTask,
+        ClassMaskingActionForm,
+        scope_resolver=lambda collection: {"source_image_collection_id": collection.pk},
+        name_resolver=lambda task_cls, collection: (
+            f"Post-processing: {task_cls.name} on Capture Set {collection.pk}"
+        ),
+    )
     actions = [
         populate_collection,
         populate_collection_async,
         run_small_size_filter,
+        run_class_masking,
     ]
 
     # Hide images many-to-many field from form. This would list all source images in the database.
diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py
index 2855633e3..6c899c2f7 100644
--- a/ami/main/api/serializers.py
+++ b/ami/main/api/serializers.py
@@ -941,10 +941,26 @@ class ClassificationPredictionItemSerializer(serializers.Serializer):
     logit = serializers.FloatField(read_only=True)
 
 
+class ClassificationAppliedToSerializer(serializers.ModelSerializer):
+    """Lightweight nested representation of the parent classification this was derived from.
+
+    Post-processing tasks (class masking, rank rollup) record provenance via
+    ``Classification.applied_to``; this exposes just enough to show what a result
+    was derived from without recursing back into the full classification.
+    """
+
+    algorithm = AlgorithmSerializer(read_only=True)
+
+    class Meta:
+        model = Classification
+        fields = ["id", "created_at", "algorithm"]
+
+
 class ClassificationSerializer(DefaultSerializer):
     taxon = TaxonNestedSerializer(read_only=True)
     algorithm = AlgorithmSerializer(read_only=True)
     top_n = ClassificationPredictionItemSerializer(many=True, read_only=True)
+    applied_to = ClassificationAppliedToSerializer(read_only=True)
 
     class Meta:
         model = Classification
@@ -957,6 +973,7 @@ class Meta:
             "scores",
             "logits",
             "top_n",
+            "applied_to",
             "created_at",
             "updated_at",
         ]
@@ -979,6 +996,8 @@ class Meta(ClassificationSerializer.Meta):
 
 
 class ClassificationListSerializer(DefaultSerializer):
+    applied_to = ClassificationAppliedToSerializer(read_only=True)
+
     class Meta:
         model = Classification
         fields = [
@@ -987,6 +1006,7 @@ class Meta:
             "taxon",
             "score",
             "algorithm",
+            "applied_to",
             "created_at",
             "updated_at",
         ]
@@ -1006,6 +1026,7 @@ class Meta:
             "score",
             "terminal",
             "algorithm",
+            "applied_to",
             "created_at",
         ]
 
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index 6ad39a2e5..591a4a000 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -2060,7 +2060,7 @@ class ClassificationViewSet(DefaultViewSet, ProjectMixin):
     """
 
     require_project_for_list = True  # Unfiltered list scans are too expensive on this table
-    queryset = Classification.objects.all().select_related("taxon", "algorithm")  # , "detection")
+    queryset = Classification.objects.all().select_related("taxon", "algorithm", "applied_to__algorithm")
     serializer_class = ClassificationSerializer
     filterset_fields = [
         # Docs about slow loading API browser because of large choice fields
diff --git a/ami/main/models_future/occurrence.py b/ami/main/models_future/occurrence.py
index 012554169..ad5cc0ac6 100644
--- a/ami/main/models_future/occurrence.py
+++ b/ami/main/models_future/occurrence.py
@@ -58,7 +58,10 @@ def _detections_prefetch(*, ordering: tuple[str, ...], with_source_image: bool)
     qs = Detection.objects.prefetch_related(
         Prefetch(
             "classifications",
-            queryset=Classification.objects.select_related("taxon", "algorithm"),
+            # applied_to__algorithm: post-processed classifications (class masking,
+            # rank rollup) serialize their provenance parent; pull it here so the
+            # nested applied_to render doesn't issue a query per classification.
+            queryset=Classification.objects.select_related("taxon", "algorithm", "applied_to__algorithm"),
         )
     ).order_by(*ordering)
     if with_source_image:
diff --git a/ami/ml/management/commands/run_class_masking.py b/ami/ml/management/commands/run_class_masking.py
new file mode 100644
index 000000000..940956c6e
--- /dev/null
+++ b/ami/ml/management/commands/run_class_masking.py
@@ -0,0 +1,84 @@
+from django.core.management.base import BaseCommand, CommandError
+
+from ami.main.models import SourceImageCollection, TaxaList
+from ami.ml.models.algorithm import Algorithm
+from ami.ml.post_processing.class_masking import ClassMaskingTask
+
+
+class Command(BaseCommand):
+    help = (
+        "Run class masking post-processing on a source image collection. "
+        "Masks classifier logits for species not in the given taxa list and recalculates softmax scores."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument("--collection-id", type=int, required=True, help="SourceImageCollection ID to process")
+        parser.add_argument("--taxa-list-id", type=int, required=True, help="TaxaList ID to use as the species mask")
+        parser.add_argument(
+            "--algorithm-id", type=int, required=True, help="Algorithm ID whose classifications to mask"
+        )
+        parser.add_argument("--dry-run", action="store_true", help="Show what would be done without making changes")
+
+    def handle(self, *args, **options):
+        collection_id = options["collection_id"]
+        taxa_list_id = options["taxa_list_id"]
+        algorithm_id = options["algorithm_id"]
+        dry_run = options["dry_run"]
+
+        # Validate inputs
+        try:
+            collection = SourceImageCollection.objects.get(pk=collection_id)
+        except SourceImageCollection.DoesNotExist:
+            raise CommandError(f"SourceImageCollection {collection_id} does not exist.")
+
+        try:
+            taxa_list = TaxaList.objects.get(pk=taxa_list_id)
+        except TaxaList.DoesNotExist:
+            raise CommandError(f"TaxaList {taxa_list_id} does not exist.")
+
+        try:
+            algorithm = Algorithm.objects.get(pk=algorithm_id)
+        except Algorithm.DoesNotExist:
+            raise CommandError(f"Algorithm {algorithm_id} does not exist.")
+
+        if not algorithm.category_map:
+            raise CommandError(f"Algorithm '{algorithm.name}' does not have a category map.")
+
+        from ami.main.models import Classification
+
+        classification_count = (
+            Classification.objects.filter(
+                detection__source_image__collections=collection,
+                terminal=True,
+                algorithm=algorithm,
+                scores__isnull=False,
+                logits__isnull=False,
+            )
+            .distinct()
+            .count()
+        )
+
+        taxa_count = taxa_list.taxa.count()
+
+        self.stdout.write(
+            f"Collection:     {collection.name} (id={collection.pk})\n"
+            f"Taxa list:      {taxa_list.name} (id={taxa_list.pk}, {taxa_count} taxa)\n"
+            f"Algorithm:      {algorithm.name} (id={algorithm.pk})\n"
+            f"Classifications to process: {classification_count}"
+        )
+
+        if classification_count == 0:
+            raise CommandError("No terminal classifications with scores found for this collection/algorithm.")
+
+        if dry_run:
+            self.stdout.write(self.style.WARNING("Dry run — no changes made."))
+            return
+
+        self.stdout.write("Running class masking...")
+        task = ClassMaskingTask(
+            source_image_collection_id=collection_id,
+            taxa_list_id=taxa_list_id,
+            algorithm_id=algorithm_id,
+        )
+        task.run()
+        self.stdout.write(self.style.SUCCESS("Class masking completed."))
diff --git a/ami/ml/post_processing/__init__.py b/ami/ml/post_processing/__init__.py
index 3517ed47c..c94be9ae9 100644
--- a/ami/ml/post_processing/__init__.py
+++ b/ami/ml/post_processing/__init__.py
@@ -1 +1,2 @@
+from . import class_masking  # noqa: F401
 from . import small_size_filter  # noqa: F401
diff --git a/ami/ml/post_processing/admin/actions.py b/ami/ml/post_processing/admin/actions.py
index 07a299c1f..4e7729256 100644
--- a/ami/ml/post_processing/admin/actions.py
+++ b/ami/ml/post_processing/admin/actions.py
@@ -247,10 +247,12 @@ def _render(form: BasePostProcessingActionForm) -> TemplateResponse:
             )
             return None
 
+        # Hand the form the selected rows so it can scope its fields to the
+        # selection (e.g. only offer algorithms that ran on the chosen occurrence).
         if not request.POST.get("confirm"):
-            return _render(form_class())
+            return _render(form_class(scope_queryset=queryset))
 
-        form = form_class(request.POST)
+        form = form_class(request.POST, scope_queryset=queryset)
         if not form.is_valid():
             return _render(form)
 
diff --git a/ami/ml/post_processing/admin/class_masking_form.py b/ami/ml/post_processing/admin/class_masking_form.py
new file mode 100644
index 000000000..6e2f5f884
--- /dev/null
+++ b/ami/ml/post_processing/admin/class_masking_form.py
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from django import forms
+
+from ami.main.models import Occurrence, TaxaList
+from ami.ml.models import Algorithm
+from ami.ml.models.algorithm import AlgorithmTaskType
+from ami.ml.post_processing.admin.forms import BasePostProcessingActionForm
+
+
+class ClassMaskingActionForm(BasePostProcessingActionForm):
+    """Knobs surfaced when an admin triggers Class masking.
+
+    The operator picks the source classifier and the taxa list to keep; the
+    scope (which collection or occurrence) is supplied by the admin entry point,
+    not the form. Selections are model instances, so ``to_config`` hands the
+    schema their primary keys (``ClassMaskingConfig`` expects ``*_id`` ints).
+    """
+
+    algorithm_id = forms.ModelChoiceField(
+        queryset=Algorithm.objects.filter(task_type=AlgorithmTaskType.CLASSIFICATION.value).order_by("name"),
+        label="Source classifier",
+        help_text="The classification algorithm whose terminal predictions will be re-scored.",
+    )
+    taxa_list_id = forms.ModelChoiceField(
+        queryset=TaxaList.objects.all().order_by("name"),
+        label="Taxa list to keep",
+        help_text=(
+            "Classes whose taxon is not in this list are masked out; each "
+            "classification's softmax is renormalised over the classes that remain."
+        ),
+    )
+    reweight = forms.BooleanField(
+        required=False,
+        initial=True,
+        label="Reweight (renormalise) scores",
+        help_text=(
+            "Renormalise the kept classes to sum to 1. "
+            "Off = keep the model's raw absolute scores; the chosen species is unchanged either way."
+        ),
+    )
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # Narrow the classifier dropdown to algorithms that actually produced
+        # classifications in the selected scope, so the operator cannot pick a
+        # classifier whose masking would be a no-op for the chosen rows. This is
+        # only done for an occurrence scope, where the lookup touches the handful
+        # of classifications under the picked occurrences. A collection scope
+        # keeps the full classifier list on purpose: the equivalent lookup is an
+        # unbounded DISTINCT over every classification in the collection (hundreds
+        # of thousands of rows on a large collection) and can time out while the
+        # form renders. An over-broad option is harmless — masking a classifier
+        # that produced nothing in scope changes nothing.
+        if self.scope_queryset is not None and self.scope_queryset.model is Occurrence:
+            self.fields["algorithm_id"].queryset = self._algorithms_for_scope(self.scope_queryset)
+
+    @staticmethod
+    def _algorithms_for_scope(scope_queryset):
+        """Classification algorithms that produced classifications within the
+        selected occurrences."""
+        return (
+            Algorithm.objects.filter(
+                task_type=AlgorithmTaskType.CLASSIFICATION.value,
+                classifications__detection__occurrence__in=scope_queryset,
+            )
+            .distinct()
+            .order_by("name")
+        )
+
+    def to_config(self) -> dict:
+        return {
+            "algorithm_id": self.cleaned_data["algorithm_id"].pk,
+            "taxa_list_id": self.cleaned_data["taxa_list_id"].pk,
+            "reweight": self.cleaned_data["reweight"],
+        }
diff --git a/ami/ml/post_processing/admin/forms.py b/ami/ml/post_processing/admin/forms.py
index 7a2dbf5c9..c9c162808 100644
--- a/ami/ml/post_processing/admin/forms.py
+++ b/ami/ml/post_processing/admin/forms.py
@@ -20,6 +20,16 @@ class BasePostProcessingActionForm(forms.Form):
     optional fields, derive computed values, rename keys).
     """
 
+    def __init__(self, *args, scope_queryset=None, **kwargs):
+        """Capture the admin selection the action will run on.
+
+        ``scope_queryset`` is the queryset of rows the operator picked (e.g. the
+        chosen occurrences or collections). Subclasses may use it to constrain
+        their fields to that selection; forms that don't need it ignore it.
+        """
+        self.scope_queryset = scope_queryset
+        super().__init__(*args, **kwargs)
+
     def to_config(self) -> dict:
         """Return ``cleaned_data`` shaped for ``Job.params['config']``."""
         return dict(self.cleaned_data)
diff --git a/ami/ml/post_processing/base.py b/ami/ml/post_processing/base.py
index 863312344..8f197f192 100644
--- a/ami/ml/post_processing/base.py
+++ b/ami/ml/post_processing/base.py
@@ -75,7 +75,11 @@ def update_progress(self, progress: float):
 
         if self.job:
             self.job.progress.update_stage(self.job.job_type_key, progress=progress)
-            self.job.save(update_fields=["progress"])
+            # Bump updated_at alongside progress: the stale-job reaper
+            # (check_stale_jobs) revokes running jobs whose updated_at is older
+            # than STALLED_JOBS_MAX_MINUTES. A long post-processing run that only
+            # touched "progress" would look frozen and be reaped mid-flight.
+            self.job.save(update_fields=["progress", "updated_at"])
 
         else:
             # No job object — fallback to plain logging
@@ -99,7 +103,9 @@ def report_stage_metrics(self, metrics: dict[str, Any]):
         stage_key = self.job.job_type_key
         for label, value in metrics.items():
             self.job.progress.add_or_update_stage_param(stage_key, label, value)
-        self.job.save(update_fields=["progress"])
+        # Bump updated_at so the stale-job reaper sees an actively-progressing
+        # run; see update_progress for the full reasoning.
+        self.job.save(update_fields=["progress", "updated_at"])
 
     @abc.abstractmethod
     def run(self) -> None:
diff --git a/ami/ml/post_processing/class_masking.py b/ami/ml/post_processing/class_masking.py
new file mode 100644
index 000000000..ffc77f346
--- /dev/null
+++ b/ami/ml/post_processing/class_masking.py
@@ -0,0 +1,339 @@
+import logging
+from collections.abc import Callable
+
+import numpy as np
+import pydantic
+from django.db import transaction
+from django.db.models import QuerySet
+from django.utils import timezone
+
+from ami.main.models import Classification, Occurrence, SourceImageCollection, TaxaList
+from ami.ml.models.algorithm import Algorithm, AlgorithmTaskType
+from ami.ml.post_processing.base import BasePostProcessingTask
+
+logger = logging.getLogger(__name__)
+
+
+class ClassMaskingConfig(pydantic.BaseModel):
+    # Scope: exactly one of these identifies which classifications to re-score. A
+    # capture set is the bulk path; a single occurrence is the spot/dev path (fast
+    # feedback while tuning a taxa list). This mirrors SmallSizeFilterConfig's
+    # discriminated-scope shape — the shared pattern for per-occurrence triggers.
+    source_image_collection_id: int | None = None
+    occurrence_id: int | None = None
+    # The taxa list to keep: classes whose taxon is not in this list are masked out.
+    taxa_list_id: int
+    # The source classifier whose terminal classifications are re-scored.
+    algorithm_id: int
+    # When True (default), renormalise the kept classes' scores to sum to 1 after
+    # masking. When False, the kept classes retain their original absolute scores and
+    # the excluded classes are zeroed; the chosen species is identical either way.
+    reweight: bool = True
+
+    @pydantic.root_validator(skip_on_failure=True)
+    def _exactly_one_scope(cls, values: dict) -> dict:
+        scopes = [values.get("source_image_collection_id"), values.get("occurrence_id")]
+        if sum(s is not None for s in scopes) != 1:
+            raise ValueError("Provide exactly one of source_image_collection_id or occurrence_id")
+        return values
+
+    class Config:
+        extra = "forbid"
+
+
+def make_classifications_filtered_by_taxa_list(
+    classifications: QuerySet[Classification],
+    taxa_list: TaxaList,
+    algorithm: Algorithm,
+    new_algorithm: Algorithm,
+    *,
+    batch_size: int = 200,
+    reweight: bool = True,
+    task_logger: logging.Logger = logger,
+    on_batch: Callable[[dict], None] | None = None,
+) -> dict[str, int]:
+    """Re-score ``classifications`` by masking out classes absent from ``taxa_list``.
+
+    For each terminal classification produced by ``algorithm``, the logits of
+    classes whose taxon is not in ``taxa_list`` are masked, the softmax is
+    renormalised over the remaining classes (or, when ``reweight=False``, the kept
+    classes retain their original absolute scores), and a new terminal classification
+    (attributed to ``new_algorithm``, linked back via ``applied_to``) records the
+    masked prediction. The original classification is demoted to non-terminal.
+
+    Commits in batches of ``batch_size`` so memory stays bounded and the job
+    health-check reaper sees regular heartbeats. ``on_batch`` is called after every
+    flush with running counters:
+    ``{"classifications_checked": i, "classifications_total": total,
+       "classifications_masked": masked_count, "occurrences_updated": n_changed}``.
+
+    ``occurrences_updated`` counts only occurrences whose determination actually
+    changed (not just any occurrence touched), matching the size-filter convention.
+
+    Returns final counters (checked / masked / occurrences updated) for stage metrics.
+    """
+    taxa_in_list = set(taxa_list.taxa.all())
+
+    total = classifications.count()
+    task_logger.info(f"Found {total} terminal classifications with scores to re-score.")
+
+    if not algorithm.category_map:
+        raise ValueError(f"Algorithm {algorithm} does not have a category map.")
+    category_map = algorithm.category_map
+
+    # Resolve each category's taxon once. Indices absent from this map, or whose
+    # taxon is not in the taxa list, are masked. Building included from the taxa
+    # list (rather than excluded from the map) means a class with no resolvable
+    # taxon is masked too, never silently kept.
+    task_logger.info(f"Retrieving category map with Taxa instances for algorithm {algorithm}")
+    category_map_with_taxa = category_map.with_taxa()
+    index_to_taxon = {int(category["index"]): category["taxon"] for category in category_map_with_taxa}
+    num_categories = len(category_map.labels)
+    included_indices = [i for i in range(num_categories) if index_to_taxon.get(i) in taxa_in_list]
+    excluded_indices = [i for i in range(num_categories) if i not in set(included_indices)]
+
+    if not included_indices:
+        raise ValueError(
+            f"Taxa list '{taxa_list.name}' excludes every class in algorithm '{algorithm.name}'s "
+            "category map; there is nothing to keep."
+        )
+
+    task_logger.info(
+        f"Category map has {num_categories} classes, "
+        f"{len(excluded_indices)} masked, {len(included_indices)} kept, "
+        f"{total} classifications to check"
+    )
+
+    classifications_to_demote: list[Classification] = []
+    classifications_to_add: list[Classification] = []
+    occurrences_to_update: set[Occurrence] = set()
+    # Tracks occurrences whose determination actually changed across all batches.
+    changed_occurrence_ids: set[int] = set()
+
+    timestamp = timezone.now()
+    masked_count = 0
+
+    for i, classification in enumerate(classifications.iterator(chunk_size=batch_size), start=1):
+        logits = classification.logits
+        if not isinstance(logits, list) or not all(isinstance(x, (int, float)) for x in logits):
+            raise ValueError(f"Logits for classification {classification.pk} are not a list of numbers: {logits}")
+        elif len(logits) != num_categories:
+            task_logger.warning(
+                f"Classification {classification.pk}: {len(logits)} logits != {num_categories} categories; skipping"
+            )
+        else:
+            # Everything is derived from ``logits`` alone — the stored ``scores`` field
+            # is being retired, so masking must not depend on it. Recompute the model's
+            # full softmax (over every class) from the logits, then drop the excluded
+            # classes to exactly zero. An excluded class can never win argmax or carry
+            # probability.
+            shifted = np.asarray(logits, dtype=float)
+            shifted -= shifted.max()  # stabilises exp without changing the softmax
+            full_softmax = np.exp(shifted)
+            full_softmax /= full_softmax.sum()  # p over all classes; matches the retired ``scores``
+
+            kept = full_softmax.copy()
+            kept[excluded_indices] = 0.0
+            kept_sum = kept.sum()  # > 0: at least one class is kept (guaranteed upstream)
+            top_index = int(np.argmax(kept))  # over kept classes only (excluded are 0)
+
+            if reweight:
+                # Renormalise: excluded classes stay 0; kept classes sum to 1.
+                new_scores_np = kept / kept_sum
+            else:
+                # No renormalisation: kept classes keep their original absolute
+                # probability; excluded classes are zeroed. Winner is unchanged.
+                new_scores_np = kept
+            new_scores = new_scores_np.tolist()
+            score = float(new_scores_np[top_index])
+
+            # No-change short-circuit: if masking shifted no probability (the classes
+            # this taxa list drops carried ~zero probability here), leave the row
+            # untouched. Compared against the unmasked softmax, not the stored scores.
+            if np.allclose(full_softmax, new_scores_np, atol=1e-9):
+                task_logger.debug(f"Classification {classification.pk} unchanged by masking; skipping")
+            else:
+                top_taxon = index_to_taxon.get(top_index)  # guaranteed in taxa_in_list (top_index is kept)
+
+                classification.terminal = False
+                classification.updated_at = timestamp
+
+                new_classification = Classification(
+                    detection=classification.detection,
+                    taxon=top_taxon,
+                    algorithm=new_algorithm,
+                    category_map=new_algorithm.category_map,
+                    score=score,
+                    scores=new_scores,
+                    # Store the raw logits unchanged (JSON-safe): the mask is fully captured
+                    # by ``scores`` (dropped classes -> 0) and the ``applied_to`` lineage.
+                    logits=logits,
+                    terminal=True,
+                    timestamp=classification.timestamp,
+                    applied_to=classification,
+                    created_at=timestamp,
+                    updated_at=timestamp,
+                )
+                classifications_to_demote.append(classification)
+                classifications_to_add.append(new_classification)
+                masked_count += 1
+
+                detection = classification.detection
+                if detection is not None and detection.occurrence is not None:
+                    occurrences_to_update.add(detection.occurrence)
+
+        # Flush every batch_size items and at the final item. The flush fires even
+        # when nothing was accumulated so the job health-check sees a heartbeat during
+        # stretches where masking is a no-op (all short-circuited).
+        if i % batch_size == 0 or i == total:
+            with transaction.atomic():
+                if classifications_to_demote:
+                    Classification.objects.bulk_update(classifications_to_demote, ["terminal", "updated_at"])
+                if classifications_to_add:
+                    Classification.objects.bulk_create(classifications_to_add)
+                # Count an occurrence only when saving its new terminal classification
+                # actually changes the determination. Re-saving recomputes it in place,
+                # so an occurrence pinned to a human identification keeps its taxon
+                # and must not inflate the metric.
+                for occurrence in occurrences_to_update:
+                    prev = occurrence.determination_id
+                    occurrence.save(update_determination=True)
+                    if occurrence.pk is not None and occurrence.determination_id != prev:
+                        changed_occurrence_ids.add(occurrence.pk)
+
+            classifications_to_demote.clear()
+            classifications_to_add.clear()
+            occurrences_to_update.clear()
+
+            if on_batch is not None:
+                on_batch(
+                    {
+                        "classifications_checked": i,
+                        "classifications_total": total,
+                        "classifications_masked": masked_count,
+                        "occurrences_updated": len(changed_occurrence_ids),
+                    }
+                )
+
+    task_logger.info(
+        f"Re-scored {masked_count} of {total} classifications; updated {len(changed_occurrence_ids)} occurrences."
+    )
+    return {
+        "classifications_checked": total,
+        "classifications_masked": masked_count,
+        "occurrences_updated": len(changed_occurrence_ids),
+    }
+
+
+class ClassMaskingTask(BasePostProcessingTask):
+    key = "class_masking"
+    name = "Class masking"
+    config_schema = ClassMaskingConfig
+
+    def _get_or_create_masking_algorithm(
+        self, source_algorithm: Algorithm, taxa_list: TaxaList, *, reweight: bool
+    ) -> Algorithm:
+        """Get or create the output algorithm for this (source algorithm, taxa list, reweight mode).
+
+        One masking algorithm per (source algorithm, taxa list, reweight mode)
+        keeps provenance reproducible: re-running the same mask reuses the same
+        Algorithm row. The reweight mode is part of the identity because the two
+        modes persist different score semantics (renormalised vs original
+        absolute), so a masked classification's ``applied_to.algorithm`` can tell
+        them apart. Its category map is the source map (indices still align with
+        the masked score vector) and is persisted — earlier code set it in memory
+        only, so masked classifications referenced a null map.
+        """
+        mode = "reweighted" if reweight else "absolute"
+        algorithm, created = Algorithm.objects.get_or_create(
+            key=f"{source_algorithm.key}_filtered_by_taxa_list_{taxa_list.pk}_{mode}",
+            defaults={
+                "name": f"{source_algorithm.name} (filtered by taxa list {taxa_list.name}, {mode} scores)",
+                "description": (
+                    f"Classifications from {source_algorithm.name} re-scored against taxa list "
+                    f"{taxa_list.name} ({mode} scores)"
+                ),
+                "task_type": AlgorithmTaskType.CLASSIFICATION.value,
+                "category_map": source_algorithm.category_map,
+            },
+        )
+        if not created and algorithm.category_map_id != source_algorithm.category_map_id:
+            algorithm.category_map = source_algorithm.category_map
+            algorithm.save(update_fields=["category_map"])
+        return algorithm
+
+    def _scoped_classifications(
+        self, config: ClassMaskingConfig, source_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.
+        """
+        base = Classification.objects.filter(
+            terminal=True,
+            algorithm=source_algorithm,
+            scores__isnull=False,
+            logits__isnull=False,
+        ).select_related("detection", "detection__occurrence")
+
+        if config.occurrence_id is not None:
+            if not Occurrence.objects.filter(pk=config.occurrence_id).exists():
+                raise ValueError(f"Occurrence {config.occurrence_id} not found")
+            return (
+                base.filter(detection__occurrence_id=config.occurrence_id).distinct(),
+                f"occurrence {config.occurrence_id}",
+            )
+
+        try:
+            collection = SourceImageCollection.objects.get(pk=config.source_image_collection_id)
+        except SourceImageCollection.DoesNotExist:
+            raise ValueError(f"SourceImageCollection {config.source_image_collection_id} not found")
+        return (
+            base.filter(detection__source_image__collections=collection).distinct(),
+            f"collection {collection.pk}",
+        )
+
+    def run(self) -> None:
+        config: ClassMaskingConfig = self.config  # type: ignore[assignment]
+        self.logger.info(f"=== Starting {self.name} ===")
+
+        try:
+            source_algorithm = Algorithm.objects.get(pk=config.algorithm_id)
+        except Algorithm.DoesNotExist:
+            raise ValueError(f"Algorithm {config.algorithm_id} not found")
+        try:
+            taxa_list = TaxaList.objects.get(pk=config.taxa_list_id)
+        except TaxaList.DoesNotExist:
+            raise ValueError(f"TaxaList {config.taxa_list_id} not found")
+        if not source_algorithm.category_map:
+            raise ValueError(f"Algorithm '{source_algorithm.name}' has no category map; cannot mask classes.")
+
+        masking_algorithm = self._get_or_create_masking_algorithm(
+            source_algorithm, taxa_list, reweight=config.reweight
+        )
+        classifications, scope_desc = self._scoped_classifications(config, source_algorithm)
+        self.logger.info(f"Applying class masking on {scope_desc} using taxa list {taxa_list.pk}")
+
+        def _on_batch(m: dict) -> None:
+            total = m["classifications_total"]
+            self.update_progress(m["classifications_checked"] / total if total else 1.0)
+            self.report_stage_metrics(
+                {
+                    "classifications_checked": m["classifications_checked"],
+                    "classifications_masked": m["classifications_masked"],
+                    "occurrences_updated": m["occurrences_updated"],
+                }
+            )
+
+        metrics = make_classifications_filtered_by_taxa_list(
+            classifications=classifications,
+            taxa_list=taxa_list,
+            algorithm=source_algorithm,
+            new_algorithm=masking_algorithm,
+            reweight=config.reweight,
+            task_logger=self.logger,
+            on_batch=_on_batch,
+        )
+        self.report_stage_metrics(metrics)
+        self.logger.info(f"=== Completed {self.name} ===")
diff --git a/ami/ml/post_processing/registry.py b/ami/ml/post_processing/registry.py
index c85f607f9..308be18ae 100644
--- a/ami/ml/post_processing/registry.py
+++ b/ami/ml/post_processing/registry.py
@@ -1,8 +1,10 @@
 # Registry of available post-processing tasks
+from ami.ml.post_processing.class_masking import ClassMaskingTask
 from ami.ml.post_processing.small_size_filter import SmallSizeFilterTask
 
 POSTPROCESSING_TASKS = {
     SmallSizeFilterTask.key: SmallSizeFilterTask,
+    ClassMaskingTask.key: ClassMaskingTask,
 }
 
 
diff --git a/ami/ml/post_processing/tests/test_class_masking.py b/ami/ml/post_processing/tests/test_class_masking.py
new file mode 100644
index 000000000..b0578272c
--- /dev/null
+++ b/ami/ml/post_processing/tests/test_class_masking.py
@@ -0,0 +1,488 @@
+"""Domain tests for the class masking post-processing task.
+
+Class masking re-scores a classifier's terminal predictions against a taxa list:
+classes whose taxon is not in the list are masked, the softmax is renormalised
+over the rest, and a new terminal classification (linked back via ``applied_to``)
+records the masked result. These tests cover the masking maths (including that an
+excluded class can never win even when it had the highest logit), the provenance
+link, the persisted output algorithm, and both admin scopes (collection / single
+occurrence) end to end through ``ClassMaskingTask.run()``.
+"""
+import datetime
+import math
+import pathlib
+import uuid
+
+from django.test import TestCase
+
+from ami.main.models import (
+    Classification,
+    Detection,
+    Occurrence,
+    SourceImage,
+    SourceImageCollection,
+    TaxaList,
+    Taxon,
+    TaxonRank,
+    group_images_into_events,
+)
+from ami.ml.models import Algorithm, AlgorithmCategoryMap
+from ami.ml.models.algorithm import AlgorithmTaskType
+from ami.ml.post_processing.class_masking import ClassMaskingTask, make_classifications_filtered_by_taxa_list
+from ami.tests.fixtures.main import create_taxa, setup_test_project
+
+
+def _softmax(logits: list[float]) -> list[float]:
+    shifted = [x - max(logits) for x in logits]
+    exp = [math.exp(x) for x in shifted]
+    total = sum(exp)
+    return [e / total for e in exp]
+
+
+class TestPostProcessingClassMasking(TestCase):
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(project=self.project)
+        self._create_images_with_dimensions(deployment=self.deployment)
+        group_images_into_events(deployment=self.deployment)
+
+        self.collection = SourceImageCollection.objects.create(
+            name="Test PostProcessing Collection",
+            project=self.project,
+            method="manual",
+            kwargs={"image_ids": list(self.deployment.captures.values_list("pk", flat=True))},
+        )
+        self.collection.populate_sample()
+
+        self.species_taxon = Taxon.objects.filter(rank=TaxonRank.SPECIES.name).first()
+        self.genus_taxon = self.species_taxon.parent if self.species_taxon else None
+        self.assertIsNotNone(self.species_taxon)
+        self.assertIsNotNone(self.genus_taxon)
+        self.algorithm = self._create_category_map_with_algorithm()
+        self.species_taxa = list(self.project.taxa.filter(rank=TaxonRank.SPECIES.name).order_by("name")[:3])
+
+    # ----- fixtures -------------------------------------------------------
+
+    def _create_images_with_dimensions(self, deployment, num_images=5, width=640, height=480):
+        base_time = datetime.datetime.now(datetime.timezone.utc)
+        for i in range(num_images):
+            path = pathlib.Path("test") / f"{uuid.uuid4().hex[:8]}_{i}.jpg"
+            SourceImage.objects.create(
+                deployment=deployment,
+                project=deployment.project,
+                timestamp=base_time + datetime.timedelta(minutes=i * 5),
+                path=path,
+                width=width,
+                height=height,
+            )
+        deployment.save(update_calculated_fields=True, regroup_async=False)
+
+    def _create_category_map_with_algorithm(self) -> Algorithm:
+        species_taxa = list(self.project.taxa.filter(rank=TaxonRank.SPECIES.name).order_by("name")[:3])
+        assert species_taxa, "No species taxa found in project; run create_taxa() first."
+        data = [{"index": i, "label": taxon.name} for i, taxon in enumerate(species_taxa)]
+        category_map = AlgorithmCategoryMap.objects.create(
+            data=data,
+            labels=[item["label"] for item in data],
+            version="v1.0",
+            description="Species-level category map for testing",
+        )
+        return Algorithm.objects.create(
+            name="Test Species Classifier",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=category_map,
+        )
+
+    def _create_classification_with_logits(self, detection, taxon, scores, logits) -> Classification:
+        return Classification.objects.create(
+            detection=detection,
+            taxon=taxon,
+            score=max(scores),
+            scores=scores,
+            logits=logits,
+            terminal=True,
+            timestamp=datetime.datetime.now(datetime.timezone.utc),
+            algorithm=self.algorithm,
+        )
+
+    def _detection_with_occurrence(self) -> tuple[Detection, Occurrence]:
+        det = Detection.objects.create(source_image=self.collection.images.first(), bbox=[0, 0, 200, 200])
+        occ = Occurrence.objects.create(project=self.project, event=self.deployment.events.first())
+        occ.detections.add(det)
+        return det, occ
+
+    # ----- make_classifications_filtered_by_taxa_list ---------------------
+
+    def test_excluded_class_never_wins_even_with_highest_logit(self):
+        """The core guarantee: a masked class cannot be selected, even if it had
+        the single highest logit before masking."""
+        # index 2 (excluded) has the highest logit, so it is the original top.
+        logits = [2.0, 1.0, 5.0]
+        scores = _softmax(logits)
+        self.assertEqual(scores.index(max(scores)), 2)
+
+        taxa_list = TaxaList.objects.create(name="Keep first two")
+        taxa_list.taxa.set(self.species_taxa[:2])  # excludes species_taxa[2]
+
+        det, _ = self._detection_with_occurrence()
+        original = self._create_classification_with_logits(det, self.species_taxa[2], scores, logits)
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked",
+            key="masked_test",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        metrics = make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk=original.pk),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_algorithm,
+        )
+
+        original.refresh_from_db()
+        self.assertFalse(original.terminal, "Source classification is demoted to non-terminal")
+        new_clf = Classification.objects.get(detection=det, terminal=True)
+        # Highest logit among the kept classes is index 0.
+        self.assertEqual(new_clf.taxon, self.species_taxa[0])
+        self.assertAlmostEqual(new_clf.scores[2], 0.0, places=10, msg="Masked class score is exactly zero")
+        self.assertAlmostEqual(sum(new_clf.scores), 1.0, places=5)
+        self.assertEqual(new_clf.applied_to, original, "Provenance links back to the source classification")
+        self.assertEqual(metrics["classifications_masked"], 1)
+
+    def test_single_allowed_class_gets_all_probability(self):
+        logits = [2.0, 3.0, 4.0]
+        taxa_list = TaxaList.objects.create(name="Keep one")
+        taxa_list.taxa.set([self.species_taxa[0]])
+
+        det, _ = self._detection_with_occurrence()
+        self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits)
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked2",
+            key="masked_test2",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(detection=det, terminal=True),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_algorithm,
+        )
+        new_clf = Classification.objects.get(detection=det, terminal=True)
+        self.assertAlmostEqual(new_clf.scores[0], 1.0, places=5)
+        self.assertAlmostEqual(new_clf.scores[1], 0.0, places=10)
+        self.assertAlmostEqual(new_clf.scores[2], 0.0, places=10)
+
+    def test_no_change_when_all_classes_in_list(self):
+        logits = [3.0, 1.0, 0.5]
+        taxa_list = TaxaList.objects.create(name="Keep all")
+        taxa_list.taxa.set(self.species_taxa)
+
+        det, _ = self._detection_with_occurrence()
+        original = self._create_classification_with_logits(det, self.species_taxa[0], _softmax(logits), logits)
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked3",
+            key="masked_test3",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk=original.pk),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_algorithm,
+        )
+        original.refresh_from_db()
+        self.assertTrue(original.terminal, "Nothing masked, so the source stays terminal")
+        self.assertEqual(Classification.objects.filter(detection=det).count(), 1, "No new classification created")
+
+    def test_all_classes_excluded_raises(self):
+        # A taxa list sharing nothing with the category map leaves no class to keep.
+        taxa_list = TaxaList.objects.create(name="Unrelated")
+        taxa_list.taxa.set([self.genus_taxon])  # genus name is not a category-map label
+
+        det, _ = self._detection_with_occurrence()
+        logits = [2.0, 1.0, 5.0]
+        self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits)
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked4",
+            key="masked_test4",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        with self.assertRaises(ValueError):
+            make_classifications_filtered_by_taxa_list(
+                classifications=Classification.objects.filter(detection=det, terminal=True),
+                taxa_list=taxa_list,
+                algorithm=self.algorithm,
+                new_algorithm=new_algorithm,
+            )
+
+    # ----- ClassMaskingTask.run() end to end ------------------------------
+
+    def test_task_run_collection_scope_persists_masking_algorithm(self):
+        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="Regional list")
+        taxa_list.taxa.set(self.species_taxa[:2])
+
+        det, occ = self._detection_with_occurrence()
+        original = self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits)
+
+        ClassMaskingTask(
+            source_image_collection_id=self.collection.pk,
+            taxa_list_id=taxa_list.pk,
+            algorithm_id=self.algorithm.pk,
+        ).run()
+
+        # The per-(source algorithm, taxa list, reweight mode) masking algorithm
+        # exists and kept its category map (the bug being guarded: it used to be
+        # set in memory only). Default reweight=True → the "reweighted" mode.
+        masking_algo = Algorithm.objects.get(
+            key=f"{self.algorithm.key}_filtered_by_taxa_list_{taxa_list.pk}_reweighted"
+        )
+        self.assertIsNotNone(masking_algo.category_map_id)
+        self.assertEqual(masking_algo.category_map_id, self.algorithm.category_map_id)
+
+        new_clf = Classification.objects.get(detection=det, terminal=True, algorithm=masking_algo)
+        self.assertEqual(new_clf.taxon, self.species_taxa[1])
+        self.assertEqual(new_clf.applied_to, original)
+        occ.refresh_from_db()
+        self.assertEqual(occ.determination, self.species_taxa[1], "Occurrence determination follows the masked result")
+
+    def test_reweight_modes_get_distinct_masking_algorithms(self):
+        """The reweight mode is part of the masking algorithm's identity.
+
+        reweight=True and reweight=False persist different score semantics
+        (renormalised vs original absolute), so they must resolve to different
+        Algorithm rows — otherwise a masked classification's
+        ``applied_to.algorithm`` could not tell which mode produced it. Both keys
+        derive from the same (source algorithm, taxa list); only the mode suffix
+        differs.
+        """
+        logits = [0.5, 3.0, 3.5]
+        taxa_list = TaxaList.objects.create(name="Reweight identity list")
+        taxa_list.taxa.set(self.species_taxa[:2])
+
+        det_t, _ = self._detection_with_occurrence()
+        self._create_classification_with_logits(det_t, self.species_taxa[2], _softmax(logits), logits)
+        ClassMaskingTask(
+            source_image_collection_id=self.collection.pk,
+            taxa_list_id=taxa_list.pk,
+            algorithm_id=self.algorithm.pk,
+            reweight=True,
+        ).run()
+
+        det_f, _ = self._detection_with_occurrence()
+        self._create_classification_with_logits(det_f, self.species_taxa[2], _softmax(logits), logits)
+        ClassMaskingTask(
+            source_image_collection_id=self.collection.pk,
+            taxa_list_id=taxa_list.pk,
+            algorithm_id=self.algorithm.pk,
+            reweight=False,
+        ).run()
+
+        base = f"{self.algorithm.key}_filtered_by_taxa_list_{taxa_list.pk}"
+        reweighted = Algorithm.objects.get(key=f"{base}_reweighted")
+        absolute = Algorithm.objects.get(key=f"{base}_absolute")
+        self.assertNotEqual(reweighted.pk, absolute.pk, "Each reweight mode gets its own masking algorithm")
+
+        # Each detection's masked classification points at the algorithm for its mode.
+        self.assertTrue(Classification.objects.filter(detection=det_t, terminal=True, algorithm=reweighted).exists())
+        self.assertTrue(Classification.objects.filter(detection=det_f, terminal=True, algorithm=absolute).exists())
+
+    def test_task_run_occurrence_scope(self):
+        logits = [2.0, 1.0, 5.0]
+        taxa_list = TaxaList.objects.create(name="Occ scope list")
+        taxa_list.taxa.set(self.species_taxa[:2])
+
+        det, occ = self._detection_with_occurrence()
+        self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits)
+
+        ClassMaskingTask(
+            occurrence_id=occ.pk,
+            taxa_list_id=taxa_list.pk,
+            algorithm_id=self.algorithm.pk,
+        ).run()
+
+        new_clf = Classification.objects.filter(detection=det, terminal=True).exclude(algorithm=self.algorithm).first()
+        self.assertIsNotNone(new_clf)
+        self.assertEqual(new_clf.taxon, self.species_taxa[0])
+
+    # ----- batched commit + heartbeat -------------------------------------
+
+    def test_batched_commit_flushes_multiple_times(self):
+        """With batch_size=2 and 5 classifications, on_batch fires at i=2, 4, 5 — more than once.
+
+        Verifies that all classifications are correctly masked across flush boundaries
+        (correctness preserved) and that the on_batch callback receives a call per batch
+        (heartbeat wiring works)."""
+        taxa_list = TaxaList.objects.create(name="Batch flush test")
+        taxa_list.taxa.set(self.species_taxa[:1])  # keep only index 0; indices 1 and 2 excluded
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked_batch",
+            key="masked_batch_test",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+
+        # Create 5 detections, each with a classification whose top logit is index 2
+        # (excluded), so every one should be masked and re-assigned to index 0.
+        logits = [2.0, 1.0, 5.0]
+        pks = []
+        for _ in range(5):
+            det, _ = self._detection_with_occurrence()
+            clf = self._create_classification_with_logits(det, self.species_taxa[2], _softmax(logits), logits)
+            pks.append(clf.pk)
+
+        batch_calls: list[dict] = []
+        metrics = make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk__in=pks),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_algorithm,
+            batch_size=2,
+            on_batch=batch_calls.append,
+        )
+
+        self.assertGreater(len(batch_calls), 1, "on_batch must fire more than once with batch_size=2 over 5 items")
+        # i=2 and i=4 fire on the batch boundary; i=5 fires on the total boundary.
+        self.assertEqual(len(batch_calls), 3)
+        self.assertEqual(metrics["classifications_masked"], 5, "All 5 classifications must be masked across batches")
+        self.assertEqual(
+            Classification.objects.filter(algorithm=new_algorithm, terminal=True).count(),
+            5,
+            "A new terminal classification must exist for every masked row",
+        )
+
+    # ----- changed-only occurrence count ----------------------------------
+
+    def test_occurrences_updated_counts_only_changed_determinations(self):
+        """``occurrences_updated`` counts only occurrences whose determination changed,
+        not every occurrence whose detection was touched.
+
+        occ1: original winner is index 2 (excluded) — masking flips determination to index 0.
+        occ2: original winner is index 0 (kept) — masking reassigns scores but determination stays index 0.
+        Only occ1 should count."""
+        taxa_list = TaxaList.objects.create(name="Changed-only count test")
+        taxa_list.taxa.set(self.species_taxa[:2])  # excludes species_taxa[2] (index 2)
+
+        new_algorithm = Algorithm.objects.create(
+            name="masked_changed",
+            key="masked_changed_test",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+
+        # occ1: index 2 has the highest logit, so masking forces the winner to index 0.
+        logits_changes = [2.0, 1.0, 5.0]
+        det1, occ1 = self._detection_with_occurrence()
+        clf1 = self._create_classification_with_logits(
+            det1, self.species_taxa[2], _softmax(logits_changes), logits_changes
+        )
+        # Persist the pre-masking determination so the loaded instance has it set.
+        occ1.save(update_determination=True)  # determination = species_taxa[2]
+
+        # occ2: index 0 has the highest logit, so after masking it remains the winner.
+        # Masking still changes the scores (index 2 drops to 0, softmax shifts), so the
+        # classification IS demoted — but occ2's determination stays species_taxa[0].
+        logits_stays = [5.0, 1.0, 3.0]
+        det2, occ2 = self._detection_with_occurrence()
+        clf2 = self._create_classification_with_logits(
+            det2, self.species_taxa[0], _softmax(logits_stays), logits_stays
+        )
+        occ2.save(update_determination=True)  # determination = species_taxa[0]
+
+        metrics = make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk__in=[clf1.pk, clf2.pk]),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_algorithm,
+        )
+
+        self.assertEqual(metrics["classifications_masked"], 2, "Both classifications are modified by masking")
+        self.assertEqual(
+            metrics["occurrences_updated"],
+            1,
+            "Only the occurrence whose determination changed (occ1) counts",
+        )
+
+    # ----- reweight toggle ------------------------------------------------
+
+    def test_reweight_false_winner_identical_scores_differ(self):
+        """With reweight=False the winning taxon is identical to reweight=True, but the
+        stored score equals the winner's original absolute probability (not renormalised)
+        and the kept-class scores do not sum to 1.
+
+        logits = [2, 1, 5]: index 2 is excluded, so index 0 wins in both modes.
+        reweight=True:  new_scores renormalised — sums to 1, score = renormalised p(index 0).
+        reweight=False: new_scores = original with index 2 zeroed — sums < 1, score = original p(index 0)."""
+        logits = [2.0, 1.0, 5.0]
+        scores = _softmax(logits)
+        taxa_list = TaxaList.objects.create(name="Reweight compare")
+        taxa_list.taxa.set(self.species_taxa[:2])  # excludes index 2
+
+        # --- reweight=True (default) ---
+        det_t, _ = self._detection_with_occurrence()
+        clf_t = self._create_classification_with_logits(det_t, self.species_taxa[2], scores, logits)
+        new_alg_true = Algorithm.objects.create(
+            name="masked_rw_true",
+            key="masked_rw_true_test",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk=clf_t.pk),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_alg_true,
+            reweight=True,
+        )
+
+        # --- reweight=False ---
+        det_f, _ = self._detection_with_occurrence()
+        clf_f = self._create_classification_with_logits(det_f, self.species_taxa[2], scores, logits)
+        new_alg_false = Algorithm.objects.create(
+            name="masked_rw_false",
+            key="masked_rw_false_test",
+            task_type=AlgorithmTaskType.CLASSIFICATION.value,
+            category_map=self.algorithm.category_map,
+        )
+        make_classifications_filtered_by_taxa_list(
+            classifications=Classification.objects.filter(pk=clf_f.pk),
+            taxa_list=taxa_list,
+            algorithm=self.algorithm,
+            new_algorithm=new_alg_false,
+            reweight=False,
+        )
+
+        new_clf_t = Classification.objects.get(detection=det_t, terminal=True)
+        new_clf_f = Classification.objects.get(detection=det_f, terminal=True)
+
+        # Winner is the same in both modes.
+        self.assertEqual(new_clf_t.taxon, self.species_taxa[0])
+        self.assertEqual(new_clf_f.taxon, self.species_taxa[0], "Winner is identical with reweight=False")
+
+        # Excluded class is zeroed in both modes.
+        self.assertAlmostEqual(new_clf_t.scores[2], 0.0, places=10)
+        self.assertAlmostEqual(new_clf_f.scores[2], 0.0, places=10)
+
+        # reweight=True: kept-class scores are renormalised and sum to 1.
+        self.assertAlmostEqual(sum(new_clf_t.scores), 1.0, places=5)
+
+        # reweight=False: kept classes retain original absolute values; sum is < 1.
+        self.assertAlmostEqual(
+            new_clf_f.scores[0], scores[0], places=6, msg="Kept class retains original score with reweight=False"
+        )
+        self.assertAlmostEqual(new_clf_f.scores[1], scores[1], places=6)
+        self.assertNotAlmostEqual(
+            sum(new_clf_f.scores), 1.0, places=2, msg="Scores must not sum to 1 with reweight=False"
+        )
+
+        # Stored confidence: reweight=False uses the original pre-mask probability.
+        self.assertAlmostEqual(new_clf_f.score, scores[0], places=6)
+        self.assertNotAlmostEqual(new_clf_t.score, scores[0], places=2)
diff --git a/ami/ml/post_processing/tests/test_class_masking_admin.py b/ami/ml/post_processing/tests/test_class_masking_admin.py
new file mode 100644
index 000000000..4664f52d3
--- /dev/null
+++ b/ami/ml/post_processing/tests/test_class_masking_admin.py
@@ -0,0 +1,224 @@
+"""Schema validation + admin-action wiring tests for class masking.
+
+These are deliberately lightweight: they exercise the pydantic config contracts
+and the admin trigger flow (intermediate page -> Job creation with the right
+config payload) without the full project fixture. The masking maths is covered
+in ``test_class_masking``.
+"""
+import pydantic
+from django.contrib import admin as django_admin
+from django.test import Client, TestCase
+from django.urls import reverse
+from django.utils import timezone
+
+from ami.jobs.models import Job
+from ami.main.models import (
+    Classification,
+    Deployment,
+    Detection,
+    Occurrence,
+    Project,
+    SourceImage,
+    SourceImageCollection,
+    TaxaList,
+)
+from ami.ml.models import Algorithm
+from ami.ml.models.algorithm import AlgorithmTaskType
+from ami.ml.post_processing.admin.class_masking_form import ClassMaskingActionForm
+from ami.ml.post_processing.class_masking import ClassMaskingConfig
+from ami.users.models import User
+
+
+class TestClassMaskingConfig(TestCase):
+    def test_collection_scope_is_valid(self):
+        config = ClassMaskingConfig(source_image_collection_id=1, taxa_list_id=2, algorithm_id=3)
+        self.assertEqual(config.source_image_collection_id, 1)
+        self.assertIsNone(config.occurrence_id)
+
+    def test_occurrence_scope_is_valid(self):
+        config = ClassMaskingConfig(occurrence_id=5, taxa_list_id=2, algorithm_id=3)
+        self.assertEqual(config.occurrence_id, 5)
+
+    def test_both_scopes_is_invalid(self):
+        with self.assertRaises(pydantic.ValidationError):
+            ClassMaskingConfig(source_image_collection_id=1, occurrence_id=5, taxa_list_id=2, algorithm_id=3)
+
+    def test_no_scope_is_invalid(self):
+        with self.assertRaises(pydantic.ValidationError):
+            ClassMaskingConfig(taxa_list_id=2, algorithm_id=3)
+
+    def test_missing_required_fields_is_invalid(self):
+        with self.assertRaises(pydantic.ValidationError):
+            ClassMaskingConfig(source_image_collection_id=1)  # no taxa_list_id / algorithm_id
+
+    def test_extra_field_is_forbidden(self):
+        with self.assertRaises(pydantic.ValidationError):
+            ClassMaskingConfig(source_image_collection_id=1, taxa_list_id=2, algorithm_id=3, bogus=1)
+
+    def test_reweight_defaults_to_true(self):
+        config = ClassMaskingConfig(source_image_collection_id=1, taxa_list_id=2, algorithm_id=3)
+        self.assertTrue(config.reweight)
+
+    def test_reweight_can_be_set_false(self):
+        config = ClassMaskingConfig(source_image_collection_id=1, taxa_list_id=2, algorithm_id=3, reweight=False)
+        self.assertFalse(config.reweight)
+
+
+class _PostProcessingAdminCase(TestCase):
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.superuser = User.objects.create_superuser(email=f"ppadmin+{cls.__name__}@example.com", password="x")
+        cls.project = Project.objects.create(name=f"PP admin test ({cls.__name__})")
+        cls.collection = SourceImageCollection.objects.create(project=cls.project, name="PP admin collection")
+        cls.occurrence = Occurrence.objects.create(project=cls.project)
+        cls.taxa_list = TaxaList.objects.create(name="PP admin taxa list")
+        cls.algorithm = Algorithm.objects.create(
+            name="PP admin classifier", task_type=AlgorithmTaskType.CLASSIFICATION.value
+        )
+        # Wire the classifier to both scopes (the collection's image and the
+        # occurrence) so the class-mask form offers it — it only lists algorithms
+        # that actually produced classifications within the selection.
+        cls.deployment = Deployment.objects.create(project=cls.project, name="PP admin dep")
+        source_image = SourceImage.objects.create(deployment=cls.deployment, project=cls.project, path="pp-admin.jpg")
+        cls.collection.images.add(source_image)
+        detection = Detection.objects.create(source_image=source_image, bbox=[0, 0, 1, 1], occurrence=cls.occurrence)
+        Classification.objects.create(detection=detection, algorithm=cls.algorithm, timestamp=timezone.now())
+
+    def setUp(self) -> None:
+        self.client = Client()
+        self.client.force_login(self.superuser)
+
+
+class TestClassMaskingAdmin(_PostProcessingAdminCase):
+    def _post_collection(self, data: dict):
+        url = reverse("admin:main_sourceimagecollection_changelist")
+        return self.client.post(
+            url,
+            data={
+                "action": "run_class_masking",
+                django_admin.helpers.ACTION_CHECKBOX_NAME: [str(self.collection.pk)],
+                **data,
+            },
+        )
+
+    def test_renders_intermediate_page_without_confirm(self):
+        response = self._post_collection({})
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"Run Class masking", response.content)
+        self.assertIn(b'name="taxa_list_id"', response.content)
+        self.assertIn(b'name="algorithm_id"', response.content)
+        self.assertEqual(Job.objects.filter(project=self.project).count(), 0)
+
+    def test_valid_post_creates_collection_scoped_job(self):
+        response = self._post_collection(
+            {"confirm": "yes", "taxa_list_id": str(self.taxa_list.pk), "algorithm_id": str(self.algorithm.pk)}
+        )
+        self.assertEqual(response.status_code, 302)
+        job = Job.objects.get(project=self.project, job_type_key="post_processing")
+        self.assertEqual(job.params["task"], "class_masking")
+        self.assertEqual(job.params["config"]["source_image_collection_id"], self.collection.pk)
+        self.assertEqual(job.params["config"]["taxa_list_id"], self.taxa_list.pk)
+        self.assertEqual(job.params["config"]["algorithm_id"], self.algorithm.pk)
+        self.assertIsNone(job.params["config"].get("occurrence_id"))
+
+    def test_valid_post_on_occurrence_creates_occurrence_scoped_job(self):
+        url = reverse("admin:main_occurrence_changelist")
+        response = self.client.post(
+            url,
+            data={
+                "action": "run_class_masking",
+                django_admin.helpers.ACTION_CHECKBOX_NAME: [str(self.occurrence.pk)],
+                "confirm": "yes",
+                "taxa_list_id": str(self.taxa_list.pk),
+                "algorithm_id": str(self.algorithm.pk),
+            },
+        )
+        self.assertEqual(response.status_code, 302)
+        job = Job.objects.get(project=self.project, job_type_key="post_processing")
+        self.assertEqual(job.params["task"], "class_masking")
+        self.assertEqual(job.params["config"]["occurrence_id"], self.occurrence.pk)
+        self.assertIsNone(job.params["config"].get("source_image_collection_id"))
+
+
+class TestClassMaskingFormReweight(_PostProcessingAdminCase):
+    """The admin form exposes a reweight toggle and passes it through to the job config."""
+
+    def _post_collection_reweight(self, include_reweight: bool):
+        data = {
+            "action": "run_class_masking",
+            django_admin.helpers.ACTION_CHECKBOX_NAME: [str(self.collection.pk)],
+            "confirm": "yes",
+            "taxa_list_id": str(self.taxa_list.pk),
+            "algorithm_id": str(self.algorithm.pk),
+        }
+        if include_reweight:
+            data["reweight"] = "on"
+        return self.client.post(reverse("admin:main_sourceimagecollection_changelist"), data=data)
+
+    def test_form_has_reweight_field(self):
+        form = ClassMaskingActionForm()
+        self.assertIn("reweight", form.fields)
+        self.assertTrue(form.fields["reweight"].initial, "reweight must default to True (checked)")
+
+    def test_to_config_includes_reweight_true_when_checked(self):
+        """When the operator checks the reweight box, the job config carries reweight=True."""
+        response = self._post_collection_reweight(include_reweight=True)
+        self.assertEqual(response.status_code, 302)
+        job = Job.objects.get(project=self.project, job_type_key="post_processing")
+        self.assertTrue(job.params["config"]["reweight"])
+
+    def test_to_config_includes_reweight_false_when_unchecked(self):
+        """An unchecked reweight box (no value in POST) yields reweight=False in the job config."""
+        response = self._post_collection_reweight(include_reweight=False)
+        self.assertEqual(response.status_code, 302)
+        job = Job.objects.get(project=self.project, job_type_key="post_processing")
+        self.assertFalse(job.params["config"]["reweight"])
+
+
+class TestClassMaskingFormScopeFiltering(TestCase):
+    """The class-mask form offers only classifiers that actually produced
+    classifications within the selected scope, so an operator cannot pick an
+    algorithm whose masking would be a no-op for the chosen occurrence."""
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.project = Project.objects.create(name="CM scope filter project")
+        cls.deployment = Deployment.objects.create(project=cls.project, name="dep")
+        cls.source_image = SourceImage.objects.create(
+            deployment=cls.deployment, project=cls.project, path="cm-scope.jpg"
+        )
+        cls.used = Algorithm.objects.create(name="used classifier", task_type=AlgorithmTaskType.CLASSIFICATION.value)
+        cls.unused = Algorithm.objects.create(
+            name="unused classifier", task_type=AlgorithmTaskType.CLASSIFICATION.value
+        )
+
+        cls.occurrence = Occurrence.objects.create(project=cls.project, deployment=cls.deployment)
+        detection = Detection.objects.create(
+            source_image=cls.source_image, bbox=[0, 0, 1, 1], occurrence=cls.occurrence
+        )
+        Classification.objects.create(detection=detection, algorithm=cls.used, timestamp=timezone.now())
+
+    def test_form_offers_only_algorithms_used_on_the_occurrence(self):
+        form = ClassMaskingActionForm(scope_queryset=Occurrence.objects.filter(pk=self.occurrence.pk))
+        offered = set(form.fields["algorithm_id"].queryset.values_list("pk", flat=True))
+        self.assertEqual(offered, {self.used.pk})
+
+    def test_form_without_scope_offers_all_classifiers(self):
+        form = ClassMaskingActionForm()
+        offered = set(form.fields["algorithm_id"].queryset.values_list("pk", flat=True))
+        self.assertIn(self.used.pk, offered)
+        self.assertIn(self.unused.pk, offered)
+
+    def test_collection_scope_offers_all_classifiers(self):
+        """A collection scope intentionally keeps the full classifier list rather
+        than narrowing to the classifiers used in the collection. The narrowing
+        lookup is an unbounded DISTINCT over every classification in the
+        collection, which can time out while rendering the form on a large
+        collection. This pins that the collection path stays unfiltered so the
+        expensive filter is not re-added there by mistake."""
+        collection = SourceImageCollection.objects.create(project=self.project, name="scope coll")
+        collection.images.add(self.source_image)
+        form = ClassMaskingActionForm(scope_queryset=SourceImageCollection.objects.filter(pk=collection.pk))
+        offered = set(form.fields["algorithm_id"].queryset.values_list("pk", flat=True))
+        self.assertIn(self.used.pk, offered)
+        self.assertIn(self.unused.pk, offered)
diff --git a/ami/ml/tests.py b/ami/ml/tests.py
index 6781f80ee..cb88ee6fd 100644
--- a/ami/ml/tests.py
+++ b/ami/ml/tests.py
@@ -1603,6 +1603,54 @@ def test_run_reports_stage_metrics_on_job(self):
         # count equals the detection count.
         self.assertEqual(params.get("occurrences_updated"), total)
 
+    def test_progress_save_bumps_updated_at_for_reaper(self):
+        """A progress heartbeat bumps ``Job.updated_at`` so the stale-job reaper
+        leaves an actively-running post-processing job alone.
+
+        ``check_stale_jobs`` revokes running jobs whose ``updated_at`` is older
+        than ``STALLED_JOBS_MAX_MINUTES``. The progress save narrows to
+        ``update_fields``, and Django does not auto-add ``auto_now`` fields to
+        that list, so ``update_progress`` / ``report_stage_metrics`` must include
+        ``updated_at`` explicitly. Without it a long run looks frozen and is
+        reaped mid-flight even while streaming progress. This pins that both save
+        paths move ``updated_at`` forward.
+        """
+        from ami.jobs.models import Job
+
+        job = Job.objects.create(
+            project=self.project,
+            name="reaper heartbeat test",
+            job_type_key="post_processing",
+            params={
+                "task": "small_size_filter",
+                "config": {"source_image_collection_id": self.collection.pk, "size_threshold": 0.01},
+            },
+        )
+        job.progress.add_stage("Post Processing", key="post_processing")
+        job.save()
+
+        task = SmallSizeFilterTask(
+            job=job,
+            source_image_collection_id=self.collection.pk,
+            size_threshold=0.01,
+        )
+
+        # Freeze a baseline older than the reaper cutoff, then confirm each
+        # heartbeat path drags updated_at back to "now". USE_TZ is False, so
+        # updated_at is naive local time — mirror check_stale_jobs' own
+        # naive datetime.now() comparison.
+        stale = datetime.datetime.now() - datetime.timedelta(minutes=Job.STALLED_JOBS_MAX_MINUTES + 5)
+
+        Job.objects.filter(pk=job.pk).update(updated_at=stale)
+        task.update_progress(0.5)
+        job.refresh_from_db()
+        self.assertGreater(job.updated_at, stale, "update_progress must bump updated_at")
+
+        Job.objects.filter(pk=job.pk).update(updated_at=stale)
+        task.report_stage_metrics({"classifications_checked": 1})
+        job.refresh_from_db()
+        self.assertGreater(job.updated_at, stale, "report_stage_metrics must bump updated_at")
+
     def test_occurrences_updated_counts_only_changed_determinations(self):
         """``occurrences_updated`` counts occurrences whose determination actually
         changed, not every occurrence the filter re-saved.