Add Stage E streaming dispatch concurrency cap

[WilfordGrimley]

Description

Stacked on #448 (base branch is worktree-agent-aed7d968a5fd22516, PR
#448's own branch) - kept as a genuinely separate PR per Tron's own
instruction on #448's review, but the code depends on fields/wiring #448
introduces (stage_d_join_key_already_voted/_fallback_already_voted,
the try block shape around Stage C/D), so it is sequenced to merge
after #448, not independently. Please merge #448 first, then retarget/
rebase this PR onto master before merging it - do not delete #448's
branch until this PR (a stacked child) has been retargeted, per this
repo's own merge-duty discipline.

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, is #448's own fix): 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 (31 passed) via the host mpcautofill-pilot venv against the ephemeral testcontainer Postgres (never the live prod stack) - 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) and a real-OS-thread contention test with a shared "max simultaneously held" counter, not just sequential calls made to look concurrent.
  • 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).

[WilfordGrimley]

Superseded by #450: #448 merged (squash) and this branch (stacked on #448's now-gone branch) could not be cleanly rebased onto master without hand-resolving a conflict, and force-push is against this repo's own standing rule. #450 carries the same content, rebased onto current master, plus one added commit with integration tests verifying the cap against the real merged dispatch_micro_batch body.