diff --git a/ami/jobs/serializers.py b/ami/jobs/serializers.py
index f53199e73..206d3a5a5 100644
--- a/ami/jobs/serializers.py
+++ b/ami/jobs/serializers.py
@@ -192,6 +192,23 @@ class Meta:
         fields = ["id", "pipeline_slug"]
 
 
+class JobChoiceSerializer(DefaultSerializer):
+    """What a job dropdown needs to name a job.
+
+    The job counterpart of ``SourceImageCollectionNestedSerializer``, which serves the capture
+    set choices: no counts or nested objects, so the choices query stays cheap.
+    """
+
+    def get_permissions(self, instance, instance_data):
+        # A picker needs no per-job permissions, and resolving them costs several queries per row.
+        instance_data["user_permissions"] = []
+        return instance_data
+
+    class Meta:
+        model = Job
+        fields = ["id", "name", "details", "job_type_key", "created_at"]
+
+
 class MLJobTasksRequestSerializer(serializers.Serializer):
     """POST /jobs/{id}/tasks/ — request body sent by a processing service to fetch work.
 
diff --git a/ami/jobs/tests/test_jobs.py b/ami/jobs/tests/test_jobs.py
index 00b7934a7..fbed09b8f 100644
--- a/ami/jobs/tests/test_jobs.py
+++ b/ami/jobs/tests/test_jobs.py
@@ -9,6 +9,7 @@
 
 from ami.base.serializers import reverse_with_params
 from ami.jobs.models import (
+    DataExportJob,
     DataStorageSyncJob,
     Job,
     JobDispatchMode,
@@ -16,6 +17,7 @@
     JobProgress,
     JobState,
     MLJob,
+    PostProcessingJob,
     RegroupEventsJob,
     SourceImageCollectionPopulateJob,
 )
@@ -1744,3 +1746,91 @@ def test_browsable_page_renders_number_input(self):
         html = response.content.decode()
         self.assertNotIn('<select name="source_image_single"', html)
         self.assertIn('<input type="number" name="source_image_single"', html)
+
+
+class TestJobChoices(APITestCase):
+    """Choices endpoint for the occurrence job filter.
+
+    Pins what the dropdown relies on: only jobs that can write detections or
+    classifications, most recently created first, one capped response, and failed
+    jobs included because they may have written results before failing.
+    """
+
+    def setUp(self) -> None:
+        import datetime
+
+        from django.utils import timezone
+
+        self.user = User.objects.create_user(email="job-picker@insectai.org", is_staff=False)
+        self.project = Project.objects.create(name="Job picker project", owner=self.user)
+        other_project = Project.objects.create(name="Other job picker project", owner=self.user)
+        self.oldest = Job.objects.create(project=self.project, name="Oldest", job_type_key=MLJob.key)
+        self.failed = Job.objects.create(
+            project=self.project, name="Failed", job_type_key=PostProcessingJob.key, status=JobState.FAILURE
+        )
+        self.newest = Job.objects.create(project=self.project, name="Newest", job_type_key=MLJob.key)
+        for days_ago, job in enumerate([self.newest, self.failed, self.oldest]):
+            Job.objects.filter(pk=job.pk).update(created_at=timezone.now() - datetime.timedelta(days=days_ago * 10))
+        Job.objects.create(project=self.project, name="Export", job_type_key=DataExportJob.key)
+        Job.objects.create(project=other_project, name="Elsewhere", job_type_key=MLJob.key)
+        self.url = f"/api/v2/jobs/choices/?project_id={self.project.pk}"
+
+    def test_result_writing_jobs_most_recent_first(self):
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual([row["name"] for row in response.json()["results"]], ["Newest", "Failed", "Oldest"])
+
+    def test_a_project_is_required(self):
+        response = self.client.get("/api/v2/jobs/choices/")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_jobs_in_a_draft_project_are_listed_for_members_only(self):
+        draft_project = Project.objects.create(name="Draft job picker project", owner=self.user, draft=True)
+        Job.objects.create(project=draft_project, name="Secret", job_type_key=MLJob.key)
+        url = f"/api/v2/jobs/choices/?project_id={draft_project.pk}"
+        self.assertEqual(self.client.get(url).status_code, status.HTTP_404_NOT_FOUND)
+
+        self.client.force_authenticate(self.user)  # The owner is a member of their project.
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual([row["name"] for row in response.json()["results"]], ["Secret"])
+
+    def test_query_count_does_not_grow_with_the_number_of_jobs(self):
+        """No per-job work: listing eight jobs runs the same queries as listing three, for a member
+        whose permissions would otherwise be resolved on every row. Cachalot is off so every query
+        counts."""
+        from cachalot.api import cachalot_disabled
+        from django.db import connection
+        from django.test.utils import CaptureQueriesContext
+
+        self.client.force_authenticate(self.user)  # The owner is a member of their project.
+
+        def list_choices() -> tuple[int, int]:
+            with CaptureQueriesContext(connection) as queries:
+                response = self.client.get(self.url)
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            return len(response.json()["results"]), len(queries.captured_queries)
+
+        disabled = cachalot_disabled()
+        disabled.__enter__()
+        try:
+            three_rows, three_rows_queries = list_choices()
+            Job.objects.bulk_create(
+                Job(project=self.project, name=f"More {index}", job_type_key=MLJob.key) for index in range(5)
+            )
+            eight_rows, eight_rows_queries = list_choices()
+        finally:
+            # cachalot_disabled() does not restore itself when the block raises.
+            disabled.__exit__(None, None, None)
+        self.assertEqual((three_rows, eight_rows), (3, 8))
+        self.assertEqual(eight_rows_queries, three_rows_queries)
+
+    def test_a_dropdown_gets_one_capped_response_instead_of_pages(self):
+        Job.objects.bulk_create(
+            Job(project=self.project, name=f"Bulk {index}", job_type_key=MLJob.key) for index in range(120)
+        )
+        for query in ("", "&limit=500"):
+            with self.subTest(query=query):
+                body = self.client.get(f"{self.url}{query}").json()
+                self.assertEqual(len(body["results"]), 100)
+                self.assertEqual(body["count"], 123)
diff --git a/ami/jobs/views.py b/ami/jobs/views.py
index 380d9f959..09b314e4b 100644
--- a/ami/jobs/views.py
+++ b/ami/jobs/views.py
@@ -12,7 +12,7 @@
 from drf_spectacular.utils import extend_schema, extend_schema_view
 from rest_framework import serializers
 from rest_framework.decorators import action
-from rest_framework.exceptions import PermissionDenied, ValidationError
+from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError
 from rest_framework.filters import BaseFilterBackend
 from rest_framework.response import Response
 
@@ -37,11 +37,12 @@
     update_pipeline_pull_services_seen,
 )
 from ami.main.api.schemas import project_id_doc_param
-from ami.main.api.views import DefaultViewSet
+from ami.main.api.views import ChoicesPagination, DefaultViewSet
+from ami.main.models import Project
 from ami.utils.fields import url_boolean_param
 
-from .models import Job, JobDispatchMode, JobState
-from .serializers import JobListSerializer, JobSerializer, MinimalJobSerializer
+from .models import Job, JobDispatchMode, JobState, MLJob, PostProcessingJob
+from .serializers import JobChoiceSerializer, JobListSerializer, JobSerializer, MinimalJobSerializer
 
 logger = logging.getLogger(__name__)
 
@@ -257,6 +258,31 @@ def get_serializer_context(self):
             )
         return context
 
+    @extend_schema(parameters=[project_id_doc_param], responses=JobChoiceSerializer(many=True))
+    @action(detail=False, methods=["get"], name="choices")
+    def choices(self, request):
+        """Choices for the occurrence job filter: jobs that can write detections or classifications.
+
+        Follows the capture set choices pattern (``SourceImageCollectionViewSet.choices`` in
+        ami/main/api/views.py, #1381): a slim serializer, most recent first, and one response
+        capped by ``ChoicesPagination``. Failed and older jobs are included, since they may
+        have written results.
+        """
+        project = self.get_active_project()
+        if project is None:
+            raise ValidationError({"project_id": "This parameter is required."})
+        if not Project.objects.visible_for_user(request.user).filter(pk=project.pk).exists():
+            raise NotFound("Project not found.")
+        queryset = (
+            Job.objects.filter(project=project, job_type_key__in=[MLJob.key, PostProcessingJob.key])
+            .only("id", "name", "job_type_key", "created_at", "project_id")
+            .order_by("-created_at", "-pk")
+        )
+        paginator = ChoicesPagination()
+        page = paginator.paginate_queryset(queryset, request, view=self)
+        serializer = JobChoiceSerializer(page, many=True, context=self.get_serializer_context())
+        return paginator.get_paginated_response(serializer.data)
+
     @action(detail=True, methods=["post"], name="run")
     def run(self, request, pk=None):
         """
diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py
index 962c802af..1ca588980 100644
--- a/ami/main/api/serializers.py
+++ b/ami/main/api/serializers.py
@@ -1105,6 +1105,7 @@ class ClassificationSerializer(DefaultSerializer):
     algorithm = AlgorithmSerializer(read_only=True)
     top_n = ClassificationPredictionItemSerializer(many=True, read_only=True)
     applied_to = ClassificationAppliedToSerializer(read_only=True)
+    job = serializers.PrimaryKeyRelatedField(read_only=True, help_text="The job that wrote this classification.")
 
     class Meta:
         model = Classification
@@ -1118,6 +1119,7 @@ class Meta:
             "logits",
             "top_n",
             "applied_to",
+            "job",
             "created_at",
             "updated_at",
         ]
@@ -1263,6 +1265,7 @@ class DetectionSerializer(DefaultSerializer):
         queryset=Algorithm.objects.all(), source="detection_algorithm", write_only=True
     )
     classifications = ClassificationNestedSerializer(many=True, read_only=True)
+    job = serializers.PrimaryKeyRelatedField(read_only=True, help_text="The job that wrote this detection.")
 
     class Meta:
         model = Detection
@@ -1271,6 +1274,7 @@ class Meta:
             "detection_algorithm",
             "detection_algorithm_id",
             "classifications",
+            "job",
         ]
 
 
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index fdc72c963..99709eb29 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -942,8 +942,8 @@ def retrieve(self, request, pk=None):
         return response
 
 
-class CaptureSetChoicesPagination(LimitOffsetPaginationWithPermissions):
-    """Sends a whole project's capture set choices in one response.
+class ChoicesPagination(LimitOffsetPaginationWithPermissions):
+    """Sends a dropdown's choices in one response. Used by the capture set and job choices.
 
     Dropdowns cannot page, so the limit is set here rather than by each caller. It is
     capped as well as defaulted, so no caller can ask for a larger one.
@@ -1031,11 +1031,12 @@ def choices(self, request: Request) -> Response:
         Returns only what those consumers need to name a capture set, most recently
         updated first, and enough of them that a dropdown never has to page. A project
         with more capture sets than the cap needs the search field in #1380.
+        ``JobViewSet.choices`` (ami/jobs/views.py) follows the same pattern.
         """
         # Sorting by a count would fail here, since the counts are never annotated.
         self.ordering_fields = ["id", "created_at", "updated_at", "name", "method"]
         queryset = self.filter_queryset(self.get_queryset())
-        paginator = CaptureSetChoicesPagination()
+        paginator = ChoicesPagination()
         page = paginator.paginate_queryset(queryset, request, view=self)
         serializer = SourceImageCollectionNestedSerializer(page, many=True, context=self.get_serializer_context())
         return paginator.get_paginated_response(serializer.data)
@@ -1448,10 +1449,29 @@ def filter_queryset(self, request, queryset, view):
         return queryset
 
 
+class OccurrenceJobFilter(filters.BaseFilterBackend):
+    """
+    Filter occurrences created or updated by a job.
+    """
+
+    query_param = "job"
+
+    def filter_queryset(self, request, queryset, view):
+        job_id = SingleParamSerializer[int].clean(
+            param_name=self.query_param,
+            field=serializers.IntegerField(required=False, min_value=1),
+            data=request.query_params,
+        )
+        if job_id is None:
+            return queryset
+        return queryset.created_or_updated_by_job(job_id)
+
+
 OCCURRENCE_FILTER_BACKENDS = (
     CustomOccurrenceDeterminationFilter,
     OccurrenceCollectionFilter,
     OccurrenceAlgorithmFilter,
+    OccurrenceJobFilter,
     OccurrenceDateFilter,
     OccurrenceVerified,
     OccurrenceVerifiedByMeFilter,
@@ -1558,6 +1578,12 @@ def get_queryset(self) -> QuerySet["Occurrence"]:
                 required=False,
                 type=OpenApiTypes.INT,
             ),
+            OpenApiParameter(
+                name="job",
+                description="Filter occurrences created or updated by a job.",
+                required=False,
+                type=OpenApiTypes.INT,
+            ),
         ]
     )
     def list(self, request, *args, **kwargs):
diff --git a/ami/main/migrations/0096_detection_and_classification_job.py b/ami/main/migrations/0096_detection_and_classification_job.py
new file mode 100644
index 000000000..782fbd325
--- /dev/null
+++ b/ami/main/migrations/0096_detection_and_classification_job.py
@@ -0,0 +1,43 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    """Record the job that created each detection and classification.
+
+    Adding a nullable foreign key column is a catalogue change: Postgres does not scan
+    the table, so the lock on these two large tables is held only briefly. The column
+    is indexed in 0097, concurrently, so no index is built inside this transaction.
+    """
+
+    dependencies = [
+        ("jobs", "0023_alter_job_job_type_key"),
+        ("main", "0095_grant_sync_deployment_to_mldatamanager"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="classification",
+            name="job",
+            field=models.ForeignKey(
+                blank=True,
+                db_index=False,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="classifications",
+                to="jobs.job",
+            ),
+        ),
+        migrations.AddField(
+            model_name="detection",
+            name="job",
+            field=models.ForeignKey(
+                blank=True,
+                db_index=False,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="detections",
+                to="jobs.job",
+            ),
+        ),
+    ]
diff --git a/ami/main/migrations/0097_detection_and_classification_job_indexes.py b/ami/main/migrations/0097_detection_and_classification_job_indexes.py
new file mode 100644
index 000000000..459e1da80
--- /dev/null
+++ b/ami/main/migrations/0097_detection_and_classification_job_indexes.py
@@ -0,0 +1,47 @@
+from django.contrib.postgres.operations import AddIndexConcurrently
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    """Index the job columns added in 0096 for the ``?job=`` occurrence filter.
+
+    Each index covers what the filter reads (job and occurrence, or job and detection),
+    so the filter is answered from the index alone. Both are partial: rows written before
+    jobs were recorded have no job and are left out, which keeps the build short and the
+    index small. Detection and classification are the largest tables, so the indexes are
+    built CONCURRENTLY, which needs a non-atomic migration; see 0093 for why the statement
+    timeout is cleared and restored around the build.
+    """
+
+    atomic = False
+
+    dependencies = [
+        ("main", "0096_detection_and_classification_job"),
+    ]
+
+    operations = [
+        migrations.RunSQL(
+            sql="SET statement_timeout = 0;",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+        AddIndexConcurrently(
+            model_name="detection",
+            index=models.Index(
+                condition=models.Q(job__isnull=False),
+                fields=["job", "occurrence"],
+                name="det_job_occurrence_idx",
+            ),
+        ),
+        AddIndexConcurrently(
+            model_name="classification",
+            index=models.Index(
+                condition=models.Q(job__isnull=False),
+                fields=["job", "detection"],
+                name="cls_job_detection_idx",
+            ),
+        ),
+        migrations.RunSQL(
+            sql="RESET statement_timeout;",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]
diff --git a/ami/main/models.py b/ami/main/models.py
index 9943c111a..a4c0c82f1 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -2995,7 +2995,16 @@ class Classification(BaseModel):
         null=True,
         related_name="classifications",
     )
-    # job = models.CharField(max_length=255, null=True)
+    # The job that created this row. A later job that returns the same result reuses the row and
+    # does not change it. Indexed by the partial index in Meta, not by the foreign key's own index.
+    job = models.ForeignKey(
+        "jobs.Job",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="classifications",
+        db_index=False,
+    )
     applied_to = models.ForeignKey(
         "self",
         on_delete=models.SET_NULL,
@@ -3015,6 +3024,13 @@ class Classification(BaseModel):
 
     class Meta:
         ordering = ["-created_at", "-score"]
+        indexes = [
+            # Serves the ?job= occurrence filter with an index-only scan. Partial, because rows
+            # written before jobs were recorded have no job. See #1471 for the measurements.
+            models.Index(
+                fields=["job", "detection"], name="cls_job_detection_idx", condition=models.Q(job__isnull=False)
+            ),
+        ]
 
     def __str__(self) -> str:
         terminal = "Terminal" if self.terminal else "Intermediate"
@@ -3189,11 +3205,16 @@ class Detection(BaseModel):
     # @TODO not sure if this detection score is ever used
     # I think it was intended to be the score of the detection algorithm (bbox score)
     detection_score = models.FloatField(null=True, blank=True)
-    # detection_job = models.ForeignKey(
-    #     "Job",
-    #     on_delete=models.SET_NULL,
-    #     null=True,
-    # )
+    # The job that created this row. A later job that returns the same box reuses the row and
+    # does not change it. Indexed by the partial index in Meta, not by the foreign key's own index.
+    job = models.ForeignKey(
+        "jobs.Job",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="detections",
+        db_index=False,
+    )
 
     similarity_vector = models.JSONField(null=True, blank=True)
 
@@ -3275,6 +3296,11 @@ class Meta:
             # Supports the "last processed" subquery on the captures list: the
             # latest detection created_at per source image (index scan, top 1).
             models.Index(fields=["source_image", "-created_at"], name="det_srcimg_created_idx"),
+            # Serves the ?job= occurrence filter with an index-only scan. Partial, because rows
+            # written before jobs were recorded have no job. See #1471 for the measurements.
+            models.Index(
+                fields=["job", "occurrence"], name="det_job_occurrence_idx", condition=models.Q(job__isnull=False)
+            ),
         ]
 
     def best_classification(self):
@@ -3388,6 +3414,20 @@ def not_processed_by_algorithm(self, algorithm_ids) -> "OccurrenceQuerySet":
         """Occurrences with no result from any of the given algorithms."""
         return self.exclude(self._processed_by_algorithm_q(algorithm_ids))
 
+    def created_or_updated_by_job(self, job_id: int) -> "OccurrenceQuerySet":
+        """Occurrences created or updated by the given job.
+
+        An occurrence matches when the job created one of its detections, or a classification
+        on one of them, so several jobs can match the same occurrence.
+
+        Identifications are not matched: people make them, not jobs. Two EXISTS subqueries
+        return each occurrence once, where a join would return one row per matching result.
+        """
+        return self.filter(
+            Exists(Detection.objects.filter(occurrence=OuterRef("pk"), job_id=job_id))
+            | Exists(Classification.objects.filter(detection__occurrence=OuterRef("pk"), job_id=job_id))
+        )
+
     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 aab6d943d..27785fe7f 100644
--- a/ami/main/tests.py
+++ b/ami/main/tests.py
@@ -7242,6 +7242,117 @@ def test_exclude_is_the_complement_of_include(self):
         self.assertEqual(included & excluded, set())
 
 
+class TestOccurrenceJobFilter(APITestCase):
+    """
+    Covers the ``?job=`` occurrence filter: occurrences with a detection or a
+    classification written by the given job. Each occurrence must appear once, however
+    many of its rows the job wrote, and a malformed id must be a 400, not a 500.
+    """
+
+    def setUp(self):
+        from ami.main.models import Taxon
+        from ami.ml.models.algorithm import Algorithm
+
+        self.project = Project.objects.create(name="Occurrence Job 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-job-filter.jpg",
+        )
+        self.taxon = Taxon.objects.create(name="Occurrence Job Filter Taxon")
+        self.algorithm = Algorithm.objects.create(name="Job Filter Algorithm", version=1, task_type="classification")
+        self.job = Job.objects.create(project=self.project, name="Job under test")
+        self.other_job = Job.objects.create(project=self.project, name="Other job")
+
+        # The job wrote the detection only.
+        self.occ_detected = self._make_occurrence([(self.job, [None])])
+        # The job wrote a classification on another job's detection.
+        self.occ_classified = self._make_occurrence([(self.other_job, [self.job])])
+        # The job wrote three detections and their classifications: the duplicate-row trap.
+        self.occ_multi = self._make_occurrence([(self.job, [self.job, self.job])] * 3)
+        # Only the other job, and no job at all, must not match.
+        self.occ_other = self._make_occurrence([(self.other_job, [self.other_job])])
+        self.occ_none = self._make_occurrence([(None, [None])])
+
+    def _make_occurrence(self, detections) -> Occurrence:
+        """``detections`` is a list of (detection job, [classification jobs])."""
+        occ = Occurrence.objects.create(
+            project=self.project,
+            event=self.event,
+            deployment=self.deployment,
+            determination=self.taxon,
+            determination_score=0.9,
+        )
+        for detection_job, classification_jobs in detections:
+            detection = Detection.objects.create(
+                source_image=self.source_image,
+                bbox=[0.0, 0.0, 1.0, 1.0],
+                occurrence=occ,
+                job=detection_job,
+            )
+            for classification_job in classification_jobs:
+                detection.classifications.create(
+                    taxon=self.taxon,
+                    algorithm=self.algorithm,
+                    score=0.9,
+                    timestamp=datetime.datetime.now(),
+                    job=classification_job,
+                )
+        return occ
+
+    def _list(self, job_param):
+        return self.client.get(f"/api/v2/occurrences/?project_id={self.project.pk}&job={job_param}&limit=50")
+
+    def test_returns_each_occurrence_the_job_wrote_once(self):
+        response = self._list(self.job.pk)
+        self.assertEqual(response.status_code, 200)
+        ids = [row["id"] for row in response.json()["results"]]
+        self.assertEqual(len(ids), len(set(ids)))
+        self.assertEqual(set(ids), {self.occ_detected.pk, self.occ_classified.pk, self.occ_multi.pk})
+        self.assertEqual(response.json()["count"], 3)
+
+    def test_other_job_matches_only_its_own_occurrences(self):
+        response = self._list(self.other_job.pk)
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.json()["results"]}
+        self.assertEqual(ids, {self.occ_classified.pk, self.occ_other.pk})
+
+    def test_non_integer_job_is_a_bad_request(self):
+        response = self._list("abc")
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("job", response.json())
+
+    def test_filter_stays_inside_the_occurrence_query(self):
+        """The filter adds no queries of its own: one statement for the rows and one for the count,
+        with no ids read into Python first and no query per occurrence. Cachalot is off so every
+        query counts."""
+        from cachalot.api import cachalot_disabled
+
+        def filtered():
+            return Occurrence.objects.filter(project=self.project).created_or_updated_by_job(self.job.pk)
+
+        disabled = cachalot_disabled()
+        disabled.__enter__()
+        try:
+            with self.assertNumQueries(1):
+                ids = {occurrence.pk for occurrence in filtered()}
+            with self.assertNumQueries(1):
+                count = filtered().count()
+        finally:
+            # cachalot_disabled() does not restore itself when the block raises.
+            disabled.__exit__(None, None, None)
+        self.assertEqual(ids, {self.occ_detected.pk, self.occ_classified.pk, self.occ_multi.pk})
+        self.assertEqual(count, 3)
+
+
 class TestCleanupNullOnlyOccurrencesCommand(TestCase):
     """
     Covers ami/main/management/commands/cleanup_null_only_occurrences.py.
diff --git a/ami/ml/models/pipeline.py b/ami/ml/models/pipeline.py
index 8ebaf6d6d..3d5b77d58 100644
--- a/ami/ml/models/pipeline.py
+++ b/ami/ml/models/pipeline.py
@@ -523,6 +523,7 @@ def get_or_create_detection(
     algorithms_known: dict[str, Algorithm],
     save: bool = True,
     logger: logging.Logger = logger,
+    job_id: int | None = None,
 ) -> tuple[Detection, bool]:
     """
     Create a Detection object from a DetectionResponse, or update an existing one.
@@ -616,6 +617,7 @@ def get_or_create_detection(
             path=crop_url,
             detection_time=detection_resp.timestamp,
             detection_algorithm=detection_algo,
+            job_id=job_id,
         )
         if save:
             new_detection.save()
@@ -633,6 +635,7 @@ def create_detections(
     detections: list[DetectionResponse],
     algorithms_known: dict[str, Algorithm],
     logger: logging.Logger = logger,
+    job_id: int | None = None,
 ) -> list[Detection]:
     """
     Efficiently create multiple Detection objects from a list of DetectionResponse objects, grouped by source image.
@@ -663,6 +666,7 @@ def create_detections(
             algorithms_known=algorithms_known,
             save=False,
             logger=logger,
+            job_id=job_id,
         )
         if created:
             new_detections.append(detection)
@@ -756,6 +760,7 @@ def create_classification(
     algorithms_known: dict[str, Algorithm],
     save: bool = True,
     logger: logging.Logger = logger,
+    job_id: int | None = None,
 ) -> tuple[Classification, bool]:
     """
     Create a Classification object from a ClassificationResponse, or update an existing one.
@@ -843,6 +848,7 @@ def create_classification(
             scores=classification_resp.scores,
             terminal=classification_resp.terminal,
             category_map=classification_algo.category_map,
+            job_id=job_id,
         )
         classification = new_classification
 
@@ -861,6 +867,7 @@ def create_classifications(
     algorithms_known: dict[str, Algorithm],
     logger: logging.Logger = logger,
     save: bool = True,
+    job_id: int | None = None,
 ) -> list[Classification]:
     """
     Efficiently create multiple Classification objects from a list of ClassificationResponse objects,
@@ -885,6 +892,7 @@ def create_classifications(
                 algorithms_known=algorithms_known,
                 save=False,
                 logger=logger,
+                job_id=job_id,
             )
             if created:
                 new_classifications.append(classification)
@@ -1071,10 +1079,12 @@ def save_results(
             "Algorithms and category maps must be registered before processing, using /info endpoint."
         )
 
+    # New rows record the job that wrote them; rows that already existed keep theirs.
     detections = create_detections(
         detections=results.detections,
         algorithms_known=algorithms_known,
         logger=job_logger,
+        job_id=job.pk if job else None,
     )
 
     classifications = create_classifications(
@@ -1082,6 +1092,7 @@ def save_results(
         detection_responses=results.detections,
         algorithms_known=algorithms_known,
         logger=job_logger,
+        job_id=job.pk if job else None,
     )
 
     # Create a new occurrence for each detection (no tracking yet)
@@ -1123,6 +1134,7 @@ def save_results(
         detections=null_detection_responses,
         algorithms_known=algorithms_known,
         logger=job_logger,
+        job_id=job.pk if job else None,
     )
 
     total_time = time.time() - start_time
diff --git a/ami/ml/post_processing/class_masking.py b/ami/ml/post_processing/class_masking.py
index 2da2001b7..18b22c26d 100644
--- a/ami/ml/post_processing/class_masking.py
+++ b/ami/ml/post_processing/class_masking.py
@@ -1,4 +1,5 @@
 import logging
+import typing
 from collections.abc import Callable
 
 import numpy as np
@@ -11,6 +12,9 @@
 from ami.ml.models.algorithm import Algorithm, AlgorithmTaskType
 from ami.ml.post_processing.base import BasePostProcessingTask
 
+if typing.TYPE_CHECKING:
+    from ami.jobs.models import Job
+
 logger = logging.getLogger(__name__)
 
 
@@ -52,6 +56,7 @@ def make_classifications_filtered_by_taxa_list(
     task_logger: logging.Logger = logger,
     on_setup: Callable[[int], None] | None = None,
     on_batch: Callable[[dict], None] | None = None,
+    job: "Job | None" = None,
 ) -> dict[str, int]:
     """Re-score ``classifications`` by masking out classes absent from ``taxa_list``.
 
@@ -74,6 +79,8 @@ def make_classifications_filtered_by_taxa_list(
     ``occurrences_updated`` counts only occurrences whose determination actually
     changed (not just any occurrence touched), matching the size-filter convention.
 
+    New classifications record ``job`` as the run that wrote them.
+
     Returns final counters (checked / masked / occurrences updated) for stage metrics.
     """
     taxa_in_list = set(taxa_list.taxa.all())
@@ -180,6 +187,7 @@ def make_classifications_filtered_by_taxa_list(
                     terminal=True,
                     timestamp=classification.timestamp,
                     applied_to=classification,
+                    job=job,
                     created_at=timestamp,
                     updated_at=timestamp,
                 )
@@ -368,6 +376,7 @@ def _on_batch(m: dict) -> None:
             task_logger=self.logger,
             on_setup=_on_setup,
             on_batch=_on_batch,
+            job=self.job,
         )
         self.report_stage_metrics(metrics)
         self.logger.info(f"=== Completed {self.name} ===")
diff --git a/ami/ml/post_processing/small_size_filter.py b/ami/ml/post_processing/small_size_filter.py
index 6a39af780..f0b8289eb 100644
--- a/ami/ml/post_processing/small_size_filter.py
+++ b/ami/ml/post_processing/small_size_filter.py
@@ -126,6 +126,7 @@ def run(self) -> None:
                         timestamp=timezone.now(),  # How is this different from created_at?
                         algorithm=self.algorithm,
                         applied_to=None,  # Size filter is applied to original detection, not a previous classification
+                        job=self.job,
                     )
                 )
                 detections_to_update.add(det)
diff --git a/ami/ml/post_processing/tests/test_class_masking.py b/ami/ml/post_processing/tests/test_class_masking.py
index bf38decc9..541a70195 100644
--- a/ami/ml/post_processing/tests/test_class_masking.py
+++ b/ami/ml/post_processing/tests/test_class_masking.py
@@ -253,6 +253,30 @@ 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_task_run_records_the_job_on_new_classifications_only(self):
+        from ami.jobs.models import Job
+
+        logits = [0.5, 3.0, 3.5]
+        taxa_list = TaxaList.objects.create(name="Regional 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)
+        job = Job.objects.create(project=self.project, name="Class masking job test", job_type_key="post_processing")
+        job.progress.add_stage("Post Processing", key="post_processing")
+        job.save()
+
+        ClassMaskingTask(
+            job=job,
+            source_image_collection_id=self.collection.pk,
+            taxa_list_id=taxa_list.pk,
+            algorithm_id=self.algorithm.pk,
+        ).run()
+
+        new_clf = Classification.objects.get(detection=det, terminal=True, applied_to=original)
+        self.assertEqual(new_clf.job_id, job.pk)
+        original.refresh_from_db()
+        self.assertIsNone(original.job_id, "The source classification keeps the job it had")
+
     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.
diff --git a/ami/ml/tests.py b/ami/ml/tests.py
index bd92bb02f..b5f99a66e 100644
--- a/ami/ml/tests.py
+++ b/ami/ml/tests.py
@@ -672,6 +672,24 @@ def test_save_results(self):
 
         # @TODO test the cached counts for detections, etc are updated on Events, Deployments, etc.
 
+    def test_save_results_records_the_job_on_the_rows_it_creates(self):
+        from ami.jobs.models import Job
+
+        first_job = Job.objects.create(project=self.project, name="First run", pipeline=self.pipeline)
+        second_job = Job.objects.create(project=self.project, name="Second run", pipeline=self.pipeline)
+        results = self.fake_pipeline_results(self.test_images, self.pipeline)
+
+        save_results(results, job_id=first_job.pk)
+        # A second run returning the same boxes and labels reuses the rows, which keep the first job.
+        save_results(results, job_id=second_job.pk)
+
+        detections = Detection.objects.filter(source_image__in=self.test_images)
+        classifications = Classification.objects.filter(detection__in=detections)
+        self.assertTrue(detections.exists())
+        self.assertTrue(classifications.exists())
+        self.assertEqual(set(detections.values_list("job_id", flat=True)), {first_job.pk})
+        self.assertEqual(set(classifications.values_list("job_id", flat=True)), {first_job.pk})
+
     def test_skip_existing_when_all_matching(self):
         """
         When processing images, skip images that have already been processed by the same set of algorithms.
@@ -1603,6 +1621,28 @@ def test_run_reports_stage_metrics_on_job(self):
         # count equals the detection count.
         self.assertEqual(params.get("occurrences_updated"), total)
 
+    def test_new_classifications_record_the_job(self):
+        """The size filter's "Not identifiable" classifications point at the job that ran it."""
+        from ami.jobs.models import Job
+
+        for image in self.collection.images.all():
+            Detection.objects.create(
+                source_image=image,
+                bbox=[0, 0, 10, 10],  # small → flagged
+                created_at=datetime.datetime.now(datetime.timezone.utc),
+            ).associate_new_occurrence()
+        job = Job.objects.create(project=self.project, name="size filter job test", job_type_key="post_processing")
+        job.progress.add_stage("Post Processing", key="post_processing")
+        job.save()
+
+        SmallSizeFilterTask(job=job, source_image_collection_id=self.collection.pk, size_threshold=0.01).run()
+
+        new_classifications = Classification.objects.filter(
+            detection__source_image__in=self.collection.images.all(), taxon__name="Not identifiable"
+        )
+        self.assertTrue(new_classifications.exists())
+        self.assertEqual(set(new_classifications.values_list("job_id", flat=True)), {job.pk})
+
     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.
diff --git a/ui/src/components/filtering/default-filter-control.tsx b/ui/src/components/filtering/default-filter-control.tsx
index ef8673c5e..88471667c 100644
--- a/ui/src/components/filtering/default-filter-control.tsx
+++ b/ui/src/components/filtering/default-filter-control.tsx
@@ -10,6 +10,7 @@ import {
   Popover,
   Switch,
 } from 'nova-ui-kit'
+import { useId } from 'react'
 import { Link, useParams } from 'react-router-dom'
 import { APP_ROUTES } from 'utils/constants'
 import { STRING, translate } from 'utils/language'
@@ -21,6 +22,7 @@ export const DefaultFiltersControl = ({ field }: { field: string }) => {
   const { project } = useProjectDetails(projectId as string, true)
   const { filters, addFilter, clearFilter } = useFilters()
   const filter = filters.find((filter) => filter.field === field)
+  const labelId = useId()
 
   if (!filter) {
     return null
@@ -29,12 +31,16 @@ export const DefaultFiltersControl = ({ field }: { field: string }) => {
   return (
     <div className="flex items-center justify-between pl-2">
       <div className="flex items-center gap-1">
-        <span className="text-muted-foreground body-overline-small font-bold pt-0.5">
+        <span
+          id={labelId}
+          className="text-muted-foreground body-overline-small font-bold pt-0.5"
+        >
           {filter.label}
         </span>
         {project ? <DefaultFiltersTooltip project={project} /> : null}
       </div>
       <Switch
+        aria-labelledby={labelId}
         checked={stringToBoolean(filter.value) ?? true}
         onCheckedChange={(value) => {
           if (value) {
diff --git a/ui/src/components/filtering/filter-control.tsx b/ui/src/components/filtering/filter-control.tsx
index 64dafd9cc..742620c79 100644
--- a/ui/src/components/filtering/filter-control.tsx
+++ b/ui/src/components/filtering/filter-control.tsx
@@ -1,5 +1,6 @@
 import { XIcon } from 'lucide-react'
 import { Button, InfoTooltip } from 'nova-ui-kit'
+import { useId } from 'react'
 import { STRING, translate } from 'utils/language'
 import { useFilters } from 'utils/useFilters'
 import { AlgorithmFilter, NotAlgorithmFilter } from './filters/algorithm-filter'
@@ -17,6 +18,7 @@ import { TagFilter } from './filters/tag-filter'
 import { TaxaListFilter } from './filters/taxa-list-filter'
 import { TaxonFilter } from './filters/taxon-filter'
 import { TypeFilter } from './filters/type-filter'
+import { JobFilter } from './filters/job-filter'
 import { FilterProps } from './filters/types'
 import { ProcessingStatusFilter } from './filters/processing-status-filter'
 import { VerificationStatusFilter } from './filters/verification-status-filter'
@@ -37,6 +39,7 @@ const ComponentMap: {
   event: SessionFilter,
   processed: ProcessingStatusFilter,
   include_unobserved: BooleanFilter,
+  job: JobFilter,
   job_type_key: TypeFilter,
   not_algorithm: NotAlgorithmFilter,
   not_tag_id: TagFilter,
@@ -67,6 +70,7 @@ export const FilterControl = ({
 }: FilterControlProps) => {
   const { filters, addFilter, clearFilter } = useFilters()
   const filter = filters.find((filter) => filter.field === field)
+  const labelId = useId()
   const FilterComponent = ComponentMap[field]
 
   if (!filter || !FilterComponent) {
@@ -80,12 +84,20 @@ export const FilterControl = ({
   return (
     <div>
       <div className="min-h-8 flex items-center gap-1">
-        <span className="text-muted-foreground body-overline-small font-bold pt-0.5">
+        <span
+          id={labelId}
+          className="text-muted-foreground body-overline-small font-bold pt-0.5"
+        >
           {filter.label}
         </span>
         {filter.tooltip ? <InfoTooltip {...filter.tooltip} /> : null}
       </div>
-      <div className="flex items-center justify-between gap-2">
+      {/* Names the control and its clear button for screen readers (WCAG technique ARIA17). */}
+      <div
+        aria-labelledby={labelId}
+        className="flex items-center justify-between gap-2"
+        role="group"
+      >
         <FilterComponent
           data={data}
           error={filter.error}
diff --git a/ui/src/components/filtering/filters/job-filter.tsx b/ui/src/components/filtering/filters/job-filter.tsx
new file mode 100644
index 000000000..9e12bd6d9
--- /dev/null
+++ b/ui/src/components/filtering/filters/job-filter.tsx
@@ -0,0 +1,47 @@
+import { API_ROUTES } from 'data-services/constants'
+import { useEntities } from 'data-services/hooks/entities/useEntities'
+import { useJobChoice } from 'data-services/hooks/jobs/useJobChoice'
+import { Select } from 'nova-ui-kit'
+import { useParams } from 'react-router-dom'
+import { STRING, translate } from 'utils/language'
+import { FilterProps } from './types'
+
+type JobOption = { id: string; name: string }
+
+const getLabel = (job: JobOption) => `${job.name} (#${job.id})`
+
+// Same choices endpoint pattern as CaptureSetFilter; see JobViewSet.choices.
+export const JobFilter = ({ onAdd, onClear, value }: FilterProps) => {
+  const { projectId } = useParams()
+  const { entities = [], isLoading } = useEntities(API_ROUTES.JOB_CHOICES, {
+    projectId: projectId as string,
+  })
+  const options: JobOption[] = entities.map((e) => ({ id: e.id, name: e.name }))
+  const missing = !!value && !isLoading && !options.some((o) => o.id === value)
+
+  // A job linked from its details page may be older than the listed choices.
+  const { job: selectedJob } = useJobChoice(value, missing)
+  if (missing && selectedJob) {
+    options.push(selectedJob)
+  }
+
+  return (
+    <Select.Root
+      key={value}
+      disabled={isLoading || options.length === 0}
+      onValueChange={(value) => (value ? onAdd(value) : onClear())}
+      value={options.some((o) => o.id === value) ? value : ''}
+    >
+      <Select.Trigger loading={isLoading}>
+        <Select.Value placeholder={translate(STRING.SELECT_PLACEHOLDER)} />
+      </Select.Trigger>
+      <Select.Content className="max-h-72">
+        {options.map((o) => (
+          <Select.Item key={o.id} value={o.id}>
+            {getLabel(o)}
+          </Select.Item>
+        ))}
+      </Select.Content>
+    </Select.Root>
+  )
+}
diff --git a/ui/src/data-services/constants.ts b/ui/src/data-services/constants.ts
index 423a931b0..59bae6ede 100644
--- a/ui/src/data-services/constants.ts
+++ b/ui/src/data-services/constants.ts
@@ -11,6 +11,7 @@ export const API_ROUTES = {
   EXPORTS: 'exports',
   IDENTIFICATIONS: 'identifications',
   JOBS: 'jobs',
+  JOB_CHOICES: 'jobs/choices',
   LOGIN: 'auth/token/login',
   LOGOUT: 'auth/token/logout',
   ME: 'users/me',
diff --git a/ui/src/data-services/hooks/jobs/useJobChoice.ts b/ui/src/data-services/hooks/jobs/useJobChoice.ts
new file mode 100644
index 000000000..b96ddecd7
--- /dev/null
+++ b/ui/src/data-services/hooks/jobs/useJobChoice.ts
@@ -0,0 +1,19 @@
+import { API_ROUTES, API_URL } from 'data-services/constants'
+import { ServerJob } from 'data-services/models/job'
+import { useAuthorizedQuery } from '../auth/useAuthorizedQuery'
+
+/**
+ * Load the name of one job for a job picker, without the polling the job details page uses.
+ */
+export const useJobChoice = (jobId?: string, enabled = true) => {
+  const { data, isLoading } = useAuthorizedQuery<ServerJob>({
+    enabled: enabled && !!jobId,
+    queryKey: [API_ROUTES.JOBS, jobId, 'choice'],
+    url: `${API_URL}/${API_ROUTES.JOBS}/${jobId}/?logs_limit=1`,
+  })
+
+  return {
+    job: data ? { id: `${data.id}`, name: data.name } : undefined,
+    isLoading,
+  }
+}
diff --git a/ui/src/data-services/models/job.ts b/ui/src/data-services/models/job.ts
index 15c59ea53..8d919e66a 100644
--- a/ui/src/data-services/models/job.ts
+++ b/ui/src/data-services/models/job.ts
@@ -21,6 +21,7 @@ export const SERVER_JOB_TYPES = [
   'data_storage_sync',
   'populate_captures_collection',
   'data_export',
+  'post_processing',
   'unknown',
 ] as const
 
@@ -164,6 +165,7 @@ export class Job extends Entity {
       data_storage_sync: 'Data storage sync',
       populate_captures_collection: 'Populate captures collection',
       data_export: 'Data export',
+      post_processing: 'Post-processing',
       unknown: 'Unknown',
     }[key]
 
diff --git a/ui/src/pages/job-details/job-details.tsx b/ui/src/pages/job-details/job-details.tsx
index 4127576f3..14e05e4ad 100644
--- a/ui/src/pages/job-details/job-details.tsx
+++ b/ui/src/pages/job-details/job-details.tsx
@@ -109,6 +109,16 @@ const JobSummary = ({ job }: { job: Job }) => {
             value={job.pipeline.name}
           />
         ) : null}
+        {job.type.key === 'ml' || job.type.key === 'post_processing' ? (
+          <InputValue
+            label={translate(STRING.FIELD_LABEL_OCCURRENCES)}
+            value={translate(STRING.VIEW_OCCURRENCES)}
+            to={getAppRoute({
+              to: APP_ROUTES.OCCURRENCES({ projectId: projectId as string }),
+              filters: { job: job.id },
+            })}
+          />
+        ) : null}
         {job.sourceImage ? (
           <InputValue
             label={translate(STRING.FIELD_LABEL_CAPTURE)}
diff --git a/ui/src/pages/occurrences/occurrence-filters.ts b/ui/src/pages/occurrences/occurrence-filters.ts
index 61b9e9443..e63d5093c 100644
--- a/ui/src/pages/occurrences/occurrence-filters.ts
+++ b/ui/src/pages/occurrences/occurrence-filters.ts
@@ -18,5 +18,6 @@ export const FILTERS_TO_OCCURRENCES = [
   'deployment__research_site',
   'algorithm',
   'not_algorithm',
+  'job',
   'apply_defaults',
 ]
diff --git a/ui/src/pages/occurrences/occurrences.tsx b/ui/src/pages/occurrences/occurrences.tsx
index af53cb098..e0aed9b24 100644
--- a/ui/src/pages/occurrences/occurrences.tsx
+++ b/ui/src/pages/occurrences/occurrences.tsx
@@ -92,6 +92,7 @@ export const Occurrences = () => {
         <div className="space-y-6">
           <OccurrenceStats projectId={projectId} filters={filters} />
           <FilterSection defaultOpen>
+            <DefaultFiltersControl field="apply_defaults" />
             <FilterControl field="detections__source_image" readonly />
             <FilterControl field="event" readonly />
             <FilterControl field="taxon" />
@@ -103,7 +104,6 @@ export const Occurrences = () => {
             )}
             <FilterControl field="verified" />
             {user.loggedIn && <FilterControl field="verified_by_me" />}
-            <DefaultFiltersControl field="apply_defaults" />
           </FilterSection>
           <FilterSection
             title={translate(STRING.MORE_FILTERS)}
@@ -115,6 +115,7 @@ export const Occurrences = () => {
                 'deployment__research_site',
                 'algorithm',
                 'not_algorithm',
+                'job',
               ],
               activeFilters
             )}
@@ -127,6 +128,7 @@ export const Occurrences = () => {
             <FilterControl field="deployment__research_site" />
             <FilterControl field="algorithm" />
             <FilterControl field="not_algorithm" />
+            <FilterControl field="job" />
           </FilterSection>
         </div>
         <div className="w-full overflow-hidden">
diff --git a/ui/src/pages/species/species.tsx b/ui/src/pages/species/species.tsx
index 51d2f1083..23c813b90 100644
--- a/ui/src/pages/species/species.tsx
+++ b/ui/src/pages/species/species.tsx
@@ -148,6 +148,7 @@ export const Species = () => {
       <div className="flex flex-col gap-6 md:flex-row">
         <div className="space-y-6">
           <FilterSection defaultOpen>
+            <DefaultFiltersControl field="apply_defaults" />
             <FilterControl field="event" readonly />
             <FilterControl field="taxon" />
             {taxaLists.length > 0 && (
@@ -158,7 +159,6 @@ export const Species = () => {
             )}
             <FilterControl field="verified" />
             <FilterControl field="include_unobserved" />
-            <DefaultFiltersControl field="apply_defaults" />
           </FilterSection>
           <FilterSection
             title={translate(STRING.MORE_FILTERS)}
diff --git a/ui/src/utils/getAppRoute.ts b/ui/src/utils/getAppRoute.ts
index bafc8290c..354b9ed36 100644
--- a/ui/src/utils/getAppRoute.ts
+++ b/ui/src/utils/getAppRoute.ts
@@ -9,6 +9,7 @@ type FilterType =
   | 'detections__source_image'
   | 'event'
   | 'include_unobserved'
+  | 'job'
   | 'occurrence'
   | 'source_image_collection'
   | 'source_image_single'
diff --git a/ui/src/utils/language.ts b/ui/src/utils/language.ts
index feb83bed6..38e20a5e4 100644
--- a/ui/src/utils/language.ts
+++ b/ui/src/utils/language.ts
@@ -277,6 +277,7 @@ export enum STRING {
   /* TOOLTIPS */
   TOOLTIP_ALGORITHM,
   TOOLTIP_CAPTURE_SET,
+  TOOLTIP_JOB_FILTER,
   TOOLTIP_CAPTURE,
   TOOLTIP_DEPLOYMENT,
   TOOLTIP_DEVICE,
@@ -377,6 +378,7 @@ export enum STRING {
   VERIFIED_OCCURRENCES,
   VERIFIED,
   VIEW_IN_SESSION,
+  VIEW_OCCURRENCES,
   VIEW_SETTINGS,
   YOU,
 }
@@ -693,6 +695,8 @@ const ENGLISH_STRINGS: { [key in STRING]: string } = {
     'An algorithm is used in a pipeline for processing captures.',
   [STRING.TOOLTIP_CAPTURE_SET]:
     'A capture set is a group of captures. A capture set contains all or some captures in a project. Capture sets are useful when specifying what captures to process.',
+  [STRING.TOOLTIP_JOB_FILTER]:
+    'Show occurrences created or updated by this job. The 100 most recent processing and post-processing jobs are listed.',
   [STRING.TOOLTIP_CAPTURE]:
     'A capture is a source image, collected from a monitoring station. During processing, detections are extracted from captures.',
   [STRING.TOOLTIP_DEPLOYMENT]:
@@ -809,6 +813,7 @@ const ENGLISH_STRINGS: { [key in STRING]: string } = {
   [STRING.VERIFIED_OCCURRENCES]: 'Verified occurrences',
   [STRING.VERIFIED]: 'Verified',
   [STRING.VIEW_IN_SESSION]: 'View in session',
+  [STRING.VIEW_OCCURRENCES]: 'View occurrences',
   [STRING.VIEW_SETTINGS]: 'View settings',
   [STRING.YOU]: 'You',
 }
diff --git a/ui/src/utils/useFilters.ts b/ui/src/utils/useFilters.ts
index 8e9dd3336..48fa644ba 100644
--- a/ui/src/utils/useFilters.ts
+++ b/ui/src/utils/useFilters.ts
@@ -34,6 +34,17 @@ export const AVAILABLE_FILTERS = (projectId: string): FilterConfig[] => [
       },
     },
   },
+  {
+    label: translate(STRING.FIELD_LABEL_JOB),
+    field: 'job',
+    tooltip: {
+      text: translate(STRING.TOOLTIP_JOB_FILTER),
+      link: {
+        text: translate(STRING.NAV_ITEM_JOBS),
+        to: APP_ROUTES.JOBS({ projectId }),
+      },
+    },
+  },
   {
     label: translate(STRING.FIELD_LABEL_CAPTURE_SET),
     field: 'collection', // This is for viewing occurrences by capture set. @TODO: Can we update this key to "capture_set_id" to streamline?