Implementation — Operational hardening

What shipped

app/services/health.py (new) — check_supabase() (PostgREST references probe in an executor), check_rabbitmq() (publisher ping(); unconfigured broker fails only in prod), readiness() gathering both with 5s timeouts.
app/api/router.py — /health and /health/live are liveness-only (unchanged 200 body, so the Docker healthcheck and deploy probe keep working); /health/ready returns 200/503 with per-check statuses. Check details are logged, not exposed.
app/events/publisher.py — UsagePublisher.ping() verifies connection/channel and drops the cached exchange on a dead channel so the next publish re-establishes it.
requirements.txt — supabase==2.31.0, aio-pika>=9.4,<10 (scraper's specifiers).
app/services/pdf_fetcher.py (new) — SSRF-safe fetch_pdf(): http(s) only, host allowlist (PDF_SOURCE_ALLOWED_HOSTS setting, default koutoubi.mr). The Supabase origin is admitted only for Storage presigned URLs — exact scheme + host + port match against SUPABASE_URL and the /storage/v1/object/sign/ path prefix — because in prod that origin is the internal Kong gateway (GoTrue/PostgREST live there too). Private/loopback/link-local IP literals and localhost-style names are rejected outright as defense-in-depth. Manual redirect following (max 5) re-validates each hop; a redirect without a Location header raises an explicit HTTPStatusError. Used by all three admin PDF endpoints (400 on violation, generic 502 on download failure with the cause logged server-side) and the ingestion pipeline (job fails with PDF download failed: ...).
app/core/auth.py — token-verification failures log the real cause via our_logs and return a generic 401 "Authentication failed"; no internal exception text reaches clients.
app/services/ingestion.py — new claim_job(job_id): atomic conditional queued -> parsing update, audit entry records worker_id (host:pid). execute_job(..., claimed=False) claims unless the caller already won.
app/services/background_jobs.py — the drain loop selects up to 5 queued candidates and claims before executing, falling through when another instance wins a claim.
app/api/routers/admin.py — /admin/jobs/dispatch claims before scheduling the background task (claimed=True), trying up to 5 candidates.
app/core/middleware.py — documented that the in-memory rate limiter is a per-process backstop only; the gateway throttler is the enforcement point (deliberately not extended).

Notes / follow-ups

A FOR UPDATE SKIP LOCKED claim-next RPC with claimed_by/claimed_at columns requires a migration (migration repo) — the conditional-update claim already guarantees a single winner; the RPC would only let losers skip contended rows server-side.
Scraper stops inserting into ingestion_jobs in a separate task (rag owns the lifecycle).

Verification

pytest: 282 passed (36 new: pdf_fetcher allowlist/redirect cases, health liveness + readiness matrix, auth generic-error cases, claim/dispatch/drain race cases). ruff check clean.