feat: one number per lane — the cap — and the autoscaler is the mechanism (4295)

Operator, 2026-09-23: *"auto should be always on, not a setting, so that idle
instances quiet down when not running. the number that is visible and
something the user can tweak and manage should be the cap itself the number of
running workers is handled by the autoscaling function which is always on."*

They are right, and the reason it was not built this way is worth stating: the
manual dial came first (steps 2-4) and the autoscaler came last (step 7), as
an opt-in BESIDE a control that already existed. Nothing ever asked whether
the dial should still exist once something could move it automatically. Each
step was defensible; the result was three operator settings over one number.

## `slots`, `enabled` and `autoscale` are gone

`slots` was a MEASUREMENT wearing a preference's clothes. How many workers a
lane runs is read live and moved every minute; storing it meant the operator
had to keep two numbers in agreement and the autoscaler had to be told it was
allowed to touch one of them.

`autoscale` gated the mechanism behind a choice, so a lane nobody opted in
never gave its workers back — which is why an idle instance never quieted
down.

`enabled` is derived: a cap of zero means no consumers. "Off" and "may use no
workers" were two spellings of one fact, stored separately, free to disagree.

## Two sweeps become one

`reconcile_lanes_sync` drove the pool to the stored `slots`; `autoscale_lanes_
sync` moved it away from that same number; and most of step 7's hardest
reasoning — a stored value that is a FLOOR, a target of `max(stored, current)`
— existed only to stop them fighting. Delete the stored number and the problem
is not solved, it is absent.

`size_lanes_sync` runs every minute and owns both consumers and pool size. It
also subsumes what the reconcile was for: a worker restarted at its ENV
concurrency is corrected on the next tick rather than after five.

Growth is immediate, shrink is one worker per tick. Deliberately asymmetric —
"always on" is only pleasant if the ramp keeps up, and +1/minute would take
four minutes to answer a burst. Being one worker too large for a minute costs
a sleeping process; being too small costs work not happening. For ML the
asymmetry matters most: every new slot reloads a multi-GB model, so the slow
shrink is what stops a quiet patch from paying that cost again a minute later.

## The caps ship at one, and zero for ML

Per the operator. Conservative on purpose — and a conservative default nobody
knows how to raise is just a slow product, which is the other half of what
they asked for:

    "there needs to be something that tells the user to bump those numbers to
     improve processing rate or they'd never know the controls exist."

So a lane running everything its cap allows while work piles up says so, in
its own row, with the headroom named: *"4,060 waiting and all 1 worker busy.
Raise the cap to run more at once — this machine allows up to 7."*

It fires only when raising the cap would actually help. Not when the lane is
keeping up, not when the sizing pass has room it has not taken, and not at the
machine ceiling — where "raise the cap" is advice nobody can take.

## Migration 0105 rewrites the caps rather than carrying them

The old defaults (4/2/2/1) bounded a manual control and were loose because
moving within them was the ordinary act. The number now means "the most
workers this lane may use", which is a different promise; carrying the old
figure over would quadruple the worker lane on every existing install at the
moment this deploys.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/services/worker_control.py

@@ -56,7 +56,13 @@ from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import TaskRun, WorkerLane
-from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling
+from .worker_lanes import (
+    LANES,
+    LANES_BY_QUEUE_KEY,
+    MIN_POOL_SLOTS,
+    Lane,
+    derived_ceiling,
+)
 
 log = logging.getLogger(__name__)
 
@@ -66,28 +72,6 @@ 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.
 #
@@ -342,13 +326,7 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]:
     }
     missing = [lane for lane in LANES if lane.name not in rows]
     for lane in missing:
-        row = WorkerLane(
-            name=lane.name,
-            slots=lane.default_slots,
-            slots_cap=lane.default_slots_cap,
-            enabled=lane.default_enabled,
-            autoscale=lane.default_autoscale,
-        )
+        row = WorkerLane(name=lane.name, slots_cap=lane.default_slots_cap)
         session.add(row)
         rows[lane.name] = row
     if missing:
@@ -398,17 +376,18 @@ async def lane_view(session: AsyncSession) -> list[dict]:
             "name": lane.name,
             "display_name": lane.display_name,
             "queues": list(lane.queues),
-            "slots": row.slots,
             "slots_cap": row.slots_cap,
             "ceiling": derived_ceiling(lane),
-            "enabled": row.enabled,
-            "autoscale": row.autoscale,
+            # DERIVED, never stored. A cap of zero means no consumers, so
+            # "off" and "may use no workers" cannot disagree.
+            "enabled": row.slots_cap > 0,
             "memory_bound": lane.memory_bound,
             "optional": lane.optional,
-            # What enabling this lane will download, so the UI can say WHICH
-            # model and how big BEFORE the switch is thrown rather than after
-            # a multi-GB fetch has started. `measured` travels with the
-            # numbers: the card must not present an estimate as a fact.
+            # What raising this lane's cap will download, so the UI can say
+            # WHICH model and how big BEFORE the first slot is asked for
+            # rather than after a multi-GB fetch has started. `measured`
+            # travels with the numbers: the UI must not present an estimate
+            # as a fact.
             "models": [
                 {
                     "repo": m.repo,
@@ -506,119 +485,89 @@ class LaneUpdateRefused(ValueError):
     the UI shows — a greyed control with no explanation reads as a bug."""
 
 
-async def set_lane(
-    session: AsyncSession,
-    lane: Lane,
-    *,
-    slots: int | None = None,
-    slots_cap: int | None = None,
-    enabled: bool | None = None,
-    autoscale: bool | None = None,
-) -> dict:
-    """Store the operator's choice, then push it to the running lane.
+async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict:
+    """Store the operator's cap for `lane`, then make the live lane obey it.
 
-    BOTH, in one call, and the order matters. `pool_grow`/`pool_shrink` are
-    not durable — a restart drops every lane back to its env concurrency — so
-    a UI that only pushed would have its setting evaporate on the next deploy
-    with nothing to show for it (lesson #4202: the live change does not
-    survive, and nothing says so). Storing alone would be a number that
-    describes nothing until something restarts.
+    ONE value, since 2026-09-23. It used to take `slots`, `slots_cap`,
+    `enabled` and `autoscale`, which was four ways of saying two things — and
+    two of them were the caller's job to keep in agreement with each other.
+
+    ## What happens live, and what does not
+
+    A cap CHANGE is pushed immediately in one direction only: lowering it
+    shrinks the pool now, because a cap the operator just lowered should not
+    be exceeded for up to a minute. Raising it does NOT grow the pool here —
+    a cap is permission, not a request, and growing on permission would put
+    slots on a lane with nothing to do. The sizing pass adds them on its next
+    tick if there is work, which is the whole point of it being always on.
+
+    Consumers follow the cap in both directions and immediately: zero means
+    off, and off must take effect when it is asked for.
 
     A failed PUSH is not a failed setting. The value is saved either way and
-    step 3's reconcile carries it when the lane answers again; the result says
-    `applied: false` with a reason so the UI can say "saved, not yet live"
-    rather than "that didn't work".
+    the sizing pass carries it within a minute; the result says `applied:
+    false` with a reason so the UI can say "saved, not yet live" rather than
+    "that didn't work" (lesson #4202 — a live change that does not survive,
+    with nothing saying so).
     """
     rows = await _rows_by_name(session)
     row = rows[lane.name]
 
-    new_cap = row.slots_cap if slots_cap is None else slots_cap
-    new_slots = row.slots if slots is None else slots
-    new_autoscale = row.autoscale if autoscale is None else autoscale
-
-    # THE DIAL IS THE SWITCH. A lane at zero slots is a lane that is off, and
-    # there is no second control saying so.
-    #
-    # Operator, 2026-09-23, on the card that had both: *"there's nothing to
-    # describe what 'auto' means or why their needs to be or should be on/off
-    # toggles. almost all of it always needs to run there's only one optional
-    # piece and it is killed by moving the 'cap' to zero."* They are right. Of
-    # four lanes, three must run for the application to work at all, so a
-    # switch beside each of them offered a choice that was never real — and
-    # for the one lane that IS optional, "off" and "zero slots" were two ways
-    # of saying the same thing that could disagree with each other.
-    #
-    # `enabled` stays in the model and on the API. It is still the mechanism:
-    # a disabled lane keeps its process and cancels its consumers, which is
-    # what makes it visible in the roster instead of looking like a crash. It
-    # is now DERIVED from the number the operator actually sets, rather than
-    # being a second thing for them to keep in agreement with it.
-    was_enabled = row.enabled
-    if enabled is not None:
-        new_enabled = enabled
-    elif slots is not None:
-        new_enabled = new_slots > 0
-    else:
-        new_enabled = row.enabled
-
     ceiling = derived_ceiling(lane)
-    if new_cap < 0 or new_slots < 0:
-        raise LaneUpdateRefused("slots and cap cannot be negative")
-    if new_cap > ceiling:
+    if slots_cap < 0:
+        raise LaneUpdateRefused("a cap cannot be negative")
+    if slots_cap > ceiling:
         raise LaneUpdateRefused(
-            f"cap {new_cap} is above what this container can hold "
+            f"a cap of {slots_cap} is above what this container can hold "
             f"({ceiling} for {lane.display_name})"
         )
-    if new_slots > new_cap:
-        raise LaneUpdateRefused(f"slots {new_slots} is above the cap {new_cap}")
 
-    row.slots_cap = new_cap
-    row.slots = new_slots
-    row.enabled = new_enabled
-    row.autoscale = new_autoscale
+    was_on = row.slots_cap > 0
+    row.slots_cap = slots_cap
     await session.commit()
 
+    now_on = slots_cap > 0
     applied, error = True, None
-    # On the CHANGE, not on the field being present. Now that `enabled` is
-    # derived, every slots write would otherwise re-send a consumer command
-    # that changes nothing — the churn lesson #4183 keeps producing, arriving
-    # here through the new derivation.
-    if new_enabled != was_enabled:
+    if now_on != was_on:
+        applied, error = await asyncio.to_thread(set_lane_enabled_sync, lane, now_on)
+    if applied and not now_on:
+        # Down to the floor at once. The pool cannot be emptied, so "off" is
+        # one parked process with its consumers cancelled.
         applied, error = await asyncio.to_thread(
-            set_lane_enabled_sync, lane, new_enabled,
+            set_lane_slots_sync, lane, MIN_POOL_SLOTS,
         )
-    if applied and slots is not None:
-        applied, error = await asyncio.to_thread(set_lane_slots_sync, lane, new_slots)
+    elif applied and now_on:
+        # Only DOWNWARD. See the docstring: raising a cap is permission, and
+        # the sizing pass decides whether there is work to spend it on.
+        live = await asyncio.to_thread(inspect_lanes_sync)
+        current = live[lane.name].pool
+        if current is not None and current > slots_cap:
+            applied, error = await asyncio.to_thread(
+                set_lane_slots_sync, lane, slots_cap, live=live[lane.name],
+            )
 
-    # Enabling a lane that needs models is what triggers the fetch (milestone
+    # Raising the cap off zero is what triggers the model download (milestone
     # 422 step 6). Never at boot: that made every start of the ML role reach
     # HuggingFace for ~3.5GB, and rule 164 permits a runtime fetch only for a
     # feature that is optional and clearly OFF.
     #
-    # Only on the TRANSITION from off to on, so re-saving slots on a lane that
-    # is already running does not re-enqueue. This used to test `enabled is
-    # True` — the field having been sent — which stopped meaning "came on" the
-    # moment the dial became the switch: the UI no longer sends `enabled` at
-    # all, so the fetch that makes the ML lane usable would never have fired.
-    #
-    # And only when the consumer change landed: enqueueing onto a queue
-    # nothing is consuming would leave the task pending with no explanation
-    # until the lane returns.
+    # On the TRANSITION, so re-saving a cap on a lane already running does not
+    # re-enqueue. And only when the consumer change landed: enqueueing onto a
+    # queue nothing is consuming would leave the task pending with no
+    # explanation until the lane returns.
     fetching = False
-    if new_enabled and not was_enabled and lane.models and applied:
+    if now_on and not was_on and lane.models and applied:
         fetching = _enqueue_model_fetch()
 
     return {
         "name": lane.name,
-        "slots": row.slots,
         "slots_cap": row.slots_cap,
         "ceiling": ceiling,
-        "enabled": row.enabled,
-        "autoscale": row.autoscale,
+        "enabled": now_on,
         "applied": applied,
         "apply_error": error,
-        # Tells the card to say a download has started rather than leaving the
-        # operator to wonder why a freshly enabled lane is busy.
+        # Tells the UI to say a download has started rather than leaving the
+        # operator to wonder why a lane they just turned on is busy.
         "fetching_models": fetching,
     }
 
@@ -645,263 +594,166 @@ def _enqueue_model_fetch() -> bool:
         return False
 
 
-def reconcile_lanes_sync(
-    desired: dict[str, tuple[int, bool]],
-    autoscaling: frozenset[str] = frozenset(),
-) -> dict:
-    """Drive every RUNNING lane to its stored slots and enabled flag.
 
-    `desired` is lane name -> (slots, enabled), read from the database by the
-    caller. `autoscaling` names the lanes the autoscaler is allowed to move.
-
-    ## For an autoscaling lane the stored value is a FLOOR, not a target
-
-    Step 7's autoscaler raises a saturated lane's live pool without changing
-    its row — the row holds what the OPERATOR set. If this pass treated that
-    row as an exact target it would shrink the lane back on the very next
-    tick, and the two sweeps would fight forever at five-minute intervals:
-    grow, revert, grow, revert. That is lesson #4183's failure arriving
-    between two enforcers rather than inside one.
-
-    So for those lanes the target becomes `max(stored, current)` — this pass
-    still restores a lane that came back from a restart below what the
-    operator set, and never takes back what the autoscaler added. Bringing it
-    down is the autoscaler's job, and it does so only to that same floor. This function touches no database: the celery task that schedules
-    it owns the sync session, and keeping the DB out of here is what lets the
-    same code be called from anywhere that already knows the target.
-
-    ## Why this exists at all
-
-    `pool_grow` is not durable. A worker that dies and is restarted by its
-    supervisor comes back at its ENV concurrency — silently below whatever the
-    operator set — and nothing in step 2's path would ever notice. Storing the
-    value made it survivable; this is what makes it actually survive.
-
-    ## It must converge and then go quiet
-
-    One `inspect` for all lanes, and `set_lane_slots_sync` issues nothing at
-    all to a replica already at its target. So a settled system performs one
-    broker round trip per tick and sends no control messages — the reachable
-    fixed point lesson #4183 is about. An enforcer that re-sent a grow of zero
-    every tick would churn forever and bury a real correction in its own noise,
-    which is why `changed` below counts only lanes that actually moved.
-
-    ## An absent lane is SKIPPED, not corrected
-
-    `present=False` means nothing answered — a worker restarting, or a broker
-    that is unreachable. It does NOT mean zero slots. Correcting an absence
-    would be drawing a conclusion from an unswept read (snippet #3969), and
-    here it would be worse than useless: there is nothing to send the message
-    to. The lane is reported as skipped and picked up on a later tick.
-    """
-    live = inspect_lanes_sync()
-    changed: list[str] = []
-    skipped: list[str] = []
-    failed: dict[str, str] = {}
-
-    for lane in LANES:
-        target = desired.get(lane.name)
-        if target is None:
-            continue
-        slots, enabled = target
-        state = live[lane.name]
-        if not state.present:
-            skipped.append(lane.name)
-            continue
-
-        # Enabled first: a lane being turned on should be consuming before
-        # its pool is sized, so the slots it gains have work to pick up.
-        #
-        # Only when it DISAGREES. Calling this unconditionally would send
-        # add_consumer for every queue on every tick of a settled system —
-        # the exact churn lesson #4183 describes, and invisible because
-        # add_consumer on a queue already consumed is harmless.
-        consuming_all = state.consuming.issuperset(lane.queues)
-        if enabled != consuming_all:
-            ok, err = set_lane_enabled_sync(lane, enabled, live=state)
-            if not ok:
-                failed[lane.name] = err or "could not set consumers"
-                continue
-            changed.append(lane.name)
-
-        current = state.pool
-        # The floor, for a lane the autoscaler manages. See the docstring.
-        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)
-        if ok:
-            if lane.name not in changed:
-                changed.append(lane.name)
-            log.info(
-                "worker_control: %s reconciled %s -> %s slots",
-                lane.name, current, target_slots,
-            )
-        else:
-            failed[lane.name] = err or "could not resize"
-
-    return {"changed": changed, "skipped": skipped, "failed": failed}
-
-
-# --- the autoscaler (step 7) --------------------------------------------------
+# --- the sizing pass: one sweep, always on ------------------------------------
 #
-# The only part of this milestone that acts without anyone asking. Everything
-# above does what an operator pressed; this decides. So it is off by default,
-# opted into per lane, bounded by the cap the operator set, and it reports what
-# it did rather than moving numbers silently.
-
-# ## Why these three numbers are not in Settings
+# This replaced BOTH `reconcile_lanes_sync` (step 3) and `autoscale_lanes_sync`
+# (step 7) on 2026-09-23. They were two enforcers over one number, and the
+# whole of step 7's hardest reasoning — a stored value that is a FLOOR, a
+# target of `max(stored, current)` so the reconcile does not undo what the
+# autoscaler added — existed only to stop them fighting. Delete one of them and
+# the problem is not solved, it is absent.
 #
-# Rule 25 puts anything an operator might want to tune in the UI, and the
-# knobs that decide what this does ARE there: whether a lane autoscales at
-# all, its cap, and its floor — all DB-backed, all changeable without a
-# restart. What is left here is the POLICY's internals, and exposing them
-# would add four numbers per lane to a card whose whole value is being
-# readable at a glance, to tune a decision the operator has a better lever
-# for. If growth turns out to be too eager or too shy in practice, that is a
-# reason to change these values for everyone, not to ask each operator to
-# discover them.
-
-# All slots busy AND this many tasks waiting before a lane may grow.
+# Operator: *"auto should be always on, not a setting, so that idle instances
+# quiet down when not running. the number that is visible and something the
+# user can tweak and manage should be the cap itself the number of running
+# workers is handled by the autoscaling function which is always on."*
 #
-# The AND is the design. Depth with free slots means nothing — celery is about
-# to pick those up, and growing the pool would add idle children. Saturation
-# with an empty queue means nothing either: the lane is busy with exactly as
-# much work as exists. Only both together say "there is more work than this
-# lane can reach".
-AUTOSCALE_BACKLOG_THRESHOLD = 10
+# So there is one pass, it runs every minute, it reads the live pool rather
+# than any stored number, and the only thing it obeys is the cap.
+#
+# It also subsumes what the reconcile existed for. `pool_grow` is not durable:
+# a worker restarted by its supervisor comes back at its ENV concurrency,
+# silently below what the lane should be running. This pass reads the live
+# pool every minute and sizes from the backlog, so that worker is corrected on
+# the next tick — sooner than the five-minute reconcile managed, and without a
+# second sweep that could disagree with this one.
 
-# Grow by one slot per tick, never to the cap in one jump. A lane that is
-# saturated because of one slow burst settles a slot or two above where it
-# started rather than at its ceiling, and the next tick re-measures rather
-# than committing to a guess made once.
-AUTOSCALE_STEP = 1
-
-# Hysteresis: shrink only when the backlog is well BELOW the grow threshold,
-# not merely under it. Equal thresholds flap — one task arriving and leaving
-# would grow and shrink the lane forever at the tick interval, which is
-# lesson #4183's churn arriving through a different door.
-AUTOSCALE_SHRINK_BELOW = 2
+# How much work justifies a slot. `pending` is depth + reserved, so it already
+# counts what celery has prefetched into worker memory — one task, one slot.
+#
+# Growth is IMMEDIATE and shrink is one slot per tick, deliberately asymmetric.
+# A backlog of four thousand should not take an hour to reach the cap, and a
+# lane that idles for one minute should not drop every process it has: the
+# cost of being one slot too large for a minute is a sleeping process, and the
+# cost of being too small is work not happening. For ML the asymmetry matters
+# most — every new slot reloads a multi-GB model, so the slow shrink is what
+# stops a quiet patch from paying that cost again a minute later.
+SHRINK_STEP = 1
 
 
 @dataclass
-class AutoscaleDecision:
-    """What the autoscaler did to one lane, and why — in the operator's terms.
+class LaneSizing:
+    """What the pass did to one lane, and why — in the operator's terms.
 
-    A reason string on every outcome including "nothing", because an
-    autoscaler that only speaks when it acts is one nobody can debug when it
-    does not.
+    A reason on every outcome including "held", because a sizing pass that
+    only speaks when it acts is one nobody can debug when it does not.
     """
 
     lane: str
-    action: str  # "grew" | "shrank" | "held"
+    action: str  # "grew" | "shrank" | "held" | "skipped"
     slots: int
     reason: str
 
 
-def autoscale_lanes_sync(
-    lanes: dict[str, tuple[int, int, bool]],
-) -> list[AutoscaleDecision]:
-    """Decide and apply one round of autoscaling.
+def wanted_slots(cap: int, active: int, pending: int | None) -> int:
+    """How many workers this lane has work for right now, within its cap.
 
-    `lanes` is name -> (slots_cap, configured_slots, autoscale_on), read from
-    the database by the caller — this function touches no database, for the
-    same reason `reconcile_lanes_sync` does not.
+    One slot per task in flight or waiting, floored at one process and
+    ceilinged by the cap. `pending` of None means the broker did not answer
+    for this lane's queues — an unknown backlog is not an empty one (snippet
+    #3969), so it contributes nothing rather than being read as zero.
 
-    ## What is current, and what is the floor
+    A cap of zero still returns one: billiard cannot run an empty pool, and
+    the parked process is what `add_consumer` lands on when the cap goes back
+    up. "Off" is expressed by cancelling consumers, not by emptying the pool.
+    """
+    if cap <= 0:
+        return MIN_POOL_SLOTS
+    return max(MIN_POOL_SLOTS, min(cap, active + (pending or 0)))
 
-    The value this moves is the LIVE pool, read from `inspect`. The stored
-    `configured_slots` is what the operator set and is only the FLOOR: growth
-    goes above it and a shrink returns to it, never below.
 
-    The two are deliberately not the same number, and reading the stored value
-    as "current" is the mistake that makes this function useless in a way no
-    unit test of a single tick would show. The autoscaler never writes the
-    row, so the stored value never moves; a tick that computed `stored + 1`
-    would propose the same target forever, cap the lane one slot above the
-    floor no matter the load, and — because resizing a replica already at the
-    target issues nothing and reports success — claim `grew` on every tick
-    while nothing changed. Lesson #4183's non-convergence, arriving with a
-    success message attached.
+def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]:
+    """Size every lane to its backlog, within the cap. The whole control loop.
 
-    So: `current = state.pool`, `configured` is the floor, and both `grew` and
-    `shrank` mean the live pool actually moved.
+    `caps` is lane name -> slots_cap, read from the database by the caller.
+    This function touches no database: the celery task that schedules it owns
+    the session, and keeping the DB out of here is what lets it be called from
+    anywhere that already knows the caps.
+
+    ## It must converge and then go quiet
+
+    One `inspect` for all lanes, and `set_lane_slots_sync` issues nothing to a
+    replica already at its target. A settled system therefore performs one
+    broker round trip plus one LLEN sweep per tick and sends no control
+    messages at all — the reachable fixed point lesson #4183 is about. An
+    enforcer that re-sent a grow of zero every tick would churn forever and
+    bury a real correction in its own noise.
+
+    ## An absent lane is SKIPPED, not corrected
+
+    `present=False` means nothing answered — a worker restarting, or an
+    unreachable broker. It does NOT mean zero slots. Deciding from that would
+    be a verdict drawn from an unswept read, and here it is worse than
+    useless: there is nothing to send the message to.
     """
     live = inspect_lanes_sync()
     depths = _queue_depths_sync()
-    out: list[AutoscaleDecision] = []
+    out: list[LaneSizing] = []
 
     for lane in LANES:
-        target = lanes.get(lane.name)
-        if target is None:
+        cap = caps.get(lane.name)
+        if cap is None:
             continue
-        cap, configured, on = target
-        if not on:
+        state = live[lane.name]
+        if not state.present:
+            out.append(LaneSizing(lane.name, "skipped", 0, "lane is not answering"))
             continue
 
-        state = live[lane.name]
-        if not state.present or state.pool is None:
-            # Nothing answered. Not "idle" — unknown, and a decision drawn
-            # from an unswept read is exactly what snippet #3969 warns about.
-            # `configured` is reported because there is no live number to
-            # report; it is what the lane will come back at.
-            out.append(AutoscaleDecision(
-                lane.name, "held", configured, "lane is not answering",
-            ))
-            continue
+        # Consumers first, and only when they DISAGREE. Sending add_consumer
+        # for every queue on every tick of a settled system is the exact churn
+        # above, and invisible: add_consumer on a queue already consumed is
+        # harmless and reports success.
+        should_consume = cap > 0
+        if should_consume != state.consuming.issuperset(lane.queues):
+            ok, err = set_lane_enabled_sync(lane, should_consume, live=state)
+            if not ok:
+                out.append(LaneSizing(
+                    lane.name, "held", state.pool or 0,
+                    f"could not {'start' if should_consume else 'stop'} "
+                    f"consuming: {err}",
+                ))
+                continue
 
         current = state.pool
-        known = [depths.get(q) for q in lane.queues]
-        if all(d is None for d in known):
-            out.append(AutoscaleDecision(
-                lane.name, "held", current, "queue depth unavailable",
+        if current is None:
+            out.append(LaneSizing(
+                lane.name, "held", 0, "worker did not report its pool size",
             ))
             continue
-        backlog = sum(d for d in known if d is not None) + state.reserved
-        # Against CAPACITY, not against the dial: `active` is summed across
-        # replicas, so comparing it to one replica's pool size would call two
-        # half-busy replicas of 4 saturated at 4 active and grow a lane that
-        # has idle slots.
-        saturated = state.active >= state.capacity > 0
-        busy = backlog >= AUTOSCALE_BACKLOG_THRESHOLD
 
-        if saturated and busy and current < cap:
-            new = min(cap, current + AUTOSCALE_STEP)
-            ok, err = set_lane_slots_sync(lane, new, live=state)
-            out.append(AutoscaleDecision(
-                lane.name, "grew" if ok else "held", new if ok else current,
-                f"{backlog} waiting and all {state.capacity} slots busy"
-                if ok else f"could not grow: {err}",
-            ))
-        elif saturated and busy:
-            # At the cap with work still waiting. Said out loud rather than
-            # held silently: this is the operator's own ceiling doing its job,
-            # and it is the moment they would want to know they set it.
-            out.append(AutoscaleDecision(
-                lane.name, "held", current,
-                f"{backlog} waiting but the cap is {cap}",
-            ))
-        elif current > effective_slots(configured) and (
-            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,
-                f"backlog cleared, back toward {configured}"
-                if ok else f"could not shrink: {err}",
-            ))
+        known = [depths.get(q) for q in lane.queues]
+        depth = (
+            sum(d for d in known if d is not None)
+            if any(d is not None for d in known) else None
+        )
+        pending = None if depth is None else depth + state.reserved
+        want = wanted_slots(cap, state.active, pending)
+
+        if want > current:
+            new = want
+            verb = "grew"
+        elif want < current:
+            # One at a time on the way down. See SHRINK_STEP.
+            new = max(want, current - SHRINK_STEP)
+            verb = "shrank"
         else:
-            # The fixed point. A settled lane sends nothing and says so —
-            # the tick is one inspect and one LLEN sweep, no control messages.
-            out.append(AutoscaleDecision(
+            out.append(LaneSizing(
                 lane.name, "held", current,
-                f"{backlog} waiting, {state.active} busy",
+                f"{pending if pending is not None else '?'} waiting, "
+                f"{state.active} busy, cap {cap}",
             ))
+            continue
+
+        ok, err = set_lane_slots_sync(lane, new, live=state)
+        if not ok:
+            out.append(LaneSizing(
+                lane.name, "held", current, f"could not resize: {err}",
+            ))
+            continue
+        out.append(LaneSizing(
+            lane.name, verb, new,
+            f"{pending if pending is not None else '?'} waiting, "
+            f"{state.active} busy, cap {cap}",
+        ))
     return out