feat: change a lane's slots on a running system, over the broker (4292)

Milestone 422 step 2. `GET /api/system/workers` reports every lane joined to
its live pool; `POST /api/system/workers/<name>` changes it.

NO DOCKER SOCKET. Milestone 365 deferred "acting on the state" because
restarting a dead worker needs a socket the web container deliberately does
not have. That holds for restarting a CONTAINER; it does not hold for
changing how much work a RUNNING worker does. celery's pool_grow /
pool_shrink / add_consumer / cancel_consumer send a message over the Redis
the app already uses, and the worker resizes itself. No new privilege, no new
surface, and the security question that deferred this is never raised.

PERSIST AND PUSH, in one call, in that order. pool_grow is not durable — a
restart drops every lane to its env concurrency — so a UI that only pushed
would lose the setting on the next deploy with nothing to show for it (lesson
#4202). Storing alone would describe nothing until something restarted. A
failed PUSH is not a failed setting: 200 with `applied: false` and a reason,
so the UI says "saved, not yet live" rather than "that didn't work". Step 3's
reconcile carries it when the lane answers again.

PER-REPLICA DELTAS. `pool_grow(n, destination=[...])` adds n to EACH
destination, so while `worker` runs `replicas: 2` a single delta from an
aggregate is wrong for both. `slots` therefore means what CELERY_CONCURRENCY
means — one process's pool — and each replica is driven to it from its OWN
current size, so replicas that drifted apart converge rather than moving in
lockstep. I wrote this wrong first: the docstring claimed per-replica while
the code computed one delta from the max across replicas. LaneLiveState now
carries `pools` per hostname and exposes `pool` as a property.

A replica already at the target is sent nothing at all — the reachable fixed
point step 3's periodic reconcile needs, or it re-issues a grow of zero every
tick forever (lesson #4183). A replica that answered inspect but not stats is
NAMED in the error rather than skipped silently, since otherwise it would run
at a size the UI claims it does not.

`present=False` is not "zero slots", it is "nothing answered" — kept distinct
throughout, because step 3 skips an absent lane rather than correcting it.

/workers now also reports pool size (from `insp.stats()`) and RESERVED count.
Celery prefetches, so tasks that have left the Redis list but not started are
invisible to LLEN: a lane can read depth 0 with thirty tasks held in worker
memory. `pending` is depth + reserved. The UI is misleading without this and
step 7's autoscaler would be simply wrong.

Also kills the THIRD copy of the queue list: system_activity's _QUEUE_NAMES,
whose own comment admitted the coupling ("must match celery_app.task_routes")
and which sat alongside task_routes and the ROLE_NAMES copy step 1 collapsed.
Now derived from LANES. The rendered order changes to lane grouping, which is
the better shape for a lane-oriented UI.

Separate blueprint rather than folding into system_activity, which states in
its first line that it is read-only and answers a different question — its
/workers is keyed on celery HOSTNAME and reports which nodes answered.

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

backend/app/api/__init__.py

@@ -41,6 +41,7 @@ def all_blueprints() -> list[Blueprint]:
     from .system_health import system_health_bp
     from .tags import tags_bp
     from .thumbnails import thumbnails_bp
+    from .workers import workers_bp
     return [
         api_bp,
         attachments_bp,
@@ -52,6 +53,7 @@ def all_blueprints() -> list[Blueprint]:
         showcase_bp,
         settings_bp,
         system_activity_bp,
+        workers_bp,
         system_health_bp,
         system_backup_bp,
         admin_bp,

backend/app/api/system_activity.py

@@ -21,18 +21,22 @@ from ..config import get_config
 from ..extensions import get_session
 from ..models import TaskRun
 from ..services.scheduler_service import scheduler_status
+from ..services.worker_lanes import LANES
 
 system_activity_bp = Blueprint(
     "system_activity", __name__, url_prefix="/api/system/activity",
 )
 
-# Canonical queue order — must match celery_app.task_routes. UI renders
-# in this order; queues with no LLEN response show as null rather than
-# absent.
-_QUEUE_NAMES = (
-    "default", "import", "thumbnail", "ml",
-    "download", "scan", "maintenance", "maintenance_long",
-)
+# Every queue, grouped by the lane that consumes it. DERIVED from
+# `worker_lanes.LANES` (milestone 422 step 1) rather than written out:
+# this was a hand-kept third copy of "which queues exist", alongside
+# celery_app.task_routes and service_roster.ROLE_NAMES, and its own comment
+# admitted the coupling — "must match celery_app.task_routes".
+#
+# The rendered ORDER changes with this: lane order rather than the previous
+# hand-chosen one. That is the better grouping for a lane-oriented UI, and
+# queues with no LLEN response still show as null rather than absent.
+_QUEUE_NAMES = tuple(q for lane in LANES for q in lane.queues)
 
 # Cache module-level so all requests share the cache between polls.
 # Tests can reset via direct dict mutation if needed.

backend/app/api/workers.py

@@ -0,0 +1,103 @@
+"""Worker lanes: what each is doing, and the dial that changes it.
+
+Milestone 422 step 2. The write half of a surface `api/system_activity.py`
+only reads.
+
+## Why this is a separate blueprint
+
+`system_activity` says in its own first line that it is read-only, and it
+answers a different question: its `/workers` is keyed on celery HOSTNAME and
+reports which nodes answered. That stays as it is — the existing
+SystemActivityTab consumes it.
+
+This is keyed on LANE, joins the stored settings to the live pool, and
+accepts writes. Two endpoints answering "which celery processes exist" and
+"how much work is each lane allowed to do" are not the same endpoint, and
+folding the second into the first would make a read-only module a write one.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from quart import Blueprint, jsonify, request
+
+from ..extensions import get_session
+from ..services.worker_control import LaneUpdateRefused, lane_view, set_lane
+from ..services.worker_lanes import LANES_BY_NAME
+from ._responses import error_response as _bad
+
+workers_bp = Blueprint("workers", __name__, url_prefix="/api/system/workers")
+
+
+@workers_bp.route("", methods=["GET"])
+async def list_lanes():
+    """Every lane: configured slots, the cap, the ceiling, and live state.
+
+    Response: {lanes: [...], fetched_at: iso8601}
+
+    Deliberately NOT cached, unlike system_activity's 2s/5s caches. This is
+    the surface an operator watches while dragging a stepper, and a cached
+    reply would show them the value from before their own change and read as
+    the control having failed.
+    """
+    async with get_session() as session:
+        lanes = await lane_view(session)
+    return jsonify({
+        "lanes": lanes,
+        "fetched_at": datetime.now(UTC).isoformat(),
+    })
+
+
+@workers_bp.route("/<name>", methods=["POST"])
+async def update_lane(name: str):
+    """Set a lane's slots, cap and/or enabled flag. Stores, then pushes live.
+
+    Partial: only the keys present are changed, so the UI's stepper can send
+    `{"slots": 3}` without restating the cap it did not touch.
+
+    Two failure kinds, deliberately different statuses:
+
+    * **400** — the value is not allowed (above the cap, above the ceiling,
+      negative). Nothing was stored. The body carries `detail`, which is the
+      sentence the UI shows; a refused control with no reason reads as a bug.
+    * **200 with `applied: false`** — the value WAS stored but could not be
+      pushed, because the lane is not currently answering. That is not an
+      error: step 3's reconcile carries it when the lane comes back, and the
+      UI should say "saved, not yet live" rather than "that didn't work".
+    """
+    lane = LANES_BY_NAME.get(name)
+    if lane is None:
+        return _bad("unknown_lane", detail=name, known=sorted(LANES_BY_NAME))
+
+    body = await request.get_json()
+    if not isinstance(body, dict):
+        return _bad("invalid_body", detail="body must be a JSON object")
+
+    fields: dict = {}
+    for key in ("slots", "slots_cap"):
+        if key in body:
+            value = body[key]
+            # Rejected rather than coerced: `True` is an int in Python, and
+            # silently reading it as 1 slot would be a control that appears to
+            # work and sets something nobody asked for.
+            if not isinstance(value, int) or isinstance(value, bool):
+                return _bad("invalid_body", detail=f"{key} must be an integer")
+            fields[key] = value
+    if "enabled" in body:
+        if not isinstance(body["enabled"], bool):
+            return _bad("invalid_body", detail="enabled must be a boolean")
+        fields["enabled"] = body["enabled"]
+
+    if not fields:
+        return _bad(
+            "invalid_body",
+            detail="give at least one of slots, slots_cap, enabled",
+        )
+
+    async with get_session() as session:
+        try:
+            result = await set_lane(session, lane, **fields)
+        except LaneUpdateRefused as exc:
+            return _bad("refused", detail=str(exc))
+    return jsonify(result)

backend/app/services/worker_control.py

@@ -47,10 +47,15 @@ autoscaler.
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from dataclasses import dataclass, field
 
-from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import WorkerLane
+from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling
 
 log = logging.getLogger(__name__)
 
@@ -74,11 +79,22 @@ class LaneLiveState:
 
     present: bool = False
     replicas: int = 0
-    # Per-process pool size. Equal across replicas unless one has drifted.
-    pool: int | None = None
     active: int = 0
     reserved: int = 0
     hostnames: list[str] = field(default_factory=list)
+    # 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
 
 
 def _lane_for_queues(queues: tuple[str, ...]) -> Lane | None:
@@ -124,7 +140,7 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]:
         # answer, which leaves pool=None — unknown, not zero.
         pool = (stats.get(hostname) or {}).get("pool", {}).get("max-concurrency")
         if isinstance(pool, int):
-            state.pool = pool if state.pool is None else max(state.pool, pool)
+            state.pools[hostname] = pool
 
     for state in out.values():
         state.hostnames.sort()
@@ -150,16 +166,22 @@ def set_lane_slots_sync(lane: Lane, target: int) -> tuple[bool, str | None]:
         live = inspect_lanes_sync()[lane.name]
         if not live.present:
             return False, "lane is not running"
-        if live.pool is None:
+        if not live.pools:
             return False, "worker did not report its pool size"
 
         control = celery_app.control
-        for hostname in live.hostnames:
-            delta = target - live.pool
+        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)
@@ -194,3 +216,176 @@ def set_lane_enabled_sync(lane: Lane, enabled: bool) -> tuple[bool, str | None]:
             "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,
+        )
+        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)
+    live = await asyncio.to_thread(inspect_lanes_sync)
+    depths = await asyncio.to_thread(_queue_depths_sync)
+
+    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,
+            "memory_bound": lane.memory_bound,
+            "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,
+        })
+    return out
+
+
+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,
+) -> 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_enabled = row.enabled if enabled is None else 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
+    await session.commit()
+
+    applied, error = True, None
+    if enabled is not None:
+        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)
+
+    return {
+        "name": lane.name,
+        "slots": row.slots,
+        "slots_cap": row.slots_cap,
+        "ceiling": ceiling,
+        "enabled": row.enabled,
+        "applied": applied,
+        "apply_error": error,
+    }

tests/test_api_workers.py

@@ -0,0 +1,172 @@
+"""/api/system/workers — the lane dial (milestone 422 step 2).
+
+Exercises the real endpoint against the real database. Only `celery inspect`
+is stubbed, and only to keep the suite fast: an unstubbed inspect blocks for
+its full 2s timeout per call with no workers to answer, which several writes
+would turn into most of the lane's runtime.
+"""
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from backend.app.models import WorkerLane
+from backend.app.services import worker_control as wc
+from backend.app.services.worker_lanes import LANES
+
+pytestmark = pytest.mark.integration
+
+
+@pytest_asyncio.fixture
+async def no_live_workers(monkeypatch):
+    """Nothing is running — which is the CI lane's actual truth, asserted
+    rather than waited for. Makes every push fail, which is the interesting
+    half: the setting must still be stored."""
+    monkeypatch.setattr(
+        wc, "inspect_lanes_sync",
+        lambda: {lane.name: wc.LaneLiveState() for lane in LANES},
+    )
+
+
+async def _lane_row(db, name: str) -> WorkerLane:
+    return (await db.execute(
+        select(WorkerLane).where(WorkerLane.name == name)
+    )).scalar_one()
+
+
+# --- reading -----------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_get_lists_every_lane_with_its_ceiling(client, no_live_workers):
+    resp = await client.get("/api/system/workers")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+
+    by_name = {lane["name"]: lane for lane in body["lanes"]}
+    assert set(by_name) == {"worker", "scheduler", "maintenance_long", "ml"}
+    for lane in body["lanes"]:
+        assert lane["ceiling"] >= 0
+        assert lane["slots"] <= lane["slots_cap"]
+        # Nothing is running, so live state must say so rather than report
+        # zeroes that read like a healthy idle lane.
+        assert lane["live"]["present"] is False
+
+
+@pytest.mark.asyncio
+async def test_ml_ships_off(client, no_live_workers):
+    """Rule 164's carve-out and the weak-hardware default in one row: enabling
+    the lane is what triggers the SigLIP download, so a fresh install must not
+    find it on."""
+    body = await (await client.get("/api/system/workers")).get_json()
+    ml = next(lane for lane in body["lanes"] if lane["name"] == "ml")
+    assert ml["enabled"] is False
+    assert ml["slots"] == 0
+
+
+# --- the persist / push split ------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_value_is_stored_even_when_it_cannot_be_pushed(
+    client, db, no_live_workers
+):
+    """THE test for this step. `pool_grow` is not durable and the lane is not
+    answering, so the push fails — and the setting must survive anyway, or a
+    UI that saved while a worker was restarting would silently lose it
+    (lesson #4202). 200 with `applied: false`, not an error.
+    """
+    resp = await client.post("/api/system/workers/worker", json={"slots": 3})
+    assert resp.status_code == 200
+    body = await resp.get_json()
+
+    assert body["slots"] == 3
+    assert body["applied"] is False
+    assert "not running" in body["apply_error"]
+
+    assert (await _lane_row(db, "worker")).slots == 3
+
+
+@pytest.mark.asyncio
+async def test_a_partial_update_leaves_the_other_fields_alone(
+    client, db, no_live_workers
+):
+    """The stepper sends `{"slots": n}` without restating a cap it did not
+    touch."""
+    before = await _lane_row(db, "worker")
+    original_cap = before.slots_cap
+
+    await client.post("/api/system/workers/worker", json={"slots": 2})
+
+    await db.refresh(before)
+    assert before.slots == 2
+    assert before.slots_cap == original_cap
+
+
+# --- what is refused ---------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_slots_above_the_cap_are_refused_and_nothing_is_stored(
+    client, db, no_live_workers
+):
+    row = await _lane_row(db, "worker")
+    resp = await client.post(
+        "/api/system/workers/worker", json={"slots": row.slots_cap + 1},
+    )
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "refused"
+    # The sentence the UI shows. A greyed control with no reason reads as a bug.
+    assert "cap" in body["detail"]
+
+    await db.refresh(row)
+    assert row.slots <= row.slots_cap
+
+
+@pytest.mark.asyncio
+async def test_a_cap_above_the_derived_ceiling_is_refused(
+    client, db, no_live_workers
+):
+    """The operator's cap is theirs to set, but not past what the container
+    can hold — the whole point of deriving a ceiling rather than typing one."""
+    resp = await client.post(
+        "/api/system/workers/ml", json={"slots_cap": 10_000},
+    )
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "refused"
+    assert "container can hold" in body["detail"]
+
+
+@pytest.mark.asyncio
+async def test_a_boolean_is_not_accepted_as_a_slot_count(client, no_live_workers):
+    """`True` is an int in Python. Coerced, it would silently set one slot —
+    a control that appears to work and does something nobody asked for."""
+    resp = await client.post("/api/system/workers/worker", json={"slots": True})
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "invalid_body"
+
+
+@pytest.mark.asyncio
+async def test_an_unknown_lane_is_refused_and_names_the_known_ones(
+    client, no_live_workers
+):
+    resp = await client.post("/api/system/workers/nonsense", json={"slots": 1})
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "unknown_lane"
+    assert "worker" in body["known"]
+
+
+@pytest.mark.asyncio
+async def test_an_empty_body_is_refused_rather_than_treated_as_a_no_op(
+    client, no_live_workers
+):
+    """A POST that changes nothing and returns 200 is indistinguishable from
+    one that worked, which is how a broken UI control goes unnoticed."""
+    resp = await client.post("/api/system/workers/worker", json={})
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "invalid_body"

tests/test_worker_control.py

@@ -0,0 +1,188 @@
+"""Changing a lane's slots on a running system (milestone 422 step 2).
+
+The celery control calls are stubbed: what is being tested is the DELTA
+ARITHMETIC and the persist/push split, not that celery can resize its own
+pool. A test that asserted celery's behaviour would be testing celery.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from backend.app.services import worker_control as wc
+from backend.app.services.worker_lanes import LANES_BY_NAME
+
+# --- what inspect reports ----------------------------------------------------
+
+
+class _Control:
+    """Records the control messages that were sent."""
+
+    def __init__(self):
+        self.grew: list[tuple[int, list[str]]] = []
+        self.shrank: list[tuple[int, list[str]]] = []
+        self.added: list[tuple[str, list[str]]] = []
+        self.cancelled: list[tuple[str, list[str]]] = []
+
+    def pool_grow(self, n, destination=None):
+        self.grew.append((n, destination))
+
+    def pool_shrink(self, n, destination=None):
+        self.shrank.append((n, destination))
+
+    def add_consumer(self, queue, destination=None):
+        self.added.append((queue, destination))
+
+    def cancel_consumer(self, queue, destination=None):
+        self.cancelled.append((queue, destination))
+
+
+def _stub_live(monkeypatch, lane_name, *, pools, present=True, reserved=0):
+    state = wc.LaneLiveState(
+        present=present,
+        replicas=len(pools),
+        hostnames=sorted(pools),
+        pools=dict(pools),
+        reserved=reserved,
+    )
+    monkeypatch.setattr(
+        wc, "inspect_lanes_sync",
+        lambda: {name: (state if name == lane_name else wc.LaneLiveState())
+                 for name in LANES_BY_NAME},
+    )
+    return state
+
+
+def _stub_control(monkeypatch):
+    control = _Control()
+
+    class _Celery:
+        pass
+
+    celery = _Celery()
+    celery.control = control
+    import sys
+    import types
+    mod = types.ModuleType("backend.app.celery_app")
+    mod.celery = celery
+    monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod)
+    return control
+
+
+def test_pool_property_is_max_not_sum():
+    """`slots` means the pool size of ONE process, so the aggregate shown to
+    the operator is the largest replica — not the total. A sum would report 8
+    for two replicas of 4 and invite them to 'reduce it to 4', which would
+    halve the lane."""
+    state = wc.LaneLiveState(pools={"a": 4, "b": 4})
+    assert state.pool == 4
+
+
+def test_pool_is_none_when_nothing_reported():
+    """Unknown, never zero — the distinction step 3's reconcile depends on."""
+    assert wc.LaneLiveState(present=True).pool is None
+
+
+# --- the delta arithmetic ----------------------------------------------------
+
+
+def test_each_replica_gets_its_own_delta(monkeypatch):
+    """The bug this exists to prevent: one delta computed from an aggregate
+    and applied to every replica. With replicas at 2 and 6 and a target of 4,
+    a shared delta moves both the same way and leaves them at 4 and 8 — or 0
+    and 4 — depending on which aggregate was used. Per-replica deltas
+    converge both on 4.
+    """
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "worker", pools={"host-a": 2, "host-b": 6})
+
+    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)
+
+    assert (applied, err) == (True, None)
+    assert control.grew == [(2, ["host-a"])]
+    assert control.shrank == [(2, ["host-b"])]
+
+
+def test_a_replica_already_at_the_target_is_sent_nothing(monkeypatch):
+    """The fixed point step 3's reconcile needs. An enforcer that re-issues a
+    grow of zero every tick never converges and re-does its own work forever
+    (lesson #4183)."""
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "worker", pools={"host-a": 4})
+
+    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)
+
+    assert (applied, err) == (True, None)
+    assert control.grew == []
+    assert control.shrank == []
+
+
+def test_resizing_an_absent_lane_reports_rather_than_raises(monkeypatch):
+    _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "worker", pools={}, present=False)
+
+    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)
+
+    assert applied is False
+    assert "not running" in err
+
+
+def test_a_replica_with_no_reported_pool_is_named_not_skipped_silently(
+    monkeypatch,
+):
+    """Resize what can be resized, then say which could not. Silence would
+    leave a replica running at a size the UI claims it is not."""
+    control = _stub_control(monkeypatch)
+    state = _stub_live(monkeypatch, "worker", pools={"host-a": 2})
+    state.hostnames = ["host-a", "host-b"]  # b answered inspect, not stats
+    state.replicas = 2
+
+    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)
+
+    assert control.grew == [(2, ["host-a"])]
+    assert applied is False
+    assert "host-b" in err
+
+
+# --- enabling and disabling --------------------------------------------------
+
+
+def test_disabling_cancels_consumers_rather_than_killing_the_worker(monkeypatch):
+    """A cancelled consumer keeps the process alive and answering inspect, so
+    a disabled lane stays visible. A killed worker reads as ABSENT, which is
+    the same signal as a crash — and milestone 365 exists precisely so those
+    two do not look alike."""
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "ml", pools={"host-a": 1})
+
+    applied, err = wc.set_lane_enabled_sync(LANES_BY_NAME["ml"], False)
+
+    assert (applied, err) == (True, None)
+    assert control.cancelled == [("ml", ["host-a"])]
+    assert control.added == []
+
+
+def test_enabling_adds_a_consumer_for_every_queue_in_the_lane(monkeypatch):
+    control = _stub_control(monkeypatch)
+    _stub_live(monkeypatch, "worker", pools={"host-a": 1})
+
+    wc.set_lane_enabled_sync(LANES_BY_NAME["worker"], True)
+
+    assert [q for q, _ in control.added] == list(LANES_BY_NAME["worker"].queues)
+
+
+# --- lane identity -----------------------------------------------------------
+
+
+def test_queue_sets_map_back_to_their_lane_in_any_order():
+    """celery does not guarantee the order it lists a worker's queues in, so
+    the lookup sorts. Unsorted, a lane would intermittently fail to match and
+    read as absent."""
+    lane = LANES_BY_NAME["worker"]
+    assert wc._lane_for_queues(tuple(reversed(lane.queues))) is lane
+
+
+def test_an_unknown_queue_set_maps_to_no_lane():
+    """A deployment slicing CELERY_QUEUES differently has no lane row to
+    control. Honest rather than an error — the roster still reports it."""
+    assert wc._lane_for_queues(("something", "else")) is None