Skip to content

Run the template processing service as a pull-mode worker, with an async job in CI - #1512

Draft
mihow wants to merge 18 commits into
feat/processing-service-parityfrom
feat/processing-service-async-worker
Draft

mihow wants to merge 18 commits into
feat/processing-service-parityfrom
feat/processing-service-async-worker

Conversation

@mihow

@mihow mihow commented Oct 13, 2026

Copy link
Copy Markdown
Collaborator

Summary

Pull (async) mode is now the default way processing services work with Antenna, but the template service in this repository only worked in push mode, and nothing in CI exercised pull mode end to end. This PR makes the minimal template work as a pull-mode worker against the current jobs API and vector contract, and adds a CI job that runs a real async job through it and checks that the results, including feature vectors, are stored. It builds on the draft worker in #1252, which is merged in rather than rewritten, and on the parity work in the base PR.

List of Changes

Change (effect) How Notes
The template can run as a pull-mode worker (MODE=worker or api+worker) that registers its pipelines and processes async jobs From #1252: start.sh, register.py, worker/ and ensure_default_project, which creates the default project, user and a small image collection at startup when ENSURE_DEFAULT_PROJECT=1 Merged from the draft, credit to its author
The worker finds its jobs the same way the AMI Data Companion worker does One GET /jobs/ call with pipeline__slug__in, ids_only, incomplete_only and dispatch_mode=async_api, using the pipeline slug each job carries Antenna records this call as the worker's heartbeat, and synchronous jobs are no longer listed only to fail at /tasks/
Async results carry feature vectors The worker runs the same pipelines as push mode; its config comes from AMI_INCLUDE_LOGITS, AMI_FEATURES_FOR_ALL_DETECTIONS and AMI_EMBEDDING_EXTRACTOR, because pull-mode tasks carry no config yet (#1422)
Feature-only pipelines are not offered in pull mode Tasks carry an image and never detections, as in the AMI Data Companion
Misconfiguration fails at startup, not on every task The worker stops when AMI_PIPELINES leaves no pull-mode pipeline, or when AMI_EMBEDDING_EXTRACTOR names an extractor no pipeline offers
CI runs an async job end to end New workflow job test-async-processing-service runs scripts/ci/async_processing_service_e2e.sh against a worker service (ml_backend_worker, behind the async compose profile); test_ml_job_e2e gains --timeout and exits non-zero when the job does not succeed The existing test job is unchanged
The worker's tests run in CI worker/test_worker.py against a fake Antenna client

Testing

  • Minimal service and worker unit tests: 24 pass.
  • The async e2e script on a fresh local CI stack: job SUCCESS, 6 vectors of length 8 stored for 6 detections.
  • Full backend suite: 938 tests pass (2 skipped); makemigrations --check reports no changes.

Notes for review

Stacked on #1511. Related: #1252, #1046, #1422, #1279, #1194.

🤖 Generated with Claude Code

https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C

mihow and others added 18 commits June 22, 2026 21:45
Adds a design document for extending processing_services/minimal/ with a
v2 worker mode so agents and CI can exercise the pull path without spinning
up the heavyweight ami-data-companion worker.

Co-Authored-By: Claude <noreply@anthropic.com>
Extends processing_services/minimal/ to support Antenna's v2 pull/async/worker
processing-service paradigm alongside the existing v1 push mode. One container
image, three modes via the MODE env var:

  MODE=api          v1 FastAPI only (unchanged, CI default)
  MODE=worker       v2 HTTP poll loop (reserve tasks -> run stub pipeline -> post results)
  MODE=api+worker   both (local dev default)

The worker talks HTTP-only to Antenna's job queue API (POST /jobs/{id}/tasks/,
POST /jobs/{id}/result/), matching the ADC contract. Pipelines are the same
stubs used by /process (ConstantPipeline, RandomDetectionRandomSpeciesPipeline),
so v1 and v2 produce identical detections.

Out-of-the-box automation:
- ami/main/management/commands/ensure_default_project.py: idempotent bootstrap
  that creates the default superuser + "Default Project" on Django startup.
  Guarded by ENSURE_DEFAULT_PROJECT=1, set in .envs/.local/.django and
  .envs/.ci/.django.
- compose/local/django/start runs it before starting gunicorn.
- processing_services/minimal/register.py self-provisions a ProcessingService
  and registers the stub pipelines with the default project. Supports user
  token auth (current main) and has TODOs for API-key auth (PR #1194).
- processing_services/docker-compose.yml sets MODE=api+worker by default.

This unblocks agent-driven e2e work on the async_api path without needing to
stand up the external ami-data-companion worker.

Design doc: docs/claude/planning/2026-04-17-minimal-worker-design.md

Co-Authored-By: Claude <noreply@anthropic.com>
…, simpler loop

Apply feedback from PR #1252 review:

- Consolidate schemas into `processing_services/minimal/api/schemas.py`. v1
  push and v2 worker now share a single source of truth; the separate
  `worker/schemas.py` mirror is removed. Client/runner/register/loop use real
  Pydantic types in signatures (PipelineProcessingTask, PipelineTaskResult,
  ProcessingServiceClientInfo, AsyncPipelineRegistrationRequest).

- Move defaults into `processing_services/minimal/.env.dev`. Strip the
  hard-coded `os.environ.get(..., "<default>")` fallbacks in register.py and
  worker_main.py; replace with `os.environ[...]`. The inline environment
  block in docker-compose.yml becomes `env_file:`.

- Replace the loop's job→slug reverse-lookup with per-slug iteration: the
  outer loop variable IS the slug, so `_slug_for_job` is gone.

- Fix register.py docstring: the PS is identified by the Authorization
  header's user (main) or the API key (#1194), not by
  `processing_service_name`. That field just labels the DB row
  `get_or_create(name=...)` lands on.

- Import pipelines directly in register.py instead of HTTP-GETting
  /info from the co-located FastAPI. Removes the FastAPI readiness wait
  entirely, and means register.py works in MODE=worker (where FastAPI isn't
  running at all).

Co-Authored-By: Claude <noreply@anthropic.com>
- Add "Comparison with prior/external implementations" section covering
  what the stub mirrors vs. diverges from the ADC worker, and why the
  NATS/API path was picked over the Celery-direct path in PR #1011.
- Update the env-var table to reflect the `.env.dev` file as the single
  source of defaults (no code fallbacks).
- Update directory layout and poll-loop pseudocode to match the current
  per-slug iteration (no job→slug reverse-lookup).

Co-Authored-By: Claude <noreply@anthropic.com>
…name my_slugs

Follow-up to the previous review round: v2-only classes go in
worker/schemas.py so the v1 push path doesn't have to know about pull-mode
types. The v1 schemas and v1/v2 shared types (PipelineResultsResponse,
PipelineConfigResponse, SourceImageResponse, ...) stay in api/schemas.py.

Also rename the awkward `my_slugs` in loop.py. The list comes directly from
`pipeline_choices` (the dict of stub pipelines exported by api.api), so
iterate it directly instead of copying keys into a local variable.

Co-Authored-By: Claude <noreply@anthropic.com>
The /jobs/ endpoint's `pipeline` filter is a ModelChoiceFilter expecting
a DB id, so slug values get rejected with 400 "Select a valid choice".
The slug-based alias exposed by JobFilterSet is `pipeline__slug`.

Caught by running the minimal worker end-to-end against a live Antenna
stack — the poll loop was logging 400 warnings every WORKER_POLL_INTERVAL.
After the fix, two async_api test jobs (pipelines `constant` and
`random-detection-random-species`) flow through worker → reserve tasks →
process → submit results → NATS ACK → Job.SUCCESS.

Co-Authored-By: Claude <noreply@anthropic.com>
…ng [skip ci]

The docstring referenced processing_services/.env.example, which does not exist; the real
file is processing_services/minimal/.env.dev.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…efault_project

Make the local/CI bootstrap genuinely zero-config for the minimal v2 worker. The worker opens
each image and reads its pixel dimensions, so it needs reachable images, not path-only rows.
ensure_default_project now seeds a small object-store-backed "Default Collection" into the
default project by reusing the test fixtures (generate images -> object store -> sync captures).
Idempotent on the collection (skips when a non-empty "Default Collection" already exists — keyed
on the collection, not on any source images, since a project may have images from other sources
yet still lack the collection the worker needs). Best-effort (a failure such as no object store
is logged and swallowed so it never breaks startup), and opt-out via --skip-seed. Adds tests for
the seed, idempotency, and the skip flag.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
…he current jobs API and vector contract

- List unfinished async jobs for all pipelines in one call, with the same query as the
  AMI Data Companion worker (pipeline__slug__in, ids_only, incomplete_only,
  dispatch_mode=async_api), and use the pipeline slug each listed job carries. Antenna
  records this call as the worker's heartbeat, and synchronous jobs are no longer
  listed only to fail when their tasks are requested.
- Do not register or poll for feature-only pipelines: pull-mode tasks carry an image and
  never detections, as in the AMI Data Companion.
- Read the request config for pull-mode tasks from AMI_INCLUDE_LOGITS,
  AMI_FEATURES_FOR_ALL_DETECTIONS and AMI_EMBEDDING_EXTRACTOR, because those tasks carry
  no config yet (#1422). Results now include feature vectors.
- Add worker tests against a fake Antenna client.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
…for the worker [skip ci]

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
… failing every task

- Stop at startup when AMI_PIPELINES leaves no pipeline that can run in pull mode;
  otherwise the worker polled with an empty pipeline filter, which matches every job.
- Apply AMI_EMBEDDING_EXTRACTOR only to pipelines that offer an extractor, and stop at
  startup when no pipeline offers the key, instead of returning an error for every task.
- Document that an empty AMI_EMBEDDING_EXTRACTOR turns extractors off.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
Nothing exercised pull mode end to end in CI. A new workflow job starts Antenna (which
creates the default project and image collection at startup), starts the minimal
service as a worker (ml_backend_worker, behind the "async" compose profile, so the
existing test job is unchanged), runs an async_api job with
constant-with-feature-extractor, and checks that every detection got a vector of the
expected length.

test_ml_job_e2e gains --timeout and now exits non-zero when the job does not succeed,
so CI can fail on it.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
… can differ

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TNNRg9NDr6opFZgK3nxx5C
@netlify

netlify Bot commented Oct 13, 2026 •

Copy link
Copy Markdown

✅ Deploy Preview for antenna-preview ready!

Name Link
🔨 Latest commit 120d023
🔍 Latest deploy log https://app.netlify.com/projects/antenna-preview/deploys/6acea07b17aebf6c60f2f877
😎 Deploy Preview https://deploy-preview-1512--antenna-preview.netlify.app
📱 Preview on mobile
Toggle QR Code...

QR Code

Use your smartphone camera to open QR code link.
Lighthouse
Lighthouse
1 paths audited
Performance: 57 (🔴 down 8 from production)
Accessibility: 82 (🔴 down 7 from production)
Best Practices: 92 (🔴 down 8 from production)
SEO: 92 (no change from production)
PWA: 80 (no change from production)
View the detailed breakdown and full score reports
🤖 Make changes Run an agent on this branch

To edit notification comments on pull requests, go to your Netlify project configuration.

@coderabbitai

coderabbitai Bot commented Oct 13, 2026

Copy link
Copy Markdown
Contributor

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
  • 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

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

@mihow

mihow commented Oct 29, 2026 •

Copy link
Copy Markdown
Collaborator Author

Claude says: A related gap, found while fixing project creation in #1413: a pull-mode worker registers its pipelines only for the projects that exist when it registers. A project created later gets none of them until registration runs again. In this template, start.sh runs register.py once at container start, for the single project named by ANTENNA_PROJECT_ID or ANTENNA_DEFAULT_PROJECT_NAME. That suits a local stack, but it is the same one-off pattern.

The data companion side is tracked in RolnickLab/ami-data-companion#179. If the template keeps its own registration, it may be worth matching whatever is decided there, for example re-registering on a schedule. #1194 (API-key auth for processing services) overlaps the same registration code.

@mihow mihow added PSv2 Async & distributed ML backend (PSv2): job state, NATS dispatch, result handling. Umbrella #515. Pipeline API Updates to the requests & responses to/from processing service workers for ML pipelines labels Oct 29, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Pipeline API Updates to the requests & responses to/from processing service workers for ML pipelines PSv2 Async & distributed ML backend (PSv2): job state, NATS dispatch, result handling. Umbrella #515.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant