Add Stage E streaming dispatch concurrency cap

[WilfordGrimley]

Description

Replaces #449, which was stacked on #448's now-merged (squashed) branch
and could not be cleanly rebased onto master without hand-resolving the
lock-context re-indentation of dispatch_micro_batch against the squashed
already_voted/ignore_conflicts body - rebased and force-push was ruled
out per this repo's own no-force-push rule, so this is a fresh branch/PR
carrying the same content plus one added commit (integration tests against
the real, merged dispatch_micro_batch body). See #449 for the full
original review history/discussion.

Adds a concurrency cap for Stage E's streaming dispatch loop, addressing
the SECOND, separate failure the 2026-07-24 shakedown run hit (the first,
the vote-collision IntegrityError, was #448's own fix, now on master):
eight concurrent dispatch_micro_batch calls, all running CPU-bound
OCR/phash extraction at once, tripped the envelope's host-load bar
(envtrip-20260724T214616-be6e5db9, observed load 11.85 against the
7.0 ceiling) on a host with only 7 usable compute cores
(docs/features/catalog-completion-plan.md L1794/2248/2366's hardware
citation and its own 0.31x-slower-than-sequential CPU-bound-
oversubscription finding).

The envelope's own host-load bar is REACTIVE - check_envelope only trips
after a fresh signal sample crosses 7.0, necessarily after load has
already spiked (the incident's own trip landed 0.43s after the vote had
already landed). This cap is PROACTIVE: dispatch_micro_batch now refuses
to even start once settings.STAGE_E_MAX_CONCURRENT_DISPATCHES (default
2, env-tunable) dispatches are already running concurrently, anywhere
across this box's django-q2 worker processes - the host is never driven
past a bounded concurrency level by Stage E's own dispatches in the first
place. Both mechanisms stay in place; neither supersedes the other.

Mechanism - Postgres session-scoped advisory locks
(cardpicker.stage_e_concurrency, new module, no migration), not a
cache-based counter or a dedicated low-worker django-q queue:

• A cache-based counter was rejected: this app's cache backend is
  Django's default, per-PROCESS LocMemCache (no CACHES override in
  settings.py) and django-q2's 8 workers are separate OS processes
  (multiprocessing) - a cache-based counter would silently fail to
  coordinate across them.
• A dedicated low-worker django-q queue was rejected as disproportionate -
  a second Cluster process needs its own supervisor/deployment wiring, a
  much larger blast radius than a primitive enforced inside
  dispatch_micro_batch itself.
• A DB-row-based atomic counter was considered and rejected specifically
  for CRASH SAFETY: a kill -9'd worker would leave a row-based counter's
  slot permanently "claimed" with no reconciliation mechanism, unlike a
  Postgres session-scoped advisory lock, which Postgres auto-releases the
  instant the holding connection dies - matching this pipeline's own
  "truthful ledger, idempotent re-entry, zero manual cleanup" ethos
  (scripts/ops/crash_drill.sh, TestKillSafetyResumeContract) for free.

New DispatchOutcome status "throttled-concurrency-cap" - writes no
PilotRunLedger row, matching the other halted statuses' own convention.

Checklist

• I have installed pre-commit and installed the hooks with pre-commit install before creating any commits.
• I have updated any related tests for code I modified or added new tests where appropriate.
• I have manually tested my changes as follows:
  • pytest cardpicker/tests/test_stage_e_concurrency.py cardpicker/tests/test_stage_e_dispatch.py cardpicker/tests/test_local_calculate_verdicts.py (149 passed) via the host mpcautofill-pilot venv against the ephemeral testcontainer Postgres (never the live prod stack), run AFTER rebasing this branch onto current master (which now contains Fix Stage E concurrent-dispatch vote-collision IntegrityError #448's squashed content) to confirm the cap still triggers correctly against the real, merged dispatch_micro_batch body - not just the pre-merge module-level unit tests.
  • Includes genuine cross-session race tests (a raw psycopg2 connection standing in for a second django-q worker process, matching production's real "each dispatch = its own connection" shape - a discovered, load-bearing subtlety: Postgres session-level advisory locks are RE-ENTRANT within one session, so simulating "two dispatchers" by calling the acquire function twice on the SAME connection silently re-acquires the same slot instead of testing anything real; every multi-dispatcher test here uses a genuinely independent connection per simulated dispatcher), a real-OS-thread contention test with a shared "max simultaneously held" counter, and (new in this branch) a full-dispatch_micro_batch-level integration test confirming status="throttled-concurrency-cap" with zero ledger/evidence/vote writes when every slot is externally held, and normal completion once released.
  • black --check, ruff check, isort --check, mypy all clean on every touched file; docs_lint.py clean (including the mechanical CLEAN-row tether for the new extractable-primitives.md row).
• I have updated any relevant documentation or created new documentation where appropriate.
  • docs/features/stage-e-operations.md: new "Concurrency cap" subsection, renumbered dispatch-ordering steps, and the reactive-vs-proactive framing against the envelope.
  • docs/upstreaming/extractable-primitives.md: new CLEAN row for cardpicker.stage_e_concurrency (zero fork-only imports - a genuinely generic Postgres-advisory-lock concurrency primitive).