fix(jobs): show the collect stage as started while a job collects images

collect_images() persists the collect stage as STARTED for every path
(filtered and reprocess-all) immediately after fetching the job, so the UI
shows the stage in progress as soon as collection begins instead of only at
SUCCESS. The existing throttled progress emit (#1322) is unchanged and carries
the STARTED status forward in-memory on the filtered path; reprocess-all has no
per-image loop, so this single mark is its only collect heartbeat.

Adds tests pinning STARTED on both the filtered and reprocess-all paths.

Author: mihow

ami/ml/models/pipeline.py

@@ -233,10 +233,15 @@ def collect_images(
     """
     task_logger = logger
     if job_id:
-        from ami.jobs.models import Job
+        from ami.jobs.models import Job, JobState
 
         job = Job.objects.get(pk=job_id)
         task_logger = job.logger
+        # Persist the collect stage as STARTED for every path: the filtered branch's throttle
+        # below updates only `progress` (never status), and reprocess-all has no loop at all,
+        # so without this neither path reports STARTED while collection runs.
+        job.progress.update_stage("collect", status=JobState.STARTED, progress=0)
+        job.save(update_fields=["progress", "updated_at"])
     else:
         job = None
 

ami/ml/tests.py

@@ -1184,6 +1184,38 @@ def test_filter_processed_images_skips_progress_emission_without_job(self):
 
         self.assertEqual(mock_save.call_count, 0)
 
+    def test_collect_images_marks_collect_started_on_every_path(self):
+        """
+        collect_images() must persist the collect stage as STARTED for every path, otherwise the
+        stage keeps its default status during collection and only reports STARTED once collection
+        finishes. The filtered path's throttle updates `progress` but never `status`, and the
+        reprocess-all path has no per-image loop at all, so neither persists STARTED on its own.
+        """
+        from ami.jobs.models import Job, JobState, MLJob
+        from ami.ml.models.pipeline import collect_images
+
+        for reprocess_all in (False, True):
+            with self.subTest(reprocess_all=reprocess_all):
+                job = Job.objects.create(
+                    project=self.project,
+                    name=f"collect-started test (reprocess_all={reprocess_all})",
+                    pipeline=self.pipeline,
+                    job_type_key=MLJob.key,
+                )
+                images = [SourceImage.objects.create(path=f"cs-{reprocess_all}-{i}.jpg") for i in range(3)]
+
+                list(
+                    collect_images(
+                        source_images=images,
+                        job_id=job.pk,
+                        pipeline=self.pipeline,
+                        reprocess_all_images=reprocess_all,
+                    )
+                )
+
+                job.refresh_from_db()
+                self.assertEqual(job.progress.get_stage("collect").status, JobState.STARTED)
+
     def test_null_detections_are_algorithm_specific(self):
         """
         Null detections from different pipelines/algorithms should not be shared.