bvandeusenandClaude Opus 5
445164c852
CI and images / lint (push) Successful in 4s
CI and images / extension-version (push) Successful in 4s
CI and images / frontend-build (push) Successful in 24s
CI and images / integration (push) Failing after 24s
CI and images / backend-lint-and-test (push) Failing after 34s
CI and images / sign-extension (push) Skipped
CI and images / build-web (push) Skipped
CI and images / smoke-web (push) Skipped
CI and images / promote (push) Skipped
CI and images / build-agent (push) Skipped
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
233 lines
8.4 KiB
Python
233 lines
8.4 KiB
Python
"""/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_cap"] <= lane["ceiling"]
|
|
# 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_enabled_is_derived_from_the_cap_and_never_stored(
|
|
client, no_live_workers,
|
|
):
|
|
"""The reshape of 2026-09-23. "Off" and "may use no workers" were two
|
|
spellings of one fact, stored separately and free to disagree."""
|
|
body = await (await client.get("/api/system/workers")).get_json()
|
|
for lane in body["lanes"]:
|
|
assert lane["enabled"] == (lane["slots_cap"] > 0), lane["name"]
|
|
|
|
|
|
@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: raising
|
|
the cap is what triggers the SigLIP download, so a fresh install must not
|
|
find it above zero."""
|
|
body = await (await client.get("/api/system/workers")).get_json()
|
|
ml = next(lane for lane in body["lanes"] if lane["name"] == "ml")
|
|
assert ml["slots_cap"] == 0
|
|
assert ml["enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_other_lanes_ship_at_one(client, no_live_workers):
|
|
"""Operator: *"the cap defaults should be 1 and 0 for the ml-worker."*"""
|
|
body = await (await client.get("/api/system/workers")).get_json()
|
|
caps = {lane["name"]: lane["slots_cap"] for lane in body["lanes"]}
|
|
assert caps == {
|
|
"worker": 1, "scheduler": 1, "maintenance_long": 1, "ml": 0,
|
|
}
|
|
|
|
|
|
# --- the persist / push split ------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_cap_is_stored_even_when_it_cannot_be_pushed(
|
|
client, db, no_live_workers,
|
|
):
|
|
"""Nothing is answering, so the live 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)."""
|
|
resp = await client.post("/api/system/workers/worker", json={"slots_cap": 3})
|
|
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["slots_cap"] == 3
|
|
assert body["applied"] is False
|
|
assert "not running" in body["apply_error"]
|
|
|
|
assert (await _lane_row(db, "worker")).slots_cap == 3
|
|
|
|
|
|
# --- the cap is the switch ---------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_cap_of_zero_turns_the_lane_off(client, db, no_live_workers):
|
|
await client.post("/api/system/workers/worker", json={"slots_cap": 0})
|
|
|
|
row = await _lane_row(db, "worker")
|
|
assert row.slots_cap == 0
|
|
body = await (await client.get("/api/system/workers")).get_json()
|
|
worker = next(lane for lane in body["lanes"] if lane["name"] == "worker")
|
|
assert worker["enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_raising_it_off_zero_turns_the_lane_on(client, db, no_live_workers):
|
|
await client.post("/api/system/workers/ml", json={"slots_cap": 1})
|
|
|
|
assert (await _lane_row(db, "ml")).slots_cap == 1
|
|
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 True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_model_fetch_fires_on_the_transition_not_on_every_write(
|
|
client, db, no_live_workers, 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
|
|
be the TRANSITION rather than "a field was sent", which is what it tested
|
|
before the UI stopped sending `enabled` at all."""
|
|
monkeypatch.setattr(
|
|
wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None),
|
|
)
|
|
fired = []
|
|
monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True)
|
|
|
|
first = await (await client.post(
|
|
"/api/system/workers/ml", json={"slots_cap": 1},
|
|
)).get_json()
|
|
second = await (await client.post(
|
|
"/api/system/workers/ml", json={"slots_cap": 2},
|
|
)).get_json()
|
|
|
|
assert first["fetching_models"] is True
|
|
assert second["fetching_models"] is False
|
|
assert fired == [1]
|
|
|
|
|
|
# --- what is refused ---------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_cap_above_the_derived_ceiling_is_refused(
|
|
client, db, no_live_workers,
|
|
):
|
|
"""The ceiling is the machine's, not the operator's, and it is the one
|
|
bound they cannot lower themselves past. The detail is written to be read
|
|
by a person — a refused control with no reason reads as a bug."""
|
|
before = (await _lane_row(db, "ml")).slots_cap
|
|
|
|
resp = await client.post(
|
|
"/api/system/workers/ml", json={"slots_cap": 10_000},
|
|
)
|
|
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert "container can hold" in body["detail"]
|
|
await _refreshed(db, "ml", before)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_negative_cap_is_refused(client, db, no_live_workers):
|
|
resp = await client.post("/api/system/workers/worker", json={"slots_cap": -1})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_boolean_is_not_accepted_as_a_cap(client, no_live_workers):
|
|
"""`True` is an int in Python. Reading it as a cap of 1 would be a control
|
|
that appears to work and sets something nobody asked for."""
|
|
resp = await client.post("/api/system/workers/worker", json={"slots_cap": True})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_retired_fields_are_no_longer_accepted(client, no_live_workers):
|
|
"""`slots`, `enabled` and `autoscale` are gone. A client still sending one
|
|
must be told, not silently ignored — a POST that returns 200 having
|
|
changed nothing is the worst of the three outcomes."""
|
|
for field in ("slots", "enabled", "autoscale"):
|
|
resp = await client.post(
|
|
"/api/system/workers/worker", json={field: 2},
|
|
)
|
|
assert resp.status_code == 400, field
|
|
|
|
|
|
@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/nope", json={"slots_cap": 1})
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
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,
|
|
):
|
|
resp = await client.post("/api/system/workers/worker", json={})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
async def _refreshed(db, name: str, expected: int) -> None:
|
|
row = await _lane_row(db, name)
|
|
await db.refresh(row)
|
|
assert row.slots_cap == expected, "a refused write must store nothing"
|