Merge remote-tracking branch 'origin/main' into feat/taxa-lists

# Conflicts:
#	ami/main/tests.py
#	ami/ml/serializers.py

Author: mihow

.agents/planning/async-job-status-handling.md

@@ -27,6 +27,10 @@ __pycache__/
 *.egg
 *.whl
 
+# IPython / Jupyter
+.ipython/
+.jupyter/
+
 
 # Django / runtime artifacts
 *.log
@@ -55,6 +59,12 @@ yarn-error.log
 .vscode/
 *.iml
 
+# Development / testing
+.pytest_cache/
+.coverage
+.tox/
+.cache/
+
 # OS cruft
 .DS_Store
 Thumbs.db

.dockerignore

@@ -40,6 +40,22 @@
             ],
             "justMyCode": true
         },
+        {
+            "name": "Attach: Tests",
+            "type": "debugpy",
+            "request": "attach",
+            "connect": {
+                "host": "localhost",
+                "port": 5680
+            },
+            "pathMappings": [
+                {
+                    "localRoot": "${workspaceFolder}",
+                    "remoteRoot": "/app"
+                }
+            ],
+            "justMyCode": true
+        },
         {
             "name": "Current File",
             "type": "debugpy",

.vscode/launch.json

@@ -118,6 +118,32 @@ def has_object_permission(self, request, view, obj: BaseModel):
         return obj.check_permission(request.user, view.action)
 
 
+class ProjectPipelineConfigPermission(ObjectPermission):
+    """
+    Permission for the nested project pipelines route (/projects/{pk}/pipelines/).
+
+    Extends ObjectPermission to handle list/create actions where no object exists yet.
+    Creates a temporary ProjectPipelineConfig instance to leverage BaseModel.check_permission(),
+    which handles draft project visibility and guardian permission checks automatically.
+
+    Follows the same pattern as UserMembershipPermission.
+    """
+
+    def has_permission(self, request, view):
+        from ami.ml.models.project_pipeline_config import ProjectPipelineConfig
+
+        if view.action in ("list", "create"):
+            project = view.get_active_project()
+            if not project:
+                return False
+
+            config = ProjectPipelineConfig(project=project)
+            action = "retrieve" if view.action == "list" else "create"
+            return config.check_permission(request.user, action)
+
+        return super().has_permission(request, view)
+
+
 class UserMembershipPermission(ObjectPermission):
     """
     Custom permission for UserProjectMembershipViewSet.

ami/base/permissions.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.2.10 on 2026-02-04 20:16
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("jobs", "0018_alter_job_job_type_key"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="job",
+            name="dispatch_mode",
+            field=models.CharField(
+                choices=[("internal", "Internal"), ("sync_api", "Sync API"), ("async_api", "Async API")],
+                default="internal",
+                help_text="How the job dispatches its workload: internal, sync_api, or async_api.",
+                max_length=32,
+            ),
+        ),
+    ]

ami/jobs/migrations/0019_job_dispatch_mode.py

@@ -24,6 +24,32 @@
 logger = logging.getLogger(__name__)
 
 
+class JobDispatchMode(models.TextChoices):
+    """
+    How a job dispatches its workload.
+
+    Jobs are configured and launched by users in the UI, then dispatched to
+    Celery workers. This enum describes what the worker does with the work:
+
+    - INTERNAL: All work happens within the platform (Celery worker handles it directly).
+    - SYNC_API: Worker calls an external processing service API and waits for each response.
+    - ASYNC_API: Worker queues items to a message broker (NATS) for external processing
+      service workers to pick up and process independently.
+    """
+
+    # Work is handled entirely within the platform, no external service calls.
+    # e.g. DataStorageSyncJob, DataExportJob, SourceImageCollectionPopulateJob
+    INTERNAL = "internal", "Internal"
+
+    # Worker loops over items, sends each to an external processing service
+    # endpoint synchronously, and waits for the response before continuing.
+    SYNC_API = "sync_api", "Sync API"
+
+    # Worker publishes all items to a message broker (NATS). External processing
+    # service workers consume and process them independently, reporting results back.
+    ASYNC_API = "async_api", "Async API"
+
+
 class JobState(str, OrderedEnum):
     """
     These come from Celery, except for CREATED, which is a custom state.
@@ -83,7 +109,7 @@ def python_slugify(value: str) -> str:
 
 
 class JobProgressSummary(pydantic.BaseModel):
-    """Summary of all stages of a job"""
+    """Top-level status and progress for a job, shown in the UI."""
 
     status: JobState = JobState.CREATED
     progress: float = 0
@@ -106,7 +132,17 @@ class JobProgressStageDetail(ConfigurableStage, JobProgressSummary):
 
 
 class JobProgress(pydantic.BaseModel):
-    """The full progress of a job and its stages."""
+    """
+    The user-facing progress of a job, stored as JSONB on the Job model.
+
+    This is what the UI displays and what external APIs read. Contains named
+    stages ("process", "results") with per-stage params (progress percentage,
+    detections/classifications/captures counts, failed count).
+
+    For async (NATS) jobs, updated by _update_job_progress() in ami/jobs/tasks.py
+    which copies snapshots from the internal Redis-backed AsyncJobStateManager.
+    For sync jobs, updated directly in MLJob.process_images().
+    """
 
     summary: JobProgressSummary
     stages: list[JobProgressStageDetail]
@@ -196,6 +232,34 @@ def reset(self, status: JobState = JobState.CREATED):
         for stage in self.stages:
             stage.progress = 0
             stage.status = status
+            # Reset numeric param values to 0
+            for param in stage.params:
+                if isinstance(param.value, (int, float)):
+                    param.value = 0
+
+    def is_complete(self) -> bool:
+        """
+        Check if all stages have finished processing.
+
+        A job is considered complete when ALL of its stages have:
+        - progress >= 1.0 (fully processed)
+        - status in a final state (SUCCESS, FAILURE, or REVOKED)
+
+        This method works for any job type regardless of which stages it has.
+        It's used by the Celery task_postrun signal to determine whether to
+        set the job's final SUCCESS status, or defer to async progress handlers.
+
+        Related: Job.update_progress() calculates the aggregate
+        progress percentage across all stages for display purposes. This method
+        is a binary check for completion that considers both progress AND status.
+
+        Returns:
+            True if all stages are complete, False otherwise.
+            Returns False if job has no stages (shouldn't happen in practice).
+        """
+        if not self.stages:
+            return False
+        return all(stage.progress >= 1.0 and stage.status in JobState.final_states() for stage in self.stages)
 
     class Config:
         use_enum_values = True
@@ -398,6 +462,8 @@ def run(cls, job: "Job"):
         job.save()
 
         if job.project.feature_flags.async_pipeline_workers:
+            job.dispatch_mode = JobDispatchMode.ASYNC_API
+            job.save(update_fields=["dispatch_mode"])
             queued = queue_images_to_nats(job, images)
             if not queued:
                 job.logger.error("Aborting job %s because images could not be queued to NATS", job.pk)
@@ -407,6 +473,8 @@ def run(cls, job: "Job"):
                 job.save()
                 return
         else:
+            job.dispatch_mode = JobDispatchMode.SYNC_API
+            job.save(update_fields=["dispatch_mode"])
             cls.process_images(job, images)
 
     @classmethod
@@ -507,7 +575,8 @@ def process_images(cls, job, images):
 
         job.logger.info(f"All tasks completed for job {job.pk}")
 
-        FAILURE_THRESHOLD = 0.5
+        from ami.jobs.tasks import FAILURE_THRESHOLD
+
         if image_count and (percent_successful < FAILURE_THRESHOLD):
             job.progress.update_stage("process", status=JobState.FAILURE)
             job.save()
@@ -798,6 +867,12 @@ class Job(BaseModel):
         blank=True,
         related_name="jobs",
     )
+    dispatch_mode = models.CharField(
+        max_length=32,
+        choices=JobDispatchMode.choices,
+        default=JobDispatchMode.INTERNAL,
+        help_text="How the job dispatches its workload: internal, sync_api, or async_api.",
+    )
 
     def __str__(self) -> str:
         return f'#{self.pk} "{self.name}" ({self.status})'

ami/jobs/models.py

@@ -127,6 +127,7 @@ class Meta:
             "job_type",
             "job_type_key",
             "data_export",
+            "dispatch_mode",
             # "duration",
             # "duration_label",
             # "progress_label",
@@ -141,6 +142,7 @@ class Meta:
             "started_at",
             "finished_at",
             "duration",
+            "dispatch_mode",
         ]