AMI: Pipeline Registration (#106)

* Pipeline registration

* Convert worker tests to integration tests with real ML inference

Replaces fully mocked unit tests with integration tests that validate
the Antenna API contract and run actual ML models. Tests now exercise
the worker's unique code path (RESTDataset → rest_collate_fn) with real
image loading and inference.

Changes:
- Add trapdata/api/tests/utils.py with shared test utilities
- Add trapdata/api/tests/antenna_api_server.py to mock Antenna API
- Rewrite test_worker.py as integration tests (17 tests, all passing)
- Update test_api.py to use shared utilities

Tests validate: real detector/classifier inference, HTTP image loading,
schema compliance, batch processing, and end-to-end workflow.

Co-Authored-By: Claude Sonnet 4.5 <[email protected]>

* Add AsyncPipelineRegistrationResponse schema

Add Pydantic model to validate responses from pipeline registration API.
Fields: pipelines_created, pipelines_updated, processing_service_id.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* Refactor registration functions to use get_http_session

Update get_user_projects() and register_pipelines_for_project() to use
the session-based HTTP pattern established in PR #104:
- Use get_http_session() context manager for connection pooling
- Add retry_max and retry_backoff parameters with defaults
- Remove manual header management (session handles auth)
- Standardize URL paths (base_url now includes /api/v2)
- Use Pydantic model validation for API responses
- Fix error handling with hasattr() check

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* Add integration tests for pipeline registration

Add mock Antenna API endpoints:
- GET /api/v2/projects/ - list user's projects
- POST /api/v2/projects/{id}/pipelines/ - register pipelines

Add TestRegistrationIntegration with 2 client tests:
- test_get_user_projects
- test_register_pipelines_for_project

Update TestWorkerEndToEnd.test_full_workflow_with_real_inference to include
registration step: register → get jobs → process → post results.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* Add git add -p to recommended development practices

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* Read retry settings from Settings in get_http_session()

When max_retries or backoff_factor are not explicitly provided,
get_http_session() now reads defaults from Settings (antenna_api_retry_max
and antenna_api_retry_backoff). This centralizes retry configuration and
allows callers to omit these low-level parameters.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* Use Settings pattern in register_pipelines()

- Accept Settings object instead of base_url/auth_token params
- Remove direct os.environ.get() calls for ANTENNA_API_* vars
- Fix error message to reference correct env var (AMI_ANTENNA_API_AUTH_TOKEN)
- Remove retry params from get_user_projects() and register_pipelines_for_project()
  since get_http_session() now reads settings internally
- Remove unused os import

Co-Authored-By: Claude Opus 4.5 <[email protected]>

---------

Co-authored-by: Carlos Garcia Jurado Suarez <[email protected]>
Co-authored-by: Carlos Garcia Jurado Suarez <[email protected]>
Co-authored-by: Claude Sonnet 4.5 <[email protected]>

Author: 4 people

CLAUDE.md

@@ -13,8 +13,9 @@ This file helps AI agents (like Claude) work efficiently with the AMI Data Compa
 3. **Always prefer command line tools** to avoid expensive API requests (e.g., use git and jq instead of reading whole files)
 4. **Use bulk operations and prefetch patterns** to minimize database queries
 5. **Commit often** - Small, focused commits make debugging easier
-6. **Use TDD whenever possible** - Tests prevent regressions and document expected behavior
-7. **Keep it simple** - Always think hard and evaluate more complex approaches and alternative approaches before moving forward
+6. **Use `git add -p` for staging** - Interactive staging to add only relevant changes, creating logical commits
+7. **Use TDD whenever possible** - Tests prevent regressions and document expected behavior
+8. **Keep it simple** - Always think hard and evaluate more complex approaches and alternative approaches before moving forward
 
 ### Think Holistically
 

trapdata/api/schemas.py

@@ -382,3 +382,31 @@ class ProcessingServiceInfoResponse(pydantic.BaseModel):
             ]
         ],
     )
+
+
+class AsyncPipelineRegistrationRequest(pydantic.BaseModel):
+    """
+    Request to register pipelines from an async processing service
+    """
+
+    processing_service_name: str
+    pipelines: list[PipelineConfigResponse] = []
+
+
+class AsyncPipelineRegistrationResponse(pydantic.BaseModel):
+    """
+    Response from registering pipelines with a project.
+    """
+
+    pipelines_created: list[str] = pydantic.Field(
+        default_factory=list,
+        description="List of pipeline slugs that were created",
+    )
+    pipelines_updated: list[str] = pydantic.Field(
+        default_factory=list,
+        description="List of pipeline slugs that were updated",
+    )
+    processing_service_id: int | None = pydantic.Field(
+        default=None,
+        description="ID of the processing service that was created or updated",
+    )

trapdata/api/tests/antenna_api_server.py

@@ -14,13 +14,17 @@
     AntennaTaskResult,
     AntennaTaskResults,
     AntennaTasksListResponse,
+    AsyncPipelineRegistrationRequest,
+    AsyncPipelineRegistrationResponse,
 )
 
 app = FastAPI()
 
 # State management for tests
 _jobs_queue: dict[int, list[AntennaPipelineProcessingTask]] = {}
 _posted_results: dict[int, list[AntennaTaskResult]] = {}
+_projects: list[dict] = []
+_registered_pipelines: dict[int, list[str]] = {}  # project_id -> pipeline slugs
 
 
 @app.get("/api/v2/jobs")
@@ -84,6 +88,52 @@ def post_results(job_id: int, payload: list[dict]):
     return {"status": "ok"}
 
 
+@app.get("/api/v2/projects/")
+def get_projects():
+    """Return list of projects the user has access to.
+
+    Returns:
+        Paginated response with list of projects
+    """
+    return {"results": _projects}
+
+
+@app.post("/api/v2/projects/{project_id}/pipelines/")
+def register_pipelines(project_id: int, payload: dict):
+    """Register pipelines for a project.
+
+    Args:
+        project_id: Project ID to register pipelines for
+        payload: AsyncPipelineRegistrationRequest as dict
+
+    Returns:
+        AsyncPipelineRegistrationResponse
+    """
+    # Validate request
+    request = AsyncPipelineRegistrationRequest(**payload)
+
+    # Check if project exists
+    project_ids = [p["id"] for p in _projects]
+    if project_id not in project_ids:
+        raise HTTPException(status_code=404, detail="Project not found")
+
+    # Track registered pipelines
+    if project_id not in _registered_pipelines:
+        _registered_pipelines[project_id] = []
+
+    created = []
+    for pipeline in request.pipelines:
+        if pipeline.slug not in _registered_pipelines[project_id]:
+            _registered_pipelines[project_id].append(pipeline.slug)
+            created.append(pipeline.slug)
+
+    return AsyncPipelineRegistrationResponse(
+        pipelines_created=created,
+        pipelines_updated=[],
+        processing_service_id=1,
+    )
+
+
 # Test helper methods
 
 
@@ -109,7 +159,31 @@ def get_posted_results(job_id: int) -> list[AntennaTaskResult]:
     return _posted_results.get(job_id, [])
 
 
+def setup_projects(projects: list[dict]):
+    """Setup projects for testing.
+
+    Args:
+        projects: List of project dicts with 'id' and 'name' fields
+    """
+    _projects.clear()
+    _projects.extend(projects)
+
+
+def get_registered_pipelines(project_id: int) -> list[str]:
+    """Get list of pipeline slugs registered for a project.
+
+    Args:
+        project_id: Project ID to get pipelines for
+
+    Returns:
+        List of pipeline slugs
+    """
+    return _registered_pipelines.get(project_id, [])
+
+
 def reset():
     """Clear all state between tests."""
     _jobs_queue.clear()
     _posted_results.clear()
+    _projects.clear()
+    _registered_pipelines.clear()

trapdata/api/tests/test_worker.py

@@ -17,13 +17,19 @@
     AntennaPipelineProcessingTask,
     AntennaTaskResult,
     AntennaTaskResultError,
+    PipelineConfigResponse,
     PipelineResultsResponse,
 )
 from trapdata.api.tests import antenna_api_server
 from trapdata.api.tests.antenna_api_server import app as antenna_app
 from trapdata.api.tests.image_server import StaticFileTestServer
 from trapdata.api.tests.utils import get_test_image_urls, patch_antenna_api_requests
-from trapdata.cli.worker import _get_jobs, _process_job
+from trapdata.cli.worker import (
+    _get_jobs,
+    _process_job,
+    get_user_projects,
+    register_pipelines_for_project,
+)
 from trapdata.tests import TEST_IMAGES_BASE_PATH
 
 # ---------------------------------------------------------------------------
@@ -473,10 +479,13 @@ def _make_settings(self):
 
     def test_full_workflow_with_real_inference(self):
         """
-        Complete workflow: fetch jobs → fetch tasks → load images →
+        Complete workflow: register → fetch jobs → fetch tasks → load images →
         run detection → run classification → post results.
         """
-        # Setup job with 2 test images
+        pipeline_slug = "quebec_vermont_moths_2023"
+
+        # Setup project and job with 2 test images
+        antenna_api_server.setup_projects([{"id": 1, "name": "Test Project"}])
         image_urls = get_test_image_urls(
             self.file_server, self.test_images_dir, subdir="vermont", num=2
         )
@@ -491,25 +500,33 @@ def test_full_workflow_with_real_inference(self):
         ]
         antenna_api_server.setup_job(job_id=200, tasks=tasks)
 
-        # Step 1: Get jobs
         with patch_antenna_api_requests(self.antenna_client):
+            # Step 1: Register pipeline
+            pipeline_configs = [
+                PipelineConfigResponse(name="Vermont Moths", slug=pipeline_slug, version=1)
+            ]
+            success, _ = register_pipelines_for_project(
+                base_url="http://testserver/api/v2",
+                auth_token="test-token",
+                project_id=1,
+                service_name="Test Worker",
+                pipeline_configs=pipeline_configs,
+            )
+            assert success is True
+
+            # Step 2: Get jobs
             job_ids = _get_jobs(
                 "http://testserver/api/v2",
                 "test-token",
-                "quebec_vermont_moths_2023",
-            )
-
-        assert 200 in job_ids
-
-        # Step 2: Process job
-        with patch_antenna_api_requests(self.antenna_client):
-            result = _process_job(
-                "quebec_vermont_moths_2023", 200, self._make_settings()
+                pipeline_slug,
             )
+            assert 200 in job_ids
 
-        assert result is True
+            # Step 3: Process job
+            result = _process_job(pipeline_slug, 200, self._make_settings())
+            assert result is True
 
-        # Step 3: Validate results posted
+        # Step 4: Validate results posted
         posted_results = antenna_api_server.get_posted_results(200)
         assert len(posted_results) == 2
 
@@ -566,3 +583,52 @@ def test_multiple_batches_processed(self):
         assert all(
             isinstance(r.result, PipelineResultsResponse) for r in posted_results
         )
+
+
+# ---------------------------------------------------------------------------
+# TestRegistrationIntegration - Basic tests for registration client functions
+# ---------------------------------------------------------------------------
+
+
+class TestRegistrationIntegration(TestCase):
+    """Integration tests for registration client functions."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.antenna_client = TestClient(antenna_app)
+
+    def setUp(self):
+        antenna_api_server.reset()
+
+    def test_get_user_projects(self):
+        """Client can fetch list of projects."""
+        antenna_api_server.setup_projects([
+            {"id": 1, "name": "Project A"},
+            {"id": 2, "name": "Project B"},
+        ])
+
+        with patch_antenna_api_requests(self.antenna_client):
+            result = get_user_projects("http://testserver/api/v2", "test-token")
+
+        assert len(result) == 2
+        assert result[0]["id"] == 1
+
+    def test_register_pipelines_for_project(self):
+        """Client can register pipelines for a project."""
+        antenna_api_server.setup_projects([{"id": 10, "name": "Test Project"}])
+
+        pipeline_configs = [
+            PipelineConfigResponse(name="Test Pipeline", slug="test_pipeline", version=1)
+        ]
+
+        with patch_antenna_api_requests(self.antenna_client):
+            success, message = register_pipelines_for_project(
+                base_url="http://testserver/api/v2",
+                auth_token="test-token",
+                project_id=10,
+                service_name="Test Service",
+                pipeline_configs=pipeline_configs,
+            )
+
+        assert success is True
+        assert "Created" in message

trapdata/api/utils.py

@@ -40,8 +40,8 @@ def get_crop_fname(source_image: SourceImage, bbox: BoundingBox) -> str:
 
 def get_http_session(
     auth_token: str | None = None,
-    max_retries: int = 3,
-    backoff_factor: float = 0.5,
+    max_retries: int | None = None,
+    backoff_factor: float | None = None,
     status_forcelist: tuple[int, ...] = (500, 502, 503, 504),
 ) -> requests.Session:
     """
@@ -53,8 +53,8 @@ def get_http_session(
 
     Args:
         auth_token: Optional authentication token (adds "Token {token}" to Authorization header)
-        max_retries: Maximum number of retry attempts (default: 3)
-        backoff_factor: Exponential backoff multiplier in seconds (default: 0.5)
+        max_retries: Maximum number of retry attempts (default: from settings.antenna_api_retry_max)
+        backoff_factor: Exponential backoff multiplier in seconds (default: from settings.antenna_api_retry_backoff)
                        Delays will be: backoff_factor * (2 ** retry_number)
                        e.g., 0.5s, 1s, 2s for default settings
         status_forcelist: HTTP status codes that trigger a retry (default: 500, 502, 503, 504)
@@ -69,6 +69,16 @@ def get_http_session(
         >>> session = get_http_session(auth_token="abc123")
         >>> response = session.get("https://api.example.com/data")
     """
+    # Read defaults from settings if not explicitly provided
+    if max_retries is None or backoff_factor is None:
+        from trapdata.settings import read_settings
+
+        settings = read_settings()
+        if max_retries is None:
+            max_retries = settings.antenna_api_retry_max
+        if backoff_factor is None:
+            backoff_factor = settings.antenna_api_retry_backoff
+
     session = requests.Session()
 
     retry_strategy = Retry(

trapdata/cli/base.py

@@ -128,5 +128,38 @@ def worker(
     run_worker(pipelines=pipelines)
 
 
+@cli.command("register")
+def register(
+    name: Annotated[
+        str,
+        typer.Argument(
+            help="Name for the processing service registration (e.g., 'AMI Data Companion on DRAC gpu-03'). "
+            "Hostname will be added automatically.",
+        ),
+    ],
+    project: Annotated[
+        list[int] | None,
+        typer.Option(
+            help="Specific project IDs to register pipelines for. "
+            "If not specified, registers for all accessible projects.",
+        ),
+    ] = None,
+):
+    """
+    Register available pipelines with the Antenna platform for specified projects.
+
+    This command registers all available pipeline configurations with the Antenna platform
+    for the specified projects (or all accessible projects if none specified).
+
+    Examples:
+        ami register --name "AMI Data Companion on DRAC gpu-03" --project 1 --project 2
+        ami register --name "My Processing Service"  # registers for all accessible projects
+    """
+    from trapdata.cli.worker import register_pipelines
+
+    project_ids = project if project else []
+    register_pipelines(project_ids=project_ids, service_name=name)
+
+
 if __name__ == "__main__":
     cli()