fix(jobs): finalize async job when 0 images need processing

ami/jobs/models.py

@@ -478,6 +478,14 @@ def run(cls, job: "Job"):
                 job.finished_at = datetime.datetime.now()
                 job.save()
                 return
+            # When all stages are already complete (e.g. 0 images to process),
+            # finalize the job now since no async results will arrive to trigger completion.
+            if job.progress.is_complete():
+                has_failure = any(s.status in JobState.failed_states() for s in job.progress.stages)
+                job.update_status(JobState.FAILURE if has_failure else JobState.SUCCESS, save=False)
+                job.finished_at = datetime.datetime.now()
+                job.save()
+                cleanup_async_job_if_needed(job)
         else:
             cls.process_images(job, images)
 

ami/jobs/tests/test_jobs.py

@@ -111,6 +111,37 @@ def test_job_status_guard_prevents_premature_success(self):
         self.assertEqual(job.status, initial_status)
         self.assertNotEqual(job.status, JobState.SUCCESS.value)
 
+    def test_async_job_completes_when_zero_images(self):
+        """Job with 0 images to process should finalize immediately, not stay STARTED."""
+        from unittest.mock import patch
+
+        job = Job.objects.create(
+            job_type_key=MLJob.key,
+            project=self.project,
+            name="Test zero images job",
+            pipeline=self.pipeline,
+            source_image_collection=self.source_image_collection,
+            dispatch_mode=JobDispatchMode.ASYNC_API,
+        )
+
+        def mock_queue(job, images):
+            """Simulate queue_images_to_nats with 0 images: sets stages to SUCCESS, returns True."""
+            job.progress.update_stage("process", status=JobState.SUCCESS, progress=1.0)
+            job.progress.update_stage("results", status=JobState.SUCCESS, progress=1.0)
+            job.save()
+            return True
+
+        with (
+            patch.object(job.pipeline, "collect_images", return_value=[]),
+            patch("ami.ml.orchestration.jobs.queue_images_to_nats", side_effect=mock_queue),
+            patch("ami.jobs.tasks.cleanup_async_job_if_needed"),
+        ):
+            job.run()
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobState.SUCCESS.value)
+        self.assertIsNotNone(job.finished_at)
+
     def test_job_status_allows_failure_states_immediately(self):
         """
         Test that FAILURE and REVOKED states bypass the completion guard