perf: rewrite fair claim as LATERAL top-K per company

[Xander-AJ]

Replaces the global ROW_NUMBER() window over every eligible row with a LATERAL
top-K-per-company candidate selection, eliminating the sort that dominated the old
plan (82k rows at 100k scale, 830k rows + 32 MB on-disk merge at 1M) without
regressing any concurrency, fairness, or correctness property.

Plan comparison (old vs new, verbatim plan lines)

Workload identical for both queries at each scale: seed 42, 100k / 1M tasks, 50
companies, 80% flooder, 20% future run_after. EXPLAIN (ANALYZE, BUFFERS, VERBOSE)
with track_io_timing=on, warmup discarded, 3 steady runs.

Metric	OLD 100k	NEW 100k	OLD 1M	NEW 1M
Steady-state exec time (runs 1–3)	81.647 / 85.534 / 99.678 ms	47.258 / 53.010 / 50.109 ms	1140.892 / 1041.530 / 1108.504 ms	580.859 / 557.606 / 562.137 ms
Temp read/written (top node)	temp read=1307 written=1308	(none — no temp lines)	temp read=16068 written=16082	(none — no temp lines)
Partial index ix_tasks_pending_run_after used?	No — Index Scan using ix_tasks_company_id	No — Bitmap Index Scan on ix_tasks_company_id	No — Index Scan using ix_tasks_company_id	No — Bitmap Index Scan on ix_tasks_company_id
Sort Method on final ORDER BY c.rn, c.created_at	quicksort Memory: 122kB	quicksort Memory: 130kB	quicksort Memory: 122kB	quicksort Memory: 130kB

What changed: the old plan sorts the whole eligible set for the window — 1M:
Incremental Sort (... rows=830496 ...) ... Peak Disk: 32552kB with
temp read=16098 written=16112. The new plan runs the window over only
companies×cap rows — WindowAgg (... rows=500 ...) fed by Sort ... quicksort Memory: 56kB, in memory, no temp — with per-company work bounded by
Limit (... rows=10 ... loops=50) -> top-N heapsort Memory: 26kB -> Bitmap Heap Scan on public.tasks tasks_1. Net: ~1.7× faster at 100k, ~2× at 1M, disk sort
eliminated at both scales.

Invariants confirmed (enforcing SQL quoted)

1. Tier-2 EvalPlanQual re-check preserved — the outer WHERE on t repeats
   the eligibility predicate exactly, so a row concurrently flipped between
   candidate-selection and lock is dropped by Postgres's row re-fetch:

   WHERE (t.status = 'pending'    AND t.run_after <= :now)
      OR (t.status = 'processing' AND t.locked_at < :stale_before)

   Plans confirm the outer lock node re-applies it (Index Scan using tasks_pkey on public.tasks t ... Filter: ...). Proven by
   test_lateral_high_contention_no_duplicates (20 iterations × 10 threads),
   test_no_duplicate_concurrent_processing, test_concurrent_claim_no_duplicates.
2. FOR UPDATE OF t SKIP LOCKED on the OUTER tasks alias — not the LATERAL
   subquery / CTE alias. LockRows sits directly above Index Scan using tasks_pkey on public.tasks t in every plan.
3. Stale recovery stays unconditional — the processing branch carries no
   run_after filter in any of its three locations (eligible_companies, LATERAL
   inner, outer WHERE): OR (status = 'processing' AND locked_at < :stale_before). Proven by test_lateral_stale_recovery_bypasses_run_after
   (stale-locked task with run_after 30 days out is still claimed).
4. batch cap kept, per-company cap added — outer LIMIT :batch_size retained;
   LATERAL ORDER BY created_at LIMIT :per_company_cap bound from new
   settings.per_company_claim_cap (default 10). Proven by
   test_lateral_respects_per_company_cap.

Full suite: 30 passed. ruff clean, mypy clean. No existing test relaxed.

Honest note: the partial index remains unused

ix_tasks_pending_run_after (on run_after, partial WHERE status='pending') is
not used by the new query. The LATERAL inner scan keys on company_id
equality, so the planner picks ix_tasks_company_id and applies the eligibility
predicate as a Filter + in-memory top-N heapsort on created_at. The index that
would actually serve this access pattern is a composite like (company_id, created_at) scoped to eligible rows — deferred as separate work with its own
migration and test, out of scope for this PR. The remaining hot cost is now the
eligible_companies full-table Seq Scan + HashAggregate and the per-company bitmap
heap loops, not sorting; the 1M new plan also spends ~120 ms in JIT the old plan
didn't trigger (still ~2× faster net).