FabledCurator/tests/test_worker_control.py

"""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, consuming=None,
):
    # `consuming` defaults to the lane's full queue set — i.e. an ENABLED
    # lane. Tests for the disabled case pass an empty set explicitly.
    if consuming is None:
        consuming = set(LANES_BY_NAME[lane_name].queues)
    state = wc.LaneLiveState(
        present=present,
        replicas=len(pools),
        hostnames=sorted(pools),
        pools=dict(pools),
        reserved=reserved,
        consuming=set(consuming),
    )
    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


# --- the inspect round trips -------------------------------------------------


class _Inspect:
    """Records what was asked, and of whom."""

    def __init__(self, calls, queues, destination=None):
        self.calls = calls
        self._queues = queues
        self.destination = destination

    def active_queues(self):
        self.calls.append(("active_queues", self.destination))
        return self._queues

    def stats(self):
        self.calls.append(("stats", self.destination))
        return {}

    def active(self):
        self.calls.append(("active", self.destination))
        return {}

    def reserved(self):
        self.calls.append(("reserved", self.destination))
        return {}


def _stub_inspect(monkeypatch, queues):
    calls = []

    class _Control:
        def inspect(self, timeout=None, destination=None):
            return _Inspect(calls, queues, destination)

    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 calls


def test_only_the_first_read_is_a_broadcast(monkeypatch):
    """A broadcast cannot know how many replies to expect, so it waits out its
    whole timeout. Four of those is ~8s, and lane_view sits on the Settings
    card — that was the load time of the Worker lanes page.

    Naming the destinations lets celery stop as soon as those nodes answer.
    """
    calls = _stub_inspect(monkeypatch, {
        "worker@a": [{"name": q} for q in LANES_BY_NAME["worker"].queues],
    })

    wc.inspect_lanes_sync()

    assert calls[0] == ("active_queues", None), "the first read discovers nodes"
    for name, destination in calls[1:]:
        assert destination == ["worker@a"], (
            f"{name} broadcast instead of addressing the node that answered"
        )


def test_nothing_answering_costs_one_round_trip_not_four(monkeypatch):
    """The three later reads exist only to describe what answered, so with an
    empty roster they describe nothing — at three full timeouts.

    This is the broker-down case, which is exactly when the healthcheck and
    the card need an answer rather than a long wait.
    """
    calls = _stub_inspect(monkeypatch, {})

    out = wc.inspect_lanes_sync()

    assert calls == [("active_queues", None)]
    assert all(not state.present for state in out.values())


def test_the_deadline_exceeds_the_worst_case_it_waits_for(monkeypatch):
    """Same relation the roster needed after it shipped a budget equal to its
    own worst case. Asserted as a relation, not a number: a fifth inspect call
    is how this comes back."""
    work = wc.CONTROL_TIMEOUT_SECONDS * wc.CONTROL_ROUND_TRIPS
    assert wc.INSPECT_BUDGET_SECONDS > work
    assert wc.CONTROL_SLACK_SECONDS > 0


def test_the_round_trip_bound_matches_the_reads_actually_made():
    import inspect as _inspect

    src = _inspect.getsource(wc.inspect_lanes_sync)
    reads = src.count("insp.") + src.count("targeted.")
    assert reads == wc.CONTROL_ROUND_TRIPS, (
        f"inspect_lanes_sync makes {reads} reads but CONTROL_ROUND_TRIPS "
        f"says {wc.CONTROL_ROUND_TRIPS}"
    )




# --- the sizing pass ---------------------------------------------------------
#
# One sweep replaced two on 2026-09-23. The reconcile drove the pool to a
# stored `slots`; the autoscaler moved it away from that same number; and most
# of the autoscaler's design existed to stop the reconcile undoing its work.
# The stored number is gone, so there is nothing left to disagree about — and
# these tests no longer have to assert that two sweeps get along.
#
# The fixtures deliberately set the live pool and the cap to DIFFERENT numbers
# wherever the distinction matters. Lesson #4318: equal fixtures cannot tell
# "reads live" from "reads stored", and that is exactly how the old autoscaler
# shipped unable to do its job with every test green.


def _sizing_live(
    monkeypatch, *, pools, active, reserved, depth, consuming=None, present=True,
):
    state = wc.LaneLiveState(
        present=present,
        replicas=len(pools),
        hostnames=sorted(pools),
        pools=dict(pools),
        active=active,
        reserved=reserved,
        consuming=set(LANES_BY_NAME["worker"].queues if consuming is None else consuming),
    )
    monkeypatch.setattr(
        wc, "inspect_lanes_sync",
        lambda: {name: (state if name == "worker" else wc.LaneLiveState())
                 for name in LANES_BY_NAME},
    )
    monkeypatch.setattr(
        wc, "_queue_depths_sync",
        lambda: {q: depth for lane in LANES_BY_NAME.values() for q in lane.queues},
    )
    return state


def _worker(sized):
    return next(d for d in sized if d.lane == "worker")


# --- what a lane has work for ------------------------------------------------


def test_work_in_flight_and_waiting_each_justify_a_worker():
    # Two running plus six queued is eight tasks, so eight workers — if the
    # cap allowed it.
    assert wc.wanted_slots(cap=10, active=2, pending=6) == 8


def test_it_never_exceeds_the_cap():
    assert wc.wanted_slots(cap=2, active=2, pending=4000) == 2


def test_an_idle_lane_falls_to_one_and_not_to_zero():
    """billiard will not run an empty pool, and the parked process is what
    `add_consumer` lands on when the cap goes back up."""
    assert wc.wanted_slots(cap=8, active=0, pending=0) == 1


def test_a_lane_capped_at_zero_still_keeps_its_process():
    """`cap 0` is expressed by cancelling CONSUMERS, not by emptying the pool.
    A lane with no process reads as absent, which is the same signal as a
    crash — and the roster exists to keep those two apart."""
    assert wc.wanted_slots(cap=0, active=0, pending=0) == wc.MIN_POOL_SLOTS


def test_an_unknown_backlog_contributes_nothing_rather_than_zero():
    """The broker did not answer for these queues. Reading that as "empty"
    would shrink a lane on the strength of a failed read (snippet #3969) —
    and reading it as "huge" would grow one. It contributes nothing, and the
    work actually in flight still counts."""
    assert wc.wanted_slots(cap=8, active=3, pending=None) == 3


# --- growing -----------------------------------------------------------------


def test_a_backlog_is_met_in_one_tick_not_one_slot_per_minute(monkeypatch):
    """The operator's condition for always-on autoscaling: *"always on"* is
    only pleasant if the ramp keeps up. Growing +1 per minute would take four
    minutes to answer a burst, and the old autoscaler did exactly that."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=0, depth=4000)

    d = _worker(wc.size_lanes_sync({"worker": 4}))

    assert (d.action, d.slots) == ("grew", 4)
    assert control.grew == [(3, ["host-a"])]


def test_a_prefetched_backlog_counts(monkeypatch):
    """THE case an LLEN-only implementation misses. Celery prefetches, so a
    lane can read queue depth 0 while holding thirty tasks in worker memory —
    a sizing pass watching LLEN alone sees an idle system and shrinks."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=30, depth=0)

    assert _worker(wc.size_lanes_sync({"worker": 4})).action == "grew"
    assert control.grew == [(3, ["host-a"])]


def test_a_restarted_worker_is_pulled_back_up(monkeypatch):
    """What the five-minute reconcile existed for, now done in one minute by
    the pass that was already running. `pool_grow` is not durable: a worker
    restarted by its supervisor comes back at its ENV concurrency, silently
    below what the lane should be running."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=9, depth=0)

    assert _worker(wc.size_lanes_sync({"worker": 4})).slots == 4
    assert control.grew == [(3, ["host-a"])]


# --- shrinking ---------------------------------------------------------------


def test_an_idle_lane_gives_a_worker_back_one_at_a_time(monkeypatch):
    """Operator: *"so that idle instances quiet down when not running."*

    One per tick on the way down, against immediate growth. Being one worker
    too large for a minute costs a sleeping process; being too small costs
    work not happening. For ML it matters most — every new slot reloads a
    multi-GB model, so a slow shrink is what stops a quiet patch from paying
    that cost again a minute later."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 4}, active=0, reserved=0, depth=0)

    d = _worker(wc.size_lanes_sync({"worker": 4}))

    assert (d.action, d.slots) == ("shrank", 3)
    assert control.shrank == [(1, ["host-a"])]


def test_it_stops_shrinking_at_one(monkeypatch):
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0)

    assert _worker(wc.size_lanes_sync({"worker": 4})).action == "held"
    assert control.shrank == []


def test_lowering_the_cap_pulls_a_lane_down(monkeypatch):
    """The cap is a ceiling on the live pool, not only on future growth."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 4}, active=4, reserved=99, depth=0)

    assert _worker(wc.size_lanes_sync({"worker": 2})).slots == 3
    assert control.shrank == [(1, ["host-a"])]


# --- the fixed point ---------------------------------------------------------


def test_a_settled_lane_sends_no_control_messages(monkeypatch):
    """THE property this pass lives or dies on. It runs every minute forever,
    so a converged tick must be silent — one inspect, one LLEN sweep, nothing
    else. An enforcer that re-issues a grow of zero churns forever and buries
    a real correction in its own noise (lesson #4183)."""
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 2}, active=2, reserved=0, depth=0)

    d = _worker(wc.size_lanes_sync({"worker": 4}))

    assert d.action == "held"
    # EVERY control family, not just the pool ones: an unconditional
    # add_consumer for four queues per tick is harmless per call and unbounded
    # in aggregate, and invisible.
    assert control.grew == []
    assert control.shrank == []
    assert control.added == []
    assert control.cancelled == []


def test_an_absent_lane_is_skipped_not_corrected(monkeypatch):
    """`present=False` means nothing answered — a worker restarting, or an
    unreachable broker. It is NOT zero workers, and there is nothing to send
    a message to."""
    control = _stub_control(monkeypatch)
    _sizing_live(
        monkeypatch, pools={}, active=0, reserved=0, depth=0, present=False,
    )

    d = _worker(wc.size_lanes_sync({"worker": 4}))

    assert d.action == "skipped"
    assert control.grew == [] and control.shrank == []


def test_a_lane_with_no_row_is_left_alone(monkeypatch):
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=99)

    assert wc.size_lanes_sync({}) == []
    assert control.grew == []


# --- the cap is also the switch ----------------------------------------------


def test_a_cap_of_zero_stops_the_lane_consuming(monkeypatch):
    control = _stub_control(monkeypatch)
    _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0)

    wc.size_lanes_sync({"worker": 0})

    assert sorted(q for q, _ in control.cancelled) == sorted(
        LANES_BY_NAME["worker"].queues
    )


def test_a_cap_above_zero_starts_it_consuming(monkeypatch):
    control = _stub_control(monkeypatch)
    _sizing_live(
        monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0,
        consuming=set(),
    )

    wc.size_lanes_sync({"worker": 2})

    assert sorted(q for q, _ in control.added) == sorted(
        LANES_BY_NAME["worker"].queues
    )


def test_an_already_stopped_lane_is_not_cancelled_again(monkeypatch):
    """The other half of the fixed point. `cancel_consumer` on a queue that is
    not being consumed succeeds and does nothing, so an unconditional pass
    would churn here forever with no symptom."""
    control = _stub_control(monkeypatch)
    _sizing_live(
        monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0,
        consuming=set(),
    )

    wc.size_lanes_sync({"worker": 0})

    assert control.cancelled == []
    assert control.added == []


def test_a_stopped_lane_still_keeps_exactly_one_process(monkeypatch):
    """The live bug of 2026-09-23, from the other direction: the pass must not
    try to empty a pool billiard will not empty, and must not report having
    done so every tick."""
    control = _stub_control(monkeypatch)
    _sizing_live(
        monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0,
        consuming=set(),
    )

    assert _worker(wc.size_lanes_sync({"worker": 0})).action == "held"
    assert control.shrank == []


def test_the_sizing_task_is_registered_and_scheduled():
    """A pass nothing schedules is a pass that never runs — and since this one
    replaced two entries, a stale name in the beat schedule would leave the
    lanes unmanaged with nothing red anywhere."""
    from backend.app.celery_app import celery

    name = "backend.app.tasks.maintenance.size_worker_lanes"
    assert name in celery.tasks
    entries = celery.conf.beat_schedule
    assert any(e["task"] == name for e in entries.values())
    # The two it replaced are gone, not merely unreferenced.
    tasks = {e["task"] for e in entries.values()}
    assert "backend.app.tasks.maintenance.reconcile_worker_lanes" not in tasks
    assert "backend.app.tasks.maintenance.autoscale_worker_lanes" not in tasks