feat(ml): propagate pipeline config through NATS pull-mode tasks

ami/jobs/tests/test_jobs.py

@@ -567,6 +567,57 @@ def test_tasks_endpoint_without_pipeline(self):
         self.assertEqual(resp.status_code, 400)
         self.assertIn("pipeline", resp.json()[0].lower())
 
+    def test_queue_images_to_nats_embeds_pipeline_config(self):
+        """Tasks queued to NATS carry the pipeline config (including project overrides)."""
+        from unittest.mock import AsyncMock, MagicMock, patch
+
+        from ami.ml.models import ProjectPipelineConfig
+        from ami.ml.schemas import PipelineRequestConfigParameters
+
+        pipeline = self._create_pipeline()
+        pipeline.default_config = PipelineRequestConfigParameters({"example_param": "default"})
+        pipeline.save()
+        # _create_pipeline already called pipeline.projects.add(self.project) which
+        # created a ProjectPipelineConfig row; update it rather than creating a duplicate.
+        ProjectPipelineConfig.objects.filter(project=self.project, pipeline=pipeline).update(
+            config={"example_param": "project_override"}
+        )
+
+        job = self._create_ml_job("Config propagation test", pipeline)
+        job.dispatch_mode = JobDispatchMode.ASYNC_API
+        job.status = JobState.STARTED
+        job.save(update_fields=["dispatch_mode", "status"])
+
+        image = SourceImage.objects.create(
+            path="config_test.jpg",
+            public_base_url="http://example.com",
+            project=self.project,
+        )
+
+        published_tasks = []
+
+        mock_manager = AsyncMock()
+        mock_manager.log_async = AsyncMock()
+
+        async def capture_publish(job_id, data):
+            published_tasks.append(data)
+            return True
+
+        mock_manager.publish_task = capture_publish
+
+        mock_ctx = MagicMock()
+        mock_ctx.__aenter__ = AsyncMock(return_value=mock_manager)
+        mock_ctx.__aexit__ = AsyncMock(return_value=False)
+
+        with patch("ami.ml.orchestration.jobs.TaskQueueManager", return_value=mock_ctx):
+            with patch("ami.ml.orchestration.jobs.AsyncJobStateManager"):
+                queue_images_to_nats(job, [image])
+
+        self.assertEqual(len(published_tasks), 1)
+        task = published_tasks[0]
+        self.assertIsNotNone(task.config)
+        self.assertEqual(task.config.get("example_param"), "project_override")
+
     def test_result_endpoint_stub(self):
         """Test the result endpoint accepts results (stubbed implementation)."""
         pipeline = self._create_pipeline()

ami/ml/orchestration/jobs.py

@@ -86,6 +86,8 @@ def queue_images_to_nats(job: "Job", images: list[SourceImage]):
     job.logger.info(f"Queuing {len(images)} images to NATS stream for job '{job.pk}'")
 
     # Prepare all messages outside of async context to avoid Django ORM issues
+    pipeline_config = job.pipeline.get_config(project_id=job.project.pk) if job.pipeline else None
+
     tasks: list[tuple[int, PipelineProcessingTask]] = []
     image_ids = []
     skipped_count = 0
@@ -101,6 +103,7 @@ def queue_images_to_nats(job: "Job", images: list[SourceImage]):
             id=image_id,
             image_id=image_id,
             image_url=image_url,
+            config=pipeline_config,
         )
         tasks.append((image.pk, task))
 

ami/ml/schemas.py

@@ -239,6 +239,7 @@ class PipelineResultsResponse(pydantic.BaseModel):
     source_images: list[SourceImageResponse]
     detections: list[DetectionResponse]
     errors: list | str | None = None
+    config: PipelineRequestConfigParameters | dict | None = None
 
 
 class PipelineResultsError(pydantic.BaseModel):
@@ -257,9 +258,8 @@ class PipelineProcessingTask(pydantic.BaseModel):
     image_id: str
     image_url: str
     reply_subject: str | None = None  # The NATS subject to send the result to
-    # TODO: Do we need these?
+    config: PipelineRequestConfigParameters | dict | None = None
     # detections: list[DetectionRequest] | None = None
-    # config: PipelineRequestConfigParameters | dict | None = None
 
 
 class ProcessingServiceClientInfo(pydantic.BaseModel):

processing_services/README.md

@@ -76,6 +76,28 @@ PipelineChoice = typing.Literal[
     "new-pipeline-slug",
 ]
 ```
+## NATS Pull-Mode (Async API) Contract
+
+Processing services that operate in pull-mode (fetching tasks from Antenna via `POST /api/v2/jobs/{id}/tasks/`) receive `PipelineProcessingTask` objects. Each task now includes a `config` field carrying the pipeline configuration for that job:
+
+```json
+{
+  "id": "42",
+  "image_id": "42",
+  "image_url": "https://...",
+  "reply_subject": "antenna.results.job.7.img.42",
+  "config": {
+    "example_config_param": 3
+  }
+}
+```
+
+`config` mirrors `PipelineRequest.config` from the synchronous HTTP path. It is derived from the pipeline's `default_config` merged with any per-project `ProjectPipelineConfig` override. It may be `null` if no config is set.
+
+Workers should read `config` from each task and apply it to their processing. If `config` is absent or null, fall back to worker-level defaults (e.g. environment variables).
+
+When returning results, populate `PipelineResultsResponse.config` with the config that was actually used, so Antenna can record it for auditing.
+
 ## Demo
 
 ## `minimal` Pipelines and Output Images