test_dispatch_fairness.py

"""Fair-share dispatch adversarial suite (AD4, ADR-0018, roadmap M4).

Merge-gated property: a noisy neighbor with an arbitrarily large backlog must never prevent
another user in the same lane from being dispatched this cycle, and a global per-cycle cap must
never let a lower-priority lane preempt a higher one. If an assertion here fails, treat it like
the tenant-isolation suite (`tests/test_repository_isolation.py`) — a fairness regression, not a
cosmetic one.
"""
from __future__ import annotations

from collections import Counter
from datetime import UTC, datetime, timedelta

import pytest
from hypothesis import given
from hypothesis import strategies as st

from openjobradar.scheduling.dispatcher import (
    Dispatcher,
    LaneCapacity,
    UnknownLaneError,
    plan_lane,
)
from openjobradar.scheduling.due_work import DueWorkIndex
from openjobradar.tenancy import TenantContext

_BASE = datetime(2026, 8, 23, 0, 0, 0, tzinfo=UTC)


def ts(offset_s: int = 0) -> str:
    return (_BASE + timedelta(seconds=offset_s)).strftime("%Y-%m-%dT%H:%M:%SZ")


def ctx(name: str) -> TenantContext:
    return TenantContext.from_sub(f"{name}-aaaaaaaa")


def _generous_capacities(limit: int = 1000) -> dict[str, LaneCapacity]:
    return {lane: LaneCapacity(per_cycle_limit=limit, per_user_limit=limit) for lane in ("free", "plus", "pro")}


# --- Lane assignment -----------------------------------------------------------


def test_lane_assignment_follows_plan_lane_mapping() -> None:
    index = DueWorkIndex()
    plan_of = {ctx("pro-user").user_id: "pro", ctx("plus-user").user_id: "plus", ctx("free-user").user_id: "free"}
    for user_id in plan_of:
        index.schedule(TenantContext(user_id), "poll_org", "acme", ts(0))

    dispatcher = Dispatcher(index, lambda uid: plan_lane(plan_of[uid]), _generous_capacities())
    batch = dispatcher.dispatch(ts(0))

    lanes_by_user = {item.entry.user_id: item.lane for item in batch.items}
    assert lanes_by_user[ctx("pro-user").user_id] == "pro"
    assert lanes_by_user[ctx("plus-user").user_id] == "plus"
    assert lanes_by_user[ctx("free-user").user_id] == "free"


def test_unrecognized_plan_falls_back_to_free_lane_safely() -> None:
    assert plan_lane("enterprise-mystery-tier") == "free"


def test_unknown_lane_in_capacities_rejected_at_construction() -> None:
    index = DueWorkIndex()
    with pytest.raises(UnknownLaneError):
        Dispatcher(index, plan_lane, {"free": LaneCapacity(10), "enterprise": LaneCapacity(10)})


def test_unmapped_lane_at_dispatch_time_fails_closed() -> None:
    index = DueWorkIndex()
    index.schedule(ctx("mystery-user"), "poll_org", "acme", ts(0))
    dispatcher = Dispatcher(index, lambda uid: "enterprise", {"free": LaneCapacity(10)})
    with pytest.raises(UnknownLaneError):
        dispatcher.dispatch(ts(0))


def test_empty_index_dispatches_nothing() -> None:
    dispatcher = Dispatcher(DueWorkIndex(), plan_lane, _generous_capacities())
    batch = dispatcher.dispatch(ts(0))
    assert batch.items == () and batch.shed_user_ids == ()


@pytest.mark.parametrize("kwargs", [{"per_cycle_limit": 0}, {"per_cycle_limit": 5, "per_user_limit": 0}])
def test_lane_capacity_rejects_non_positive_limits(kwargs: dict) -> None:
    with pytest.raises(ValueError):
        LaneCapacity(**kwargs)


def test_negative_global_batch_limit_rejected() -> None:
    index = DueWorkIndex()
    index.schedule(ctx("someone"), "poll_org", "acme", ts(0))
    dispatcher = Dispatcher(index, lambda _uid: "free", _generous_capacities())
    with pytest.raises(ValueError):
        dispatcher.dispatch(ts(0), global_batch_limit=-1)


# --- Noisy-neighbor shedding -----------------------------------------------------


def test_noisy_neighbor_does_not_starve_light_users_in_the_same_lane() -> None:
    index = DueWorkIndex()
    noisy = ctx("noisy-user")
    for i in range(50):
        index.schedule(noisy, "poll_org", f"org-{i:03d}", ts(0))
    lights = [ctx(f"light-user-{c}") for c in "abcd"]
    for light in lights:
        index.schedule(light, "poll_org", "only-org", ts(0))

    capacity = {"free": LaneCapacity(per_cycle_limit=8, per_user_limit=2)}
    dispatcher = Dispatcher(index, lambda _uid: "free", capacity)
    batch = dispatcher.dispatch(ts(0))

    dispatched_by_user = Counter(item.entry.user_id for item in batch.items)
    for light in lights:
        assert dispatched_by_user[light.user_id] == 1, "a light user was starved by the noisy neighbor"
    assert dispatched_by_user[noisy.user_id] == 2  # capped at per_user_limit
    assert noisy.user_id in batch.shed_user_ids
    assert all(light.user_id not in batch.shed_user_ids for light in lights)
    assert len(batch.items) == 6  # 4 light + 2 from the noisy user


def test_per_user_limit_caps_a_single_user_even_with_room_in_the_cycle() -> None:
    index = DueWorkIndex()
    solo = ctx("solo-user")
    for i in range(20):
        index.schedule(solo, "poll_org", f"org-{i:03d}", ts(0))

    capacity = {"free": LaneCapacity(per_cycle_limit=100, per_user_limit=3)}
    dispatcher = Dispatcher(index, lambda _uid: "free", capacity)
    batch = dispatcher.dispatch(ts(0))

    assert len(batch.items) == 3
    assert batch.shed_user_ids == (solo.user_id,)


# --- Cross-lane priority under a global cap --------------------------------------


def test_global_batch_limit_prioritizes_pro_over_free_after_per_lane_fairness() -> None:
    index = DueWorkIndex()
    pro_user, free_user = ctx("pro-user"), ctx("free-user")
    for i in range(5):
        index.schedule(pro_user, "poll_org", f"org-{i}", ts(0))
        index.schedule(free_user, "poll_org", f"org-{i}", ts(0))

    lane_of = {pro_user.user_id: "pro", free_user.user_id: "free"}
    dispatcher = Dispatcher(
        index, lambda uid: lane_of[uid], {"pro": LaneCapacity(5, 5), "free": LaneCapacity(5, 5)}
    )
    batch = dispatcher.dispatch(ts(0), global_batch_limit=6)

    by_lane = Counter(item.lane for item in batch.items)
    assert by_lane["pro"] == 5, "pro's fair share was preempted by a lower lane"
    assert by_lane["free"] == 1
    assert len(batch.items) == 6
    assert free_user.user_id in batch.shed_user_ids
    assert pro_user.user_id not in batch.shed_user_ids


def test_global_batch_limit_of_zero_dispatches_nothing_but_reports_shedding() -> None:
    index = DueWorkIndex()
    index.schedule(ctx("someone"), "poll_org", "acme", ts(0))
    dispatcher = Dispatcher(index, lambda _uid: "free", _generous_capacities())
    batch = dispatcher.dispatch(ts(0), global_batch_limit=0)
    assert batch.items == ()
    assert batch.shed_user_ids == (ctx("someone").user_id,)


# --- Property-based: fairness holds for arbitrary backlogs ------------------------


@st.composite
def _backlog(draw):
    n_users = draw(st.integers(1, 8))
    per_user_counts = draw(st.lists(st.integers(1, 12), min_size=n_users, max_size=n_users))
    cycle_limit = draw(st.integers(1, 20))
    user_limit = draw(st.integers(1, 6))
    return per_user_counts, cycle_limit, user_limit


@given(_backlog())
def test_drain_never_exceeds_configured_caps_under_arbitrary_pressure(backlog) -> None:
    per_user_counts, cycle_limit, user_limit = backlog
    index = DueWorkIndex()
    users = [ctx(f"user-{i:03d}") for i in range(len(per_user_counts))]
    for user, count in zip(users, per_user_counts, strict=True):
        for k in range(count):
            index.schedule(user, "poll_org", f"org-{k:03d}", ts(0))

    dispatcher = Dispatcher(index, lambda _uid: "free", {"free": LaneCapacity(cycle_limit, user_limit)})
    batch = dispatcher.dispatch(ts(0))

    assert len(batch.items) <= cycle_limit
    per_user_dispatched = Counter(item.entry.user_id for item in batch.items)
    assert all(count <= user_limit for count in per_user_dispatched.values())

    for user, due_count in zip(users, per_user_counts, strict=True):
        dispatched = per_user_dispatched[user.user_id]
        if dispatched < due_count:
            assert user.user_id in batch.shed_user_ids


@given(_backlog())
def test_nobody_is_shed_when_capacity_is_not_actually_scarce(backlog) -> None:
    per_user_counts, _cycle_limit, _user_limit = backlog
    index = DueWorkIndex()
    users = [ctx(f"user-{i:03d}") for i in range(len(per_user_counts))]
    for user, count in zip(users, per_user_counts, strict=True):
        for k in range(count):
            index.schedule(user, "poll_org", f"org-{k:03d}", ts(0))

    capacity = LaneCapacity(per_cycle_limit=sum(per_user_counts), per_user_limit=max(per_user_counts))
    dispatcher = Dispatcher(index, lambda _uid: "free", {"free": capacity})
    batch = dispatcher.dispatch(ts(0))

    assert batch.shed_user_ids == ()
    assert len(batch.items) == sum(per_user_counts)