Add processing service v2 template to main docker compose stack

[mihow]

Summary

Extends processing_services/minimal/ to support Antenna's v2 pull/async/worker paradigm alongside the existing v1 push mode, so agents and CI can exercise the NATS/Redis/Celery lifecycle without spinning up the external ami-data-companion (ADC) worker.

One container, three modes via MODE env var:

• MODE=api — v1 FastAPI only (unchanged; CI default)
• MODE=worker — v2 HTTP poll loop
• MODE=api+worker — both (local dev default)

Intention & context

Antenna supports two processing-service paradigms:

1. v1 (push) — synchronous HTTP: Antenna POSTs PipelineRequest to /process. Good for interactive single-image inference, /api/v2/docs/ schema exposure, and admin smoke tests.
2. v2 (pull/async/worker) — workers poll POST /api/v2/jobs/{id}/tasks/, process, and POST /api/v2/jobs/{id}/result/. Antenna proxies NATS internally so workers can live behind firewalls.

Today the only v2 implementation is the external ami-data-companion (heavyweight: conda + torch + CUDA warmup). This PR adds a deterministic stub analogous to what minimal/ already is for v1.

Design doc: docs/claude/planning/2026-04-17-minimal-worker-design.md — includes explicit comparison to the ADC worker (what this mirrors vs. diverges) and to PR #1011 (the earlier Celery-direct attempt, and why NATS/API was chosen over it).

What changed

Processing service

• processing_services/minimal/start.sh — MODE-dispatching orchestrator with signal forwarding for TERM/INT. In api+worker mode: FastAPI first (so /process stays available while register.py retries), then register.py, then the worker loop.
• processing_services/minimal/register.py — self-registration. Auth priority: ANTENNA_API_KEY → ANTENNA_API_AUTH_TOKEN → login with ANTENNA_USER/ANTENNA_PASSWORD. Pipelines are imported directly from the api module (no HTTP round-trip to self). Sends both processing_service_name (main's serializer expects it as the label for ProcessingService.get_or_create(name=...)) and client_info (read by PR feat: API key auth and identity for processing services #1194).
• processing_services/minimal/worker_main.py — entry for MODE=worker. Reads the cached auth header that register.py drops at /tmp/antenna_auth_header, falls back to env-based login if absent.
• processing_services/minimal/worker/ — three small modules: client.py (HTTP wrapper with retry), loop.py (one listing call for all pipelines per poll), runner.py (task → result via the existing api/pipelines.py stubs). No duplicated pipeline or schema code.
• processing_services/minimal/worker/schemas.py — v2-only worker classes (PipelineProcessingTask, PipelineTaskResult, PipelineResultsError, ProcessingServiceClientInfo, AsyncPipelineRegistrationRequest), importing the shared wire-format types (PipelineResultsResponse, PipelineConfigResponse, ...) from api/schemas.py. Keeping the pull-mode types out of api/schemas.py means the v1 FastAPI side stays free of v2 types.
• processing_services/minimal/.env.dev — dev defaults for MODE, Antenna target, auth fallback, and worker tuning. Loaded via env_file: in processing_services/docker-compose.yml. No more duplicated defaults in Python code or in environment: blocks.
• processing_services/minimal/Dockerfile — ENV MODE=api (CI default), CMD ["/app/start.sh"].

Out-of-the-box sequencing

• ami/main/management/commands/ensure_default_project.py — idempotent bootstrap. Creates the default superuser (createsuperuser from DJANGO_SUPERUSER_* env vars) and a named project (default "Default Project"). Looks up by name, not PK, so long-lived dev DBs that already have PK 1 claimed are safe.
• compose/local/django/start — runs the command when ENSURE_DEFAULT_PROJECT=1.
• .envs/.local/.django, .envs/.ci/.django — set ENSURE_DEFAULT_PROJECT=1 and (for CI) DJANGO_SUPERUSER_*.

Review-round changes (2026-04-17)

Applied @mihow's review feedback:

• Schema typing. client.py, runner.py, and register.py take/return real Pydantic types rather than dicts. The v2-only worker classes live in worker/schemas.py (importing the shared wire-format types from api/schemas.py), so the v1 FastAPI side stays free of pull-mode types.
• Env file. All dev defaults centralized in processing_services/minimal/.env.dev. os.environ.get(..., "<default>") fallbacks in register.py and worker_main.py replaced with os.environ[...]; inline environment: duplication in docker-compose.yml removed.
• Per-slug polling. loop.py iterates for slug in my_slugs at the outer level so runner.process_task(task, slug) gets the slug as a loop variable. The prior _slug_for_job reverse-lookup is gone.
• register.py docstring. Rewritten to describe current-main identity (PS identified by the Authorization header's user; processing_service_name just labels the DB row via get_or_create(name=...)) and the feat: API key auth and identity for processing services #1194 transition (API-key identifies the PS directly).

The flow on a cold boot

1. Postgres/RabbitMQ/Redis/NATS come up.
2. Django runs migrations, then ensure_default_project creates/finds a superuser + "Default Project".
3. Minimal container starts. start.sh runs FastAPI in the background.
4. register.py imports its pipelines from api.api, logs in to Antenna (or uses ANTENNA_API_AUTH_TOKEN), resolves the project by name, and POSTs pipelines to /api/v2/projects/{id}/pipelines/.
5. worker_main.py starts polling GET /api/v2/jobs/?pipeline=<slug>&status=STARTED&ids_only=true once per slug. When a STARTED async_api job matches, it reserves a batch via POST /jobs/{id}/tasks/, runs the stub pipeline, and POSTs PipelineTaskResult items back.

Errors during stub execution are captured and sent as PipelineResultsError so the NATS ACK path still fires — important for exercising retry / stale-job-cutoff paths.

What's verified vs. what still needs verification

• docker compose -f processing_services/docker-compose.yml build ml_backend_minimal — builds clean
• MODE=api container — backward compatible with existing CI (same /process, /info, /livez, /readyz)
• MODE=api+worker container — boots, runs register.py, retries login when Antenna unreachable
• python -m py_compile — all new Python compiles
• Pre-commit (black, isort, flake8) passes
• End-to-end on a live stack — now automated: the test-async-processing-service CI job runs an async_api job through the worker (see the 2026-10-13 update below).
• ensure_default_project idempotence across repeated Django boots
• CI still green — docker-compose.ci.yml keeps ml_backend in MODE=api; the worker runs as a separate ml_backend_worker service behind the async profile
• Confirm PR interop with #1194 when it lands (one-line removal of processing_service_name)

Test plan (for reviewers)

# Bring up Antenna core
docker compose up -d

# Wait for django to apply migrations + run ensure_default_project
docker compose logs django | grep -E 'ensure_default_project|Default Project'

# Bring up the minimal container in api+worker mode
docker compose -f processing_services/docker-compose.yml up -d ml_backend_minimal
docker compose -f processing_services/docker-compose.yml logs -f ml_backend_minimal

# Expected: FastAPI up on :2000, register.py says "Registered N pipelines",
# worker loop says "Polling for jobs on pipelines: ['constant', 'random-detection-random-species']"

# Submit a sync job (v1 path still works)
curl -X POST http://localhost:8000/api/v2/jobs/ ...

# Submit an async_api job (v2 path, exercises the new worker)
curl -X POST http://localhost:8000/api/v2/jobs/?start_now=true ...

Out of scope (called out in design doc)

• Real ML in the stub — stays random/constant.
• Renaming example/ to global_moths / complete — deferred follow-up.
• API-key-only auth — depends on feat: API key auth and identity for processing services #1194.
• Competing-consumer / multi-worker tests.
• Direct-NATS workers.

Related

• #987 — job queue HTTP API (merged)
• #1011 — earlier Celery-direct PSv2 attempt (never merged); design doc explains why NATS/API wins
• #1194 — API key auth & identity (open; this PR is forward-compatible)
• #952, #992 — example/ service evolution
• ami-data-companion#94 — ADC worker (merged)
• ami-data-companion#136, #141 — ADC evolution

What has been tested

Unit (ami/main/tests.py::TestEnsureDefaultProjectSeed, 3 passed): the seed creates a non-empty Default Collection, is idempotent (re-running does not re-seed — keyed on the collection), and --skip-seed creates no images. Worker modules compile; lint green.

Local end-to-end (deployed on this branch, MODE=api+worker, no ami-data-companion): the worker registered both stub pipelines to the seeded Default Project and polled. One async_api job (constant, 6 images) ran the full pull lifecycle — STARTED → NATS task claim (POST /jobs/{id}/tasks/) → stub inference → POST /jobs/{id}/result/ ×6 → process_nats_pipeline_result ×6 → NATS ACK + consumer/stream teardown → SUCCESS in ~4s, producing 6 detections and 6 occurrences with progress.errors == []. No code changes were needed.

Auth used in this run: the worker authenticates via the Authorization header (DRF). register.py resolves it in priority order — ANTENNA_API_KEY (Authorization: Api-Key …, the #1194 path) → ANTENNA_API_AUTH_TOKEN (Authorization: Token …) → fallback login with ANTENNA_USER/ANTENNA_PASSWORD at /api/v2/auth/token/login/ to obtain a DRF token. With no key/token set, this run used the user/password → Token fallback; the resolved header is cached and reused by the poll loop.

Update 2026-10-13: brought up to main's job API, tested in CI

The worker now follows the job API on main exactly as the AMI Data Companion's worker does, and CI runs an async job through it, so pull mode is tested on every backend change instead of only on a developer's machine. Plan for the related template PRs (#992, #952) and the auth options with #1194: docs/claude/planning/2026-10-13-processing-service-templates-plan.md.

List of Changes

Change (effect)	How	Notes
A polling worker stays "online" in Antenna and never asks for tasks from jobs it cannot serve	One GET /jobs/ call for all pipelines with pipeline__slug__in, ids_only, incomplete_only and dispatch_mode=async_api, using each job's own pipeline_slug	The per-slug pipeline__slug + status=STARTED query never triggered the idle heartbeat (the service read as offline after a minute), missed RETRY jobs, and listed synchronous jobs whose /tasks/ call returns 400
Pull mode is tested on every backend change	New CI job test-async-processing-service runs scripts/ci/async_processing_service_e2e.sh: Antenna starts, the worker (ml_backend_worker, behind the async compose profile) registers and polls, an async_api job runs on constant, and every image must get a detection and a classification	test_ml_job_e2e gains --timeout and exits non-zero when the job does not succeed; the existing test job is unchanged
The template's own tests run, and in CI	The broken api/test.py (it imported a pipeline that no longer exists) is replaced with contract tests; new worker/test_worker.py against a fake Antenna client; a workflow step runs both
The templates cannot silently drift from Antenna's contract	ami/ml/test_processing_service_schemas.py reads the template schema files as source: models a service sends may use only fields Antenna reads, and models Antenna sends must keep every field	Also covers example/
minimal's responses match main's schema	Optional bbox, errors on results, logits defaults to None; the dead random pipeline choice is dropped
Service authors know what a pull-mode worker must do	README section; design-doc drift fixed (listing query, project looked up by name)
Branch is current	main merged in; the two append-only conflicts resolved

Testing (2026-10-13)

• Minimal service and worker unit tests: 10 pass.
• Async end to end on a fresh local CI stack with scripts/ci/async_processing_service_e2e.sh: the worker registered and polled, the async_api job on constant reached SUCCESS, and all 6 seeded images got a detection and a classification.
• Schema drift guard: 4 tests pass; no drift between the templates and ami/ml/schemas.py today.
• Full backend suite: 864 tests pass (2 skipped); makemigrations --check reports no changes; pre-commit passes on every changed file.

🤖 Generated with Claude Code

https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C

[netlify]

✅ Deploy Preview for antenna-preview canceled.

Name	Link
🔨 Latest commit	06799de
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-preview/deploys/6aceb6a0955ab3ca0c4d02e6

[netlify]

✅ Deploy Preview for antenna-ssec canceled.

Name	Link
🔨 Latest commit	06799de
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-ssec/deploys/6aceb6a06e53b6d869405a80

[coderabbitai]

Important

Draft PR not reviewed

Draft PRs are not automatically reviewed by default.

• Trigger a manual review

To automatically review draft PRs, update your CodeRabbit configuration:

reviews:
  auto_review:
    drafts: true

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch worktree-minimal-worker-default-processing-service

• Autofix · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[mihow]

See my inline comments. Make it clear in the PR description and/or design doc what mirrors or diverges from the implementation in the AMI Data Companion worker. Also review this previous implementation for adding a worker to the docker compose stack, but for connecting to Celery directly. We will stick with the NATS / API version, but I'm curious for you to read the details of this implementation and write up a comparison. #1011

[mihow]

Claude says: Follow-up in 58b247a9 per your note that the schemas concern was about using Pydantic at all, not about where the v2 classes live:

• Re-split schemas. v2-only classes moved back out of api/schemas.py into worker/schemas.py (PipelineProcessingTask, PipelineTaskResult, PipelineResultsError, ProcessingServiceClientInfo, AsyncPipelineRegistrationRequest, TasksResponse). The v1 push side no longer imports any pull-mode types. Shared wire-format classes (PipelineResultsResponse, PipelineConfigResponse, SourceImageResponse, ...) stay in api/schemas.py and are imported by worker/schemas.py. Typed signatures throughout client.py / runner.py / loop.py / register.py / worker_main.py are preserved.

• Dropped my_slugs. It was a redundant local for list(pipeline_choices.keys()). Loop now iterates pipeline_choices directly — for slug in pipeline_choices — so there's no intermediate and no awkward possessive name.

Live-stack e2e is up next.

[mihow]

Claude says: Rebased onto current main (was 56 behind; clean except one ami/main/tests.py conflict from #1312 having merged — resolved by keeping both), added a zero-config seed, and ran a full local async e2e through the minimal worker.

Zero-config seed. ensure_default_project now seeds a small object-store-backed Default Collection into the default project (the worker opens each image and reads pixel dimensions, so it needs reachable images, not path-only rows). Idempotent — keyed on the collection, not on "any source image", since a project can have images from other sources yet still lack the collection the worker needs. Best-effort (swallows + logs a failure such as no object store so it never breaks startup), opt-out via --skip-seed. Tests added.

E2E result (local, MODE=api+worker, no ADC). Worker registered both stub pipelines to the seeded project and polled. One async_api job (constant, 6 images):

• STARTED → worker reserved tasks via NATS → ran the stub → POST /jobs/{id}/result/ ×6 → process_nats_pipeline_result ×6 → NATS ACK → consumer/stream auto-teardown → SUCCESS in ~4s.
• 6 detections + 6 occurrences created; progress.errors == []; all three stages at 100%.
• No code changes were needed — the worker ran as-is. Networking worked out of the box via the external antenna_network.

So the full v2 pull lifecycle is exercised end-to-end without the heavyweight ADC worker. With the seed, a fresh stack boots straight into a processable state. This is ready to come out of draft for review whenever you like; the [ ] e2e item in the description can be ticked.

[mihow]

Claude says: Review for merging this as the foundational step — a correct, copy-able reference for the v2 (pull/async) API contract via a worker in the app stack. (Deeper lifecycle hardening — failure injection, concurrent-load + chaos — is intentionally out of scope here and tracked separately.)

Should-fix before merge (cheap; they protect the contract this PR exists to demonstrate):

1. Heartbeat filter is out of step with main. The poll lists jobs with pipeline__slug=<slug> (singular exact). Main's "service-live" heartbeat now fires off pipeline__slug__in (plural) plus a Celery task, so the worker never updates ProcessingService.last_seen between registrations — it reads as offline in the admin even while it's processing jobs. Since this file is meant to be the reference for talking to the v2 API, switching the poll to pipeline__slug__in keeps it in step (the filter already exists on JobFilterSet). Doesn't block processing; it's contract drift in a file people will copy.

2. Schema parity has no guard. worker/schemas.py mirrors ami/ml/schemas.py field-for-field, protected only by a "keep these in sync" docstring. For a contract reference, silent drift makes it quietly wrong. A lightweight parity test (import both, assert field names/types match, fail CI on drift) would protect exactly the thing this PR demonstrates. If you'd rather defer the test, point the docstring at the canonical module + line range so it can be diffed by hand.

Notes / small doc adds (prevent misuse of the reference):

3. Auth caveat. Worth stating in .env.dev / register.py that the user-password→Token and static-token paths are for local/CI; a real processing service should use the API key (feat: API key auth and identity for processing services #1194) so there's no stored password — otherwise someone copies password auth into a deployment.

4. MODE=api+worker liveness coupling. start.sh's wait -n takes the whole container down if either child dies. A one-line note that this is intentional (harness simplicity) keeps a reader from reading it as a bug.

5. No re-auth on 401 (minor). The poll loop catches errors and continues (good) and there is a urllib3 retry, but a rotated/revoked token would 401-loop until restart. DRF authtoken does not expire by default so it is low-risk — worth a one-line note that token refresh is a feat: API key auth and identity for processing services #1194-era concern.

Already addressed: rebased onto current main (the ami/main/tests.py conflict from #1312 having merged is resolved); the stale "worker/schemas.py deleted" body claim + register.py's .env.example path corrected; zero-config seed added so a fresh stack boots into a processable state; happy-path async e2e ran clean end-to-end.

Net: mergeable as the foundation. The only two I would gate on are #1 and #2 — both cheap, both protect the contract-reference value.