backend/app/services/worker_control.py

@@ -66,6 +66,29 @@ log = logging.getLogger(__name__)
 # report "not present", which is true, rather than hang the page.
 CONTROL_TIMEOUT_SECONDS = 2.0
 
+# A celery prefork pool cannot have ZERO processes, and a lane at zero slots
+# is expressed by cancelling its consumers rather than by emptying its pool.
+#
+# The generator already starts every lane at one process for exactly this
+# reason: `add_consumer` needs something to reach, so enabling a lane from the
+# UI would be impossible if no process existed. The consequence was missed
+# until the operator's live deploy, 2026-09-23:
+#
+#   [scheduler] worker_control: ml reconciled 1 -> 0 slots
+#   [ml] pidbox command error: ValueError("Can't shrink pool. All processes
+#        busy!")
+#
+# billiard refuses to remove the last worker, so ml could never reach 0. And
+# `set_lane_slots_sync` returns True on SENDING the control message — the
+# failure happens later, on the worker — so the reconcile logged success and
+# reported `changed: ['ml']` on every single tick. An enforcer with no
+# reachable fixed point, re-doing its own work forever and saying it worked:
+# lesson #4183, in production, on the default configuration of every install.
+#
+# So the floor is one process. Zero slots still means zero WORK, because the
+# consumers are cancelled — the idle process is the switch's landing pad.
+MIN_POOL_SLOTS = 1
+
 # The WORST case of `inspect_lanes_sync`, for callers that need a deadline.
 #
 # One broadcast plus three targeted reads. The targeted three normally return
@@ -208,6 +231,16 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]:
     return out
 
 
+def effective_slots(target: int) -> int:
+    """What a pool can actually be set to. Never below one process.
+
+    Used wherever a target is COMPARED as well as wherever one is sent: a
+    reconcile that compares against the unclamped number sees a difference
+    that no control message can ever close, and re-sends it every tick.
+    """
+    return max(MIN_POOL_SLOTS, target)
+
+
 def set_lane_slots_sync(
     lane: Lane, target: int, live: LaneLiveState | None = None,
 ) -> tuple[bool, str | None]:
@@ -223,6 +256,7 @@ def set_lane_slots_sync(
     stored the value; this says only that the live push did not land, and the
     reconcile will carry it when the lane answers again.
     """
+    target = effective_slots(target)
     try:
         from ..celery_app import celery as celery_app
 
@@ -663,6 +697,10 @@ def reconcile_lanes_sync(
         target_slots = slots
         if lane.name in autoscaling and current is not None:
             target_slots = max(slots, current)
+        # And the floor every lane has: a pool cannot be emptied. Applied to
+        # the COMPARISON, not just the send — against the raw 0 this sees a
+        # difference no control message can close and re-sends it every tick.
+        target_slots = effective_slots(target_slots)
         if current is not None and current == target_slots:
             continue
         ok, err = set_lane_slots_sync(lane, target_slots, live=state)
@@ -817,8 +855,10 @@ def autoscale_lanes_sync(
                 lane.name, "held", current,
                 f"{backlog} waiting but the cap is {cap}",
             ))
-        elif current > configured and backlog <= AUTOSCALE_SHRINK_BELOW:
+        elif current > effective_slots(configured) and (
-            new = max(configured, current - AUTOSCALE_STEP)
+            backlog <= AUTOSCALE_SHRINK_BELOW
+        ):
+            new = max(effective_slots(configured), current - AUTOSCALE_STEP)
             ok, err = set_lane_slots_sync(lane, new, live=state)
             out.append(AutoscaleDecision(
                 lane.name, "shrank" if ok else "held", new if ok else current,

tests/test_gen_supervisord.py

@@ -270,3 +270,34 @@ def test_supervisorctl_can_reach_supervisord():
     assert talking == f"unix://{listening}", (
         f"supervisorctl talks to {talking}, supervisord listens on {listening}"
     )
+
+
+def test_each_program_starts_at_the_smallest_pool_the_control_path_allows():
+    """The two ends of the same floor, asserted together.
+
+    `gen_supervisord` starts every lane at `max(1, default_slots)` because
+    billiard will not run a pool of zero. `worker_control` has the same floor
+    for the opposite reason: it cannot SHRINK to zero either —
+
+        [ml] pidbox command error:
+             ValueError("Can't shrink pool. All processes busy!")
+
+    Live, 2026-09-23. ML starts at one process and stores zero, so the
+    reconcile tried 1 -> 0 on every tick, billiard refused, and
+    `set_lane_slots_sync` — which returns True on SENDING the message —
+    reported the lane changed forever (lesson #4183, on the default
+    configuration of every install).
+
+    Two constants, in two files, that must agree or the container cannot
+    settle. Asserted through `effective_slots` rather than against a literal
+    1, so raising the floor moves both ends at once.
+    """
+    from backend.app.services.worker_control import effective_slots
+
+    cp = _parse()
+    for lane in LANES:
+        env = cp.get(f"program:{lane.name}", "environment")
+        want = effective_slots(lane.default_slots)
+        assert f"CELERY_CONCURRENCY={want}," in env, (
+            f"{lane.name} starts at a size the control path cannot reach"
+        )

tests/test_worker_control.py

@@ -666,3 +666,75 @@ def test_the_round_trip_bound_matches_the_reads_actually_made():
         f"inspect_lanes_sync makes {reads} reads but CONTROL_ROUND_TRIPS "
         f"says {wc.CONTROL_ROUND_TRIPS}"
     )
+
+
+# --- a pool cannot be emptied ------------------------------------------------
+
+
+def test_a_lane_at_zero_slots_is_never_shrunk_to_an_empty_pool(monkeypatch):
+    """The error in the operator's live log, 2026-09-23:
+
+        [scheduler] worker_control: ml reconciled 1 -> 0 slots
+        [ml] pidbox command error:
+             ValueError("Can't shrink pool. All processes busy!")
+
+    billiard refuses to remove the last worker, so a lane at zero stored slots
+    could never reach its target — and `set_lane_slots_sync` returns True on
+    SENDING the message, so the reconcile logged success and reported the lane
+    as `changed` on every tick, forever. Lesson #4183 in production, on the
+    default configuration of every install.
+    """
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "ml", pools={"host-a": 1}, consuming=set())
+
+    wc.reconcile_lanes_sync({"ml": (0, False)})
+
+    assert control.shrank == [], "tried to empty a pool billiard will not empty"
+
+
+def test_that_lane_then_reports_nothing_to_do(monkeypatch):
+    """The fixed point. It is not enough to stop sending the doomed message —
+    the reconcile must also stop CALLING the lane changed, or the log fills
+    with a correction that never corrects anything."""
+    _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "ml", pools={"host-a": 1}, consuming=set())
+
+    result = wc.reconcile_lanes_sync({"ml": (0, False)})
+
+    assert result["changed"] == []
+
+
+def test_zero_slots_still_means_zero_work(monkeypatch):
+    """The floor is one PROCESS, not one consumer. A lane at zero keeps an
+    idle worker so the enable switch has something to reach, and stays off
+    because its queues are cancelled."""
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "ml", pools={"host-a": 1}, consuming={"ml"})
+
+    wc.reconcile_lanes_sync({"ml": (0, False)})
+
+    assert control.cancelled == [("ml", ["host-a"])]
+
+
+def test_the_floor_applies_to_a_direct_resize_too(monkeypatch):
+    """Not only the reconcile — the UI dial and the autoscaler go through the
+    same function, so the clamp belongs there rather than at each caller."""
+    control = _stub_control(monkeypatch)
+    state = _stub_live(monkeypatch, "worker", pools={"host-a": 3})
+
+    wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 0, live=state)
+
+    assert control.shrank == [(2, ["host-a"])], "should stop at one, not zero"
+
+
+def test_the_autoscaler_will_not_shrink_into_an_empty_pool(monkeypatch):
+    """A lane whose operator value is 0 and whose live pool is the floor has
+    nowhere to shrink to — and saying `shrank` every minute is the same
+    non-convergence wearing a different hat."""
+    control = _stub_control(monkeypatch)
+    _autoscale_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0)
+
+    d = _worker(wc.autoscale_lanes_sync({"worker": (8, 0, True)}))
+
+    assert d.action == "held"
+    assert control.shrank == []