FabledCurator/backend/app/services/worker_control.py

"""Read and change a lane's live pool, over the broker.

Milestone 422 step 2. The half of the milestone that does something.

## No docker socket is involved, and that is the point

Milestone 365 put "acting on the state" out of scope because restarting a
dead worker needs a docker socket the web container deliberately does not
have. That is true of RESTARTING a container. It is not true of changing how
much work a RUNNING worker does: celery's remote control sends a message over
the broker and the worker resizes its own pool. Same Redis the app already
uses, no new privilege, no new surface.

    pool_grow / pool_shrink   how many slots a lane runs
    add_consumer / cancel_consumer   whether it consumes its queues at all

The operator ruled the socket out independently (2026-09-22: *"this feature
is a very invasive idea in my mind and I'd like to avoid it"*), and nothing
here raises the question.

## The setting is PER PROCESS, not per lane total

`pool_grow(n, destination=[...])` adds n slots to EACH destination it names.
While the stack still runs several containers per lane — the operator's
production `worker` is `replicas: 2` — a single delta applied to a lane's
total would be wrong for every replica.

So `slots` means what `CELERY_CONCURRENCY` means: the pool size of one
process. The reconcile below drives EACH replica to that number
independently, computing its own delta from that replica's current pool, so
replicas that have drifted apart (one restarted, one was grown) converge
rather than being moved in lockstep from a shared baseline.

After step 5 there is one process per lane and the distinction disappears.
It matters now, and getting it wrong now would be invisible — the totals
would simply be double what the UI claimed.

## Why reserved() is read alongside the queue depth

Celery PREFETCHES: a worker pulls more messages than it can run and holds
them in memory. Those have already left the Redis list, so `LLEN` — which is
what `/api/system/activity/queues` reports — can read 0 while thirty tasks
are waiting inside a worker. Any judgement about backlog that uses only LLEN
under-reports, which matters for the UI and is disqualifying for step 7's
autoscaler.
"""

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime

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

log = logging.getLogger(__name__)

# celery control is a broker round trip on a request path, so it gets a
# deadline (rule 156) — the same reasoning and the same budget as
# service_roster's inspect. A broker that stopped answering must make this
# 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
# as soon as the named nodes answer; each can still cost a full timeout if a
# node disappears mid-read, so the bound stays four.
CONTROL_ROUND_TRIPS = 4

# Slack for the `asyncio.to_thread` handoff. A budget equal to the work is a
# budget that fails under load — the roster carried exactly that bug into the
# operator's first consolidated deploy and logged a TimeoutError per refresh
# while the inspect calls underneath were working fine.
CONTROL_SLACK_SECONDS = 3.0

INSPECT_BUDGET_SECONDS = (
    CONTROL_TIMEOUT_SECONDS * CONTROL_ROUND_TRIPS + CONTROL_SLACK_SECONDS
)


@dataclass
class LaneLiveState:
    """What `celery inspect` says about one lane right now.

    `present=False` is NOT "zero slots" — it is "nothing answered". A lane
    whose worker is restarting, or whose broker is unreachable, must read as
    unknown rather than as stopped: an unswept absence is not a verdict
    (snippet #3969). The reconcile in step 3 skips an absent lane rather than
    correcting it, which is only safe because this distinction is kept.
    """

    present: bool = False
    replicas: int = 0
    active: int = 0
    reserved: int = 0
    hostnames: list[str] = field(default_factory=list)
    # The queues this lane is actually consuming right now, across replicas.
    # Distinct from the lane's CONFIGURED queues: `cancel_consumer` stops a
    # worker consuming one without changing what it was started with, which
    # is how `enabled=false` is implemented. The reconcile needs this to tell
    # "already disabled" from "needs disabling" — without it, it would re-send
    # add_consumer for every queue on every tick forever (lesson #4183).
    consuming: set[str] = field(default_factory=set)
    # Pool size PER HOSTNAME, not aggregated. The resize below computes each
    # replica's own delta from its own current pool, so replicas that have
    # drifted apart converge instead of being moved in lockstep from a shared
    # baseline — which is what an aggregate here would silently reintroduce.
    pools: dict[str, int] = field(default_factory=dict)

    @property
    def pool(self) -> int | None:
        """One number for the UI. `max` rather than a sum: `slots` means the
        pool size of ONE process (see the module docstring), so the largest
        replica is the honest answer to "what is this lane set to". None when
        no replica reported — unknown, never zero."""
        return max(self.pools.values()) if self.pools else None

    @property
    def capacity(self) -> int:
        """Total slots across replicas — how many tasks this lane can run at
        once. Distinct from `pool`, and the two must not be confused: `pool`
        is the DIAL (one process's size, what grow/shrink move), `capacity` is
        the CAPABILITY. Asking "is this lane saturated" compares `active`,
        which is summed across replicas, against this — against `pool` it
        would call two half-busy replicas of 4 saturated at 4 active."""
        return sum(self.pools.values())


def _lane_for_queues(queues: tuple[str, ...]) -> Lane | None:
    return LANES_BY_QUEUE_KEY.get(tuple(sorted(queues)))


def inspect_lanes_sync() -> dict[str, LaneLiveState]:
    """Live state per lane name. Sync — callers wrap in asyncio.to_thread.

    Never raises. Every lane is present in the result; ones nothing answered
    for carry `present=False`, so a caller cannot accidentally read a missing
    lane as an empty one by iterating only what came back.
    """
    out = {lane.name: LaneLiveState() for lane in LANES}
    try:
        from ..celery_app import celery as celery_app

        # ONE broadcast, then three TARGETED reads.
        #
        # A broadcast with no `destination` cannot know how many replies to
        # expect, so it waits out its whole timeout rather than returning on
        # the last one. Four of those is four full timeouts — about eight
        # seconds — and `lane_view` sits on the Settings card, so that was the
        # load time of the Worker lanes page every time it was opened.
        #
        # Naming the destinations lets celery stop as soon as those nodes have
        # answered, which for workers in this same container is milliseconds.
        # The worst case is unchanged: a node that vanishes between the
        # broadcast and the targeted reads costs a full timeout waiting for a
        # reply that is not coming.
        insp = celery_app.control.inspect(timeout=CONTROL_TIMEOUT_SECONDS)
        active_queues = insp.active_queues() or {}

        # Nothing answered — and the three reads below exist only to describe
        # what did. Returning here also makes the broker-down case FAST
        # (one timeout, not four), which is exactly when the healthcheck and
        # the card need an answer rather than a long wait.
        if not active_queues:
            return out

        targeted = celery_app.control.inspect(
            destination=sorted(active_queues),
            timeout=CONTROL_TIMEOUT_SECONDS,
        )
        stats = targeted.stats() or {}
        active = targeted.active() or {}
        reserved = targeted.reserved() or {}
    except Exception:
        log.warning("worker_control: celery inspect failed", exc_info=True)
        return out

    for hostname, queues in active_queues.items():
        lane = _lane_for_queues(tuple(q["name"] for q in queues))
        if lane is None:
            # A deployment slicing CELERY_QUEUES differently. Reported by the
            # roster under its raw queue list; it simply has no lane row to
            # control, which is honest rather than an error.
            continue
        state = out[lane.name]
        state.present = True
        state.replicas += 1
        state.hostnames.append(hostname)
        state.active += len(active.get(hostname, []))
        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):
            state.pools[hostname] = pool

    for state in out.values():
        state.hostnames.sort()
    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]:
    """Drive every replica of `lane` to `target` slots. Returns (applied, err).

    Per-replica deltas rather than one shared delta: see the module docstring.
    A replica already at the target is issued nothing at all, which is what
    makes step 3's periodic reconcile converge instead of re-sending a grow of
    zero forever (lesson #4183 — an enforcer without a reachable fixed point
    re-does its own work every tick).

    `applied=False` is not a failure of the SETTING. The caller has already
    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

        if live is None:
            live = inspect_lanes_sync()[lane.name]
        if not live.present:
            return False, "lane is not running"
        if not live.pools:
            return False, "worker did not report its pool size"

        control = celery_app.control
        unreported = [h for h in live.hostnames if h not in live.pools]
        for hostname, current in live.pools.items():
            delta = target - current
            if delta > 0:
                control.pool_grow(delta, destination=[hostname])
            elif delta < 0:
                control.pool_shrink(-delta, destination=[hostname])
        if unreported:
            # Resized what could be resized, and said which could not. Silence
            # here would leave a replica running at a size the UI claims it is
            # not, with nothing anywhere recording the gap.
            return False, f"no pool size reported by {', '.join(sorted(unreported))}"
        return True, None
    except Exception as exc:  # noqa: BLE001 — reported, never raised at a caller
        log.warning("worker_control: could not resize %s", lane.name, exc_info=True)
        return False, str(exc)


def set_lane_enabled_sync(
    lane: Lane, enabled: bool, live: LaneLiveState | None = None,
) -> tuple[bool, str | None]:
    """Start or stop `lane` consuming its queues, without killing the process.

    `cancel_consumer` rather than a shutdown: a stopped consumer keeps its
    worker alive and answering `inspect`, so a disabled lane stays visible and
    can be turned back on. A killed worker would read as absent, which is the
    same signal as a crash — and the whole point of the roster (#365) is that
    those two must not look alike.
    """
    try:
        from ..celery_app import celery as celery_app

        if live is None:
            live = inspect_lanes_sync()[lane.name]
        if not live.present:
            return False, "lane is not running"
        control = celery_app.control
        for queue in lane.queues:
            if enabled:
                control.add_consumer(queue, destination=live.hostnames)
            else:
                control.cancel_consumer(queue, destination=live.hostnames)
        return True, None
    except Exception as exc:  # noqa: BLE001
        log.warning(
            "worker_control: could not %s %s",
            "enable" if enabled else "disable", lane.name, exc_info=True,
        )
        return False, str(exc)


# --- the settings half, which is async ----------------------------------------
#
# Sync celery control above, async DB below, in one module. Same split
# `service_roster` already runs (`_inspect_celery_sync` beside `touch_service`)
# — the boundary is the transport, not the concern, and "control the workers"
# is one concern.


async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]:
    """Every lane's row, creating any that are missing from its LANES defaults.

    Self-heals rather than depending on a migration having run for a lane
    added later: alembic 0103 seeded the four that existed on 2026-09-22, and
    a fifth added to LANES afterwards gets its row the first time anything
    asks. Without this, a new lane would read as absent and the UI would
    simply not show it.
    """
    rows = {
        row.name: row
        for row in (await session.execute(select(WorkerLane))).scalars()
    }
    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,
        )
        session.add(row)
        rows[lane.name] = row
    if missing:
        await session.commit()
    return rows


async def lane_view(session: AsyncSession) -> list[dict]:
    """Every lane: what is configured, what is live, what it may grow to.

    One call rather than making the UI join three sources. `pending` is the
    honest backlog — Redis depth PLUS reserved — because celery prefetches and
    LLEN alone reads 0 while a worker holds tasks in memory.
    """
    rows = await _rows_by_name(session)
    # A deadline, because this is a request path and `to_thread` on its own is
    # an await with no bound (rule 156). `inspect_lanes_sync` never raises and
    # every inner call has its own timeout, so the only way past the budget is
    # the thread not being scheduled — and a page that renders "not answering"
    # is a better answer than one that does not render.
    try:
        live = await asyncio.wait_for(
            asyncio.to_thread(inspect_lanes_sync),
            timeout=INSPECT_BUDGET_SECONDS,
        )
    except TimeoutError:
        log.warning(
            "worker_control: inspect exceeded %ss; reporting every lane as "
            "not answering", INSPECT_BUDGET_SECONDS,
        )
        live = {lane.name: LaneLiveState() for lane in LANES}
    depths = await asyncio.to_thread(_queue_depths_sync)
    oldest = await _oldest_running_by_queue(session)

    now = datetime.now(UTC)
    out = []
    for lane in LANES:
        row = rows[lane.name]
        state = live[lane.name]
        # None for a queue the broker did not answer for, which must not be
        # silently summed as zero — an unknown depth is not an empty one.
        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
        out.append({
            "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,
            "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.
            "models": [
                {
                    "repo": m.repo,
                    "download_bytes": m.approx_download_bytes,
                    "resident_bytes": m.approx_resident_bytes,
                    "measured": m.measured,
                }
                for m in lane.models
            ],
            "live": {
                "present": state.present,
                "replicas": state.replicas,
                "pool": state.pool,
                "active": state.active,
                "reserved": state.reserved,
            },
            "queue_depth": depth,
            "pending": None if depth is None else depth + state.reserved,
            # How long the oldest still-running task on this lane has been
            # going, in minutes. The operator asked for a trigger here — grow
            # a lane whose tasks run past some duration — and it stayed a
            # REPORT: a long task does not finish sooner because the lane
            # gained a slot, so scaling on it would spend memory to change
            # nothing. Shown so they can see a lane wedged on one slow job,
            # which is the genuinely useful half of the idea.
            "oldest_running_minutes": _minutes_since(
                min(
                    (oldest[q] for q in lane.queues if q in oldest),
                    default=None,
                ),
                now,
            ),
        })
    return out


async def _oldest_running_by_queue(session: AsyncSession) -> dict[str, datetime]:
    """When the longest-running unfinished task on each queue started.

    Read from `task_run`, which is OUR OWN table on OUR OWN wall clock, and
    deliberately not from celery's `inspect active()`. Those entries carry a
    `time_start` taken from the WORKER's `time.monotonic()` — a clock with an
    arbitrary origin per process. Subtracting it from this process's wall
    clock produces a number that looks like a duration and is meaningless, and
    it would be meaningless in the direction that matters: plausible.

    `task_run` also already carries the per-queue staleness thresholds the
    recovery sweep uses, so a row still `running` here is one the system
    itself considers legitimately in flight rather than abandoned.
    """
    result = await session.execute(
        select(TaskRun.queue, func.min(TaskRun.started_at))
        .where(TaskRun.status == "running", TaskRun.finished_at.is_(None))
        .group_by(TaskRun.queue)
    )
    return {queue: started for queue, started in result if started is not None}


def _minutes_since(started: datetime | None, now: datetime) -> int | None:
    """Whole minutes, or None when nothing is running. Never negative: a row
    written by a container whose clock is a few seconds ahead must read as 0
    rather than as a task that starts in the future."""
    if started is None:
        return None
    return max(0, int((now - started).total_seconds() // 60))


def _queue_depths_sync() -> dict[str, int | None]:
    """Redis LLEN per queue. None for one that did not answer — see lane_view.

    Sync; the caller threads it. A per-queue try/except so one bad queue does
    not cost the whole report, matching `api/system_activity._read_queues_sync`.
    """
    import redis

    from ..config import get_config

    out: dict[str, int | None] = {}
    try:
        client = redis.Redis.from_url(get_config().celery_broker_url)
    except Exception:
        log.warning("worker_control: no broker for queue depths", exc_info=True)
        return {q: None for lane in LANES for q in lane.queues}
    for lane in LANES:
        for queue in lane.queues:
            try:
                out[queue] = int(client.llen(queue))
            except Exception:  # noqa: BLE001 — a hiccup must not break the UI
                out[queue] = None
    return out


class LaneUpdateRefused(ValueError):
    """A requested value is outside what the lane may hold. Carries the reason
    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.

    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.

    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".
    """
    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:
        raise LaneUpdateRefused(
            f"cap {new_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
    await session.commit()

    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:
        applied, error = await asyncio.to_thread(
            set_lane_enabled_sync, lane, new_enabled,
        )
    if applied and slots is not None:
        applied, error = await asyncio.to_thread(set_lane_slots_sync, lane, new_slots)

    # Enabling a lane that needs models is what triggers the fetch (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.
    fetching = False
    if new_enabled and not was_enabled 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,
        "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.
        "fetching_models": fetching,
    }


def _enqueue_model_fetch() -> bool:
    """Queue the model download. Returns whether it was accepted.

    Import inside the function: `backend.app.tasks.ml` pulls in torch, and web
    must not pay that import cost on a module that every settings request
    touches.

    Never raises. A broker that will not take the task is worth reporting, but
    the SETTING has already been stored and the lane is already enabled — so
    failing the whole request here would roll back nothing and tell the
    operator their change did not happen when it did.
    """
    try:
        from ..tasks.ml import ensure_models

        ensure_models.delay()
        return True
    except Exception:  # noqa: BLE001 — reported, never raised at a caller
        log.warning("worker_control: could not enqueue the model fetch", exc_info=True)
        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 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
#
# 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.
#
# 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

# 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


@dataclass
class AutoscaleDecision:
    """What the autoscaler 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.
    """

    lane: str
    action: str  # "grew" | "shrank" | "held"
    slots: int
    reason: str


def autoscale_lanes_sync(
    lanes: dict[str, tuple[int, int, bool]],
) -> list[AutoscaleDecision]:
    """Decide and apply one round of autoscaling.

    `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.

    ## What is current, and what is the floor

    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.

    So: `current = state.pool`, `configured` is the floor, and both `grew` and
    `shrank` mean the live pool actually moved.
    """
    live = inspect_lanes_sync()
    depths = _queue_depths_sync()
    out: list[AutoscaleDecision] = []

    for lane in LANES:
        target = lanes.get(lane.name)
        if target is None:
            continue
        cap, configured, on = target
        if not on:
            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

        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",
            ))
            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}",
            ))
        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(
                lane.name, "held", current,
                f"{backlog} waiting, {state.active} busy",
            ))
    return out