fix: the lane sizer reads the live pool, so it stops growing a lane past its cap (4409)

It read celery's `max-concurrency` as the pool size, but prefork reports that
as the limit the pool booted with. pool_grow and pool_shrink never update it.
A lane booted at 1 therefore read 1 forever. The sweep sent `target - 1` on
every tick while work waited, and could never shrink, since 1 - 1 is 0. The
System tab showed Scheduler "6 / 1" at a cap of 2.

pool_size() counts `processes` and falls back to the limit only when there is
no process list.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/services/worker_control.py

@@ -211,11 +211,10 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]:
         state.reserved += len(reserved.get(hostname, []))
         state.consuming.update(q["name"] for q in queues)
 
-        # `pool.max-concurrency` is the number pool_grow/pool_shrink move and
-        # the number the UI shows. Absent on a worker whose stats did not
-        # answer, which leaves pool=None — unknown, not zero.
-        pool = (stats.get(hostname) or {}).get("pool", {}).get("max-concurrency")
-        if isinstance(pool, int):
+        # Absent on a worker whose stats did not answer, which leaves
+        # pool=None — unknown, not zero.
+        pool = pool_size((stats.get(hostname) or {}).get("pool") or {})
+        if pool is not None:
             state.pools[hostname] = pool
 
     for state in out.values():
@@ -223,6 +222,27 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]:
     return out
 
 
+def pool_size(pool_stats: dict) -> int | None:
+    """How many processes a prefork pool is running NOW, from `inspect stats`.
+
+    The length of `processes`, not `max-concurrency`. Celery reports
+    `max-concurrency` as the pool's `limit`, set once at boot; `pool_grow` and
+    `pool_shrink` hand straight to billiard and never touch it (celery 5.6
+    `concurrency/prefork.py`: `self.grow = P.grow`). So it read 1 forever on a
+    lane booted at 1, and the sizing sweep, computing `target - 1` on every
+    tick, grew a scheduler capped at 2 to six processes while the System tab
+    showed "6 / 1" (2026-09-24) — and could never shrink one, since 1 - 1 is 0.
+
+    `max-concurrency` is the fallback only for a pool that lists no processes
+    (a non-prefork pool), where it is the best number there is.
+    """
+    procs = pool_stats.get("processes")
+    if isinstance(procs, list):
+        return len(procs)
+    limit = pool_stats.get("max-concurrency")
+    return limit if isinstance(limit, int) else None
+
+
 def effective_slots(target: int) -> int:
     """What a pool can actually be set to. Never below one process.
 

tests/test_worker_control.py

@@ -671,3 +671,36 @@ def test_an_unknown_worker_is_still_ignored_rather_than_guessed_at(monkeypatch):
     live = wc.inspect_lanes_sync()
 
     assert all(not s.present for s in live.values())
+
+
+# --- the pool size is the processes running, not the boot limit ---------------
+#
+# The operator's System tab, 2026-09-24: Scheduler "6 / 1" at a cap of 2. Celery
+# reports `max-concurrency` as the limit the pool BOOTED with; pool_grow and
+# pool_shrink never update it. The sweep read 1 forever, sent `target - 1` on
+# every tick, and grew the lane past its cap with no way to shrink it back.
+
+
+def test_pool_size_counts_the_live_processes_not_the_boot_limit():
+    """After two pool_grow calls on a pool booted at 1."""
+    assert wc.pool_size({"max-concurrency": 1, "processes": [11, 12, 13]}) == 3
+
+
+def test_pool_size_falls_back_to_the_limit_without_a_process_list():
+    assert wc.pool_size({"max-concurrency": 4}) == 4
+    assert wc.pool_size({}) is None
+
+
+def test_a_grown_pool_reads_at_its_real_size(monkeypatch):
+    """The read the sweep sizes from: a pool at its target must read AT its
+    target, or `target - current` never reaches zero and every tick grows it."""
+    _stub_active_queues(monkeypatch, {"scheduler@h": []})
+    import sys
+    insp = sys.modules["backend.app.celery_app"].celery.control.inspect
+    insp.stats = lambda self: {
+        "scheduler@h": {"pool": {"max-concurrency": 1, "processes": [1, 2]}},
+    }
+
+    live = wc.inspect_lanes_sync()
+
+    assert live["scheduler"].pool == 2