…tor RPCs run in parallel

The PG file-processing path ran files SERIALLY: worker-pg-fileproc was a single
batch=1 consumer, so multi-file ETL batches drained one at a time (Celery runs
them in parallel via prefork --concurrency). Same single-consumer bottleneck on
worker-pg-executor serialized the executor RPCs (the LLM extraction), so even
parallel file fan-out gained nothing end-to-end.

Fix: a prefork supervisor for the PG-queue consumer launcher — the PG analogue of
Celery --pool=prefork --concurrency=N.

- supervisor.py (new): when WORKER_PG_QUEUE_CONSUMER_CONCURRENCY > 1, fork N
  isolated consumer children (each does its own worker bootstrap, so no
  connections are inherited across the fork), monitor + re-fork dead children, and
  own a single fleet-liveness endpoint (503 when the oldest child stalls past the
  threshold). SKIP LOCKED distributes batches across children AND replicas; a
  single execution is still capped by MAX_PARALLEL_FILE_BATCHES; total live
  parallelism = concurrency x replicas (k8s HPA scales replicas).
- The consumer code is UNCHANGED — each child is the current single-threaded
  PgQueueConsumer.run(); concurrency is purely a launch concern. CONCURRENCY=1
  (default) keeps the byte-identical single-process path, so every other PG
  consumer is non-regressive.
- consumer.py: extract build_consumer_from_env()/consumer_env() so the children
  and main() build identical consumers.
- docker-compose: CONCURRENCY=4 on worker-pg-fileproc AND worker-pg-executor
  (mirrors WORKER_FILE_PROCESSING_CONCURRENCY). Also raise their VT_SECONDS to
  3660 — process_file_batch/execute_extraction block on the executor up to 3600s,
  so the prior vt=30 default would re-claim a long batch mid-run (latent
  double-run); health-stale sits just above vt.
- Tests: env-knob parse/clamp/validation, fleet-staleness calc, reap/restart
  matrix (dead -> re-fork, live -> left, shutdown -> not resurrected).

Process model matches Celery prefork (the cloud-trusted choice): full process
isolation, a crash is contained + re-forked, its in-flight message redelivers via
vt (at-least-once). Dev-tested end-to-end: 4 fileproc + 4 executor children, a
2-file ETL runs fully parallel (both files + both executor RPCs overlap on
separate children; ~25s vs ~42s serial), child crash auto-restarts, health green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…dening

Critical/High:
- Crash-loop no longer hides a wedged fleet from the probe: re-fork does NOT
  reseed the heartbeat (a never-polling slot ages), and a slot that dies
  immediately N times in a row forces freshness=inf → 503 (k8s restarts the pod).
- os.fork() wrapped: initial-fleet failure fails fast with an actionable message;
  reap-path failure logs + leaves the slot for the next tick (no uncaught crash /
  no SIGTERM storm against healthy children).
- Heartbeat publish loop guarded (try/except + log) so a transient error can't
  silently kill the thread and false-stale a healthy child.
- Children reset SIGTERM/SIGINT to SIG_DFL immediately after fork — a signal in
  the fork→run window no longer fires the parent's _on_term against stale pids.

Medium:
- _Fleet class owns pid/last_fork/heartbeat/crash-count/restart-schedule with a
  validated slot and consistent reap (replaces three loose slot-keyed dicts).
- Re-fork backoff is a scheduled not-before (non-blocking), re-checking stopping
  before each fork — no in-loop sleep serialising recovery, no child spawned into
  shutdown.
- Liveness bind: EADDRINUSE logged at error (vs transient), and
  liveness_probe_bound surfaced in the status payload.
- _join_children: per-child grace budget (one slow child can't starve the rest);
  _wait_for_exit helper extracted.

Low:
- real Callable annotations (drop noqas); HEALTH_PORT typed int|None; comment
  fixes (heartbeat writer reasoning, "JSON body differs" for the fleet probe);
  WORKER_TYPE selection deduped into pg_queue_consumer/_bootstrap.py.

Tests rewritten + expanded: _Fleet bookkeeping/crash-loop/freshness, reap
scheduling, restart-due gating, fork OSError + child hard-exit guard, _wait_for_exit
+ _join_children SIGKILL escalation, health no-port/bind-error. 32 tests.
Re-validated live: 4 children, fleet health (crash_looping/liveness_probe_bound
fields), crash → re-fork.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

is_crash_looping() runs in the LivenessServer daemon thread (via freshness())
while the main thread mutates _consecutive_crashes in schedule_restart() — a bare
.values() iteration could raise "dictionary changed size during iteration".
Snapshot with tuple(...) (atomic under the GIL) before iterating.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>