fix: the cap dial waited out a broker round trip it did not need (4295)

Operator: "when the number is changed the change should be queued so that
it isn't blocking of the webui or the system itself. we shouldn't have to
wait for the validation live."

Two waits, and 5b6f2ba removed neither — it stopped a Postgres connection
being HELD across them, which is what had been stalling the whole site,
and left the press itself as slow as it was.

1. The store refetched after every write. GET /api/system/workers runs a
   celery inspect on an eleven-second budget, so the stepper stayed
   disabled through a round trip the press did not need. It now patches
   the row from the reply — cap, ceiling, enabled, the three fields that
   reply actually decides — and lets the 15s poll bring the live columns,
   which are measurements it must not invent.

2. The endpoint pushed to the broker before answering. Turning a lane off
   is four cancel_consumer messages; lowering a cap reads the live pool
   first. Now it stores the cap, answers `queued`, and hands the push to a
   Quart background task. Raising a cap was already free and stays free.

Nothing is lost by not waiting: the stored cap is what the system obeys
and the sizing pass re-reads it every minute. That sweep was already the
backstop for a push that failed, which under `no_live_workers` is every
push in the suite.

Also closes a hole the move exposed: the model fetch was gated on the
consumer change having landed, so raising ML off zero while the lane was
restarting stored the cap, let the sizing pass start the consumers a
minute later, and left the lane running with no model — nothing else ever
asks for one. It now fires on the transition and waits in the ml queue.

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

tests/test_api_workers.py

@@ -10,6 +10,7 @@ import pytest
 import pytest_asyncio
 from sqlalchemy import select
 
+from backend.app.api import workers as workers_api
 from backend.app.models import WorkerLane
 from backend.app.services import worker_control as wc
 from backend.app.services.worker_lanes import LANES
@@ -28,6 +29,39 @@ async def no_live_workers(monkeypatch):
     )
 
 
+@pytest.fixture(autouse=True)
+def queued_pushes(monkeypatch):
+    """Hold the background push still, and hand the test the hand-off.
+
+    Since 2026-09-23 the endpoint stores the cap, answers, and gives the live
+    push to a Quart background task — operator: *"the change should be queued
+    so that it isn't blocking of the webui or the system itself."* A task that
+    outlived a test's monkeypatches would reach the real broker during
+    teardown and wait out its timeout there, so every test in this module
+    captures the hand-off instead, and the ones that care about what the push
+    DOES run it deliberately with `_run_pushes`.
+
+    Autouse, because a test that forgets is not a test that fails — it is a
+    test that leaks a 2s broker call into whichever test runs next.
+    """
+    scheduled: list[tuple] = []
+    monkeypatch.setattr(
+        workers_api, "_schedule_push",
+        lambda lane, slots_cap, was_cap: scheduled.append((lane, slots_cap, was_cap)),
+    )
+    return scheduled
+
+
+async def _run_pushes(scheduled: list[tuple]) -> list[dict]:
+    """Run what the endpoint queued, in order, and clear the queue."""
+    out = [
+        await wc.push_lane_cap(lane, cap, was_cap=was_cap)
+        for lane, cap, was_cap in scheduled
+    ]
+    scheduled.clear()
+    return out
+
+
 async def _lane_row(db, name: str) -> WorkerLane:
     return (await db.execute(
         select(WorkerLane).where(WorkerLane.name == name)
@@ -90,11 +124,11 @@ async def test_the_other_lanes_ship_at_one(client, no_live_workers):
 
 @pytest.mark.asyncio
 async def test_raising_a_cap_stores_it_and_pushes_nothing(
-    client, db, no_live_workers,
+    client, db, no_live_workers, queued_pushes,
 ):
     """A cap is PERMISSION, not a request. Raising it must not grow the pool
-    here — that would put workers on a lane with nothing to do — so there is
-    nothing to push and `applied` is vacuously true.
+    — that would put workers on a lane with nothing to do — so the queued
+    push has nothing to say to the broker.
 
     This asserted `applied is False` until run 7367, carried over from when
     the number meant "run this many". The code was right and the test was
@@ -105,32 +139,33 @@ async def test_raising_a_cap_stores_it_and_pushes_nothing(
     assert resp.status_code == 200
     body = await resp.get_json()
     assert body["slots_cap"] == 3
-    assert body["applied"] is True
     assert (await _lane_row(db, "worker")).slots_cap == 3
+    assert [r["applied"] for r in await _run_pushes(queued_pushes)] == [True]
 
 
 @pytest.mark.asyncio
 async def test_turning_a_lane_off_is_stored_even_when_it_cannot_be_pushed(
-    client, db, no_live_workers,
+    client, db, no_live_workers, queued_pushes,
 ):
     """The direction that DOES push. Consumers follow the cap immediately in
     both directions — off must take effect when it is asked for — so with
     nothing answering, the push fails.
 
-    That is NOT a failed setting: the value is saved and the sizing pass
-    carries it within a minute (lesson #4202 — a live change that does not
-    survive, with nothing saying so). The UI says "saved, not yet live"
-    rather than "that didn't work", which is the distinction `applied`
-    exists to carry.
+    That is NOT a failed setting, and it is why the reply does not wait for
+    it: the value is saved and the sizing pass carries it within a minute
+    (lesson #4202 — a live change that does not survive, with nothing saying
+    so). The push reports the failure to the log, where an operator can find
+    it; the table shows the lane not answering either way.
     """
     resp = await client.post("/api/system/workers/worker", json={"slots_cap": 0})
 
     assert resp.status_code == 200
-    body = await resp.get_json()
-    assert body["applied"] is False
-    assert "not running" in body["apply_error"]
     assert (await _lane_row(db, "worker")).slots_cap == 0
 
+    pushed = await _run_pushes(queued_pushes)
+    assert pushed[0]["applied"] is False
+    assert "not running" in pushed[0]["apply_error"]
+
 
 # --- the cap is the switch ---------------------------------------------------
 
@@ -158,7 +193,7 @@ async def test_raising_it_off_zero_turns_the_lane_on(client, db, no_live_workers
 
 @pytest.mark.asyncio
 async def test_the_model_fetch_fires_on_the_transition_not_on_every_write(
-    client, db, no_live_workers, monkeypatch,
+    client, db, no_live_workers, queued_pushes, monkeypatch,
 ):
     """Raising the cap off zero downloads SigLIP, once. A second nudge of the
     same dial must not re-enqueue a multi-GB download — and the trigger must
@@ -176,15 +211,42 @@ async def test_the_model_fetch_fires_on_the_transition_not_on_every_write(
     first = await (await client.post(
         "/api/system/workers/ml", json={"slots_cap": 1},
     )).get_json()
+    await _run_pushes(queued_pushes)
     second = await (await client.post(
         "/api/system/workers/ml", json={"slots_cap": 2},
     )).get_json()
+    await _run_pushes(queued_pushes)
 
+    # The reply promises it, the queued push performs it. Both halves are
+    # checked, because the reply is what the UI says out loud.
     assert first["fetching_models"] is True
     assert second["fetching_models"] is False
     assert fired == [1]
 
 
+@pytest.mark.asyncio
+async def test_the_model_fetch_is_enqueued_even_if_the_lane_is_not_answering(
+    client, no_live_workers, queued_pushes, monkeypatch,
+):
+    """Turning ML on while it is restarting must still fetch the model.
+
+    It was gated on the consumer change having landed until 2026-09-23, on
+    the reasoning that a task enqueued onto a queue nothing consumes just sits
+    there. It does — and that is the right place for it to wait. Gated, this
+    path stored the cap, let the sizing pass start the consumers a minute
+    later, and left the lane running with no model, because nothing else ever
+    asks for one.
+    """
+    fired = []
+    monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True)
+
+    await client.post("/api/system/workers/ml", json={"slots_cap": 1})
+    pushed = await _run_pushes(queued_pushes)
+
+    assert pushed[0]["applied"] is False, "the fixture must leave the push failing"
+    assert fired == [1]
+
+
 # --- what is refused ---------------------------------------------------------
 
 
@@ -306,14 +368,12 @@ async def test_the_cap_write_holds_no_session_while_it_pushes():
 
 @pytest.mark.asyncio
 async def test_raising_a_cap_costs_no_broker_round_trip_at_all(
-    client, db, no_live_workers, monkeypatch,
+    client, db, no_live_workers, queued_pushes, monkeypatch,
 ):
-    """The common case must be instant. Raising a cap is permission, not a
-    request — the sizing pass spends it — so there is nothing to tell the
-    broker, and the operator's `+` should answer immediately rather than
-    waiting out an inspect."""
-    from backend.app.services import worker_control as wc
-
+    """Raising a cap is permission, not a request — the sizing pass spends it
+    — so there is nothing to tell the broker AT ALL. Asserted on the queued
+    push rather than on the request, because the request no longer waits for
+    it either way and would pass this vacuously."""
     calls = []
     monkeypatch.setattr(
         wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {},
@@ -328,8 +388,48 @@ async def test_raising_a_cap_costs_no_broker_round_trip_at_all(
     )
 
     await client.post("/api/system/workers/worker", json={"slots_cap": 1})
+    await _run_pushes(queued_pushes)
     calls.clear()
     resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6})
+    await _run_pushes(queued_pushes)
 
     assert resp.status_code == 200
     assert calls == [], f"raising a cap talked to the broker: {calls}"
+
+
+@pytest.mark.asyncio
+async def test_the_reply_never_waits_for_the_broker_in_any_direction(
+    client, db, no_live_workers, queued_pushes, monkeypatch,
+):
+    """Operator, 2026-09-23: *"when the number is changed the change should be
+    queued so that it isn't blocking of the webui or the system itself. we
+    shouldn't have to wait for the validation live."*
+
+    Raising a cap was already free. The rest were not: turning a lane off is
+    four `cancel_consumer` messages, and lowering a cap reads the live pool
+    first — an `inspect` on an eleven-second budget — all of it between the
+    click and the response, with the stepper disabled throughout.
+
+    Asserted by making any broker call from the request path RAISE. A timing
+    assertion would be flaky, and counting calls afterwards would pass against
+    a version that made them and was merely quick about it.
+    """
+    def boom(*args, **kwargs):
+        raise AssertionError("the request path talked to the broker")
+
+    monkeypatch.setattr(wc, "inspect_lanes_sync", boom)
+    monkeypatch.setattr(wc, "set_lane_slots_sync", boom)
+    monkeypatch.setattr(wc, "set_lane_enabled_sync", boom)
+
+    # Every direction: on, up, down, off.
+    for cap in (1, 4, 2, 0):
+        resp = await client.post(
+            "/api/system/workers/worker", json={"slots_cap": cap},
+        )
+        assert resp.status_code == 200, cap
+        assert (await resp.get_json())["queued"] is True, cap
+        assert (await _lane_row(db, "worker")).slots_cap == cap
+        # The work was handed off rather than skipped — a control that answers
+        # instantly by doing nothing is the failure this could become.
+        assert queued_pushes, f"cap {cap} queued no push"
+        queued_pushes.clear()