feat: optional 28-day historical fairness for nights and weekends

app/scheduler.py

@@ -1,6 +1,6 @@
 from datetime import date
 
-from app.constants import RiskLevel
+from app.constants import RiskLevel, ShiftType
 from app.models import Employee, EmployeeProfile, Shift
 from app.personalized_workload import calculate_personalized_workload_score
 
@@ -11,7 +11,37 @@ def assign_shifts(
     workload_ratio_threshold: float = 0.75,
     average_score_threshold: float = 2.0,
     high_risk_flag_count: int = 2,
+    historical_shifts_by_employee: dict[str, list[Shift]] | None = None,
+    history_lookback_days: int = 28,
 ) -> list[Shift]:
+    if history_lookback_days < 1:
+        raise ValueError("history_lookback_days must be positive")
+
+    historical = historical_shifts_by_employee or {}
+
+    def historical_penalty(employee: Employee, shift: Shift) -> tuple[int, int]:
+        target_day = date.fromisoformat(shift.date)
+        same_type_count = 0
+        weekend_count = 0
+        is_weekend = target_day.weekday() >= 5
+
+        for past_shift in historical.get(employee.name, []):
+            past_day = date.fromisoformat(past_shift.date)
+            age_days = (target_day - past_day).days
+            if not (0 < age_days <= history_lookback_days):
+                continue
+
+            if (
+                shift.shift_type in (ShiftType.NIGHT, ShiftType.EVENING)
+                and past_shift.shift_type == shift.shift_type
+            ):
+                same_type_count += 1
+
+            if is_weekend and past_day.weekday() >= 5:
+                weekend_count += 1
+
+        return (same_type_count, weekend_count)
+
     risk_mapping = {
         RiskLevel.LOW: 0,
         RiskLevel.MEDIUM: 1,
@@ -60,6 +90,7 @@ def sort_key(e: Employee):
                 ],
                 e.assigned_shift_count,
                 personalized_score,
+                historical_penalty(e, shift),
                 e.total_workload_score,
                 len(e.skills),
                 rotating_rank[id(e)],

tests/test_scheduler.py

@@ -238,3 +238,86 @@ def test_rotation_does_not_override_skill_requirements():
     assign_shifts([anna, zoe], [shift])
 
     assert shift.assigned_employees == [zoe]
+
+
+def test_history_fairness_prefers_employee_with_fewer_recent_nights():
+    anna = Employee("Anna", 5, ["CT"])
+    zoe = Employee("Zoe", 5, ["CT"])
+    night = Shift("2026-05-20", ShiftType.NIGHT, "CT")
+
+    assign_shifts(
+        [anna, zoe],
+        [night],
+        historical_shifts_by_employee={
+            "Anna": [Shift("2026-05-13", ShiftType.NIGHT, "CT")],
+            "Zoe": [],
+        },
+    )
+
+    assert night.assigned_employees == [zoe]
+
+
+def test_history_fairness_prefers_fewer_weekends_on_weekend_shifts():
+    anna = Employee("Anna", 5, ["CT"])
+    zoe = Employee("Zoe", 5, ["CT"])
+    sunday = Shift("2026-05-24", ShiftType.DAY, "CT")
+
+    assign_shifts(
+        [anna, zoe],
+        [sunday],
+        historical_shifts_by_employee={
+            "Anna": [Shift("2026-05-17", ShiftType.DAY, "CT")],
+            "Zoe": [],
+        },
+    )
+
+    assert sunday.assigned_employees == [zoe]
+
+
+def test_historical_penalty_excludes_old_and_future_shifts():
+    anna = Employee("Anna", 5, ["CT"])
+    zoe = Employee("Zoe", 5, ["CT"])
+    target = Shift("2026-05-20", ShiftType.NIGHT, "CT")
+
+    with_history = {
+        "Anna": [
+            Shift("2026-03-01", ShiftType.NIGHT, "CT"),
+            Shift("2026-05-27", ShiftType.NIGHT, "CT"),
+        ],
+        "Zoe": [],
+    }
+    # Old and future history is ignored; the result therefore matches
+    # the normal deterministic rotation rather than imposing a penalty.
+    reference = Shift("2026-05-20", ShiftType.NIGHT, "CT")
+    reference_employees = [Employee("Anna", 5, ["CT"]), Employee("Zoe", 5, ["CT"])]
+    assign_shifts(reference_employees, [reference])
+
+    assign_shifts(
+        [anna, zoe],
+        [target],
+        historical_shifts_by_employee=with_history,
+    )
+    assert target.assigned_employees[0].name == reference.assigned_employees[0].name
+
+
+def test_history_does_not_override_skill_eligibility():
+    qualified = Employee("Qualified", 5, ["MR"])
+    other = Employee("Not qualified", 5, ["CT"])
+    night = Shift("2026-05-20", ShiftType.NIGHT, "MR")
+
+    assign_shifts(
+        [qualified, other],
+        [night],
+        historical_shifts_by_employee={
+            "Qualified": [Shift("2026-05-13", ShiftType.NIGHT, "MR")],
+        },
+    )
+
+    assert night.assigned_employees == [qualified]
+
+
+def test_invalid_history_window_is_rejected():
+    import pytest
+
+    with pytest.raises(ValueError, match="history_lookback_days"):
+        assign_shifts([], [], history_lookback_days=0)