BE-CAL-007: MinimumWorkload warning rules + BE-CAL-008: past-slot immutability

BE-CAL-007: Workload warning computation (already implemented in prior wave, verified tests pass - 24/24). Computes daily/weekly/monthly/yearly scheduled minutes and compares against user thresholds. Warnings are advisory only. BE-CAL-008: New slot_immutability service with guards for: - Forbid edit/cancel of past real slots (raises ImmutableSlotError) - Forbid edit/cancel of past virtual slots - Plan-edit/plan-cancel helper to identify past materialized slot IDs that must not be retroactively modified Tests: 19/19 passing.
2026-03-31 04:16:50 +00:00
parent 570cfee5cd
commit 4f0e933de3
4 changed files with 1037 additions and 7 deletions
--- a/tests/test_minimum_workload.py
+++ b/tests/test_minimum_workload.py
@@ -0,0 +1,451 @@
+"""Tests for MinimumWorkload warning rules (BE-CAL-007).
+
+Tests cover:
+  - _date_range_for_period computation
+  - _sum_real_slots aggregation
+  - _sum_virtual_slots aggregation
+  - check_workload_warnings comparison logic
+  - get_workload_warnings_for_date end-to-end convenience
+  - Warnings are advisory (non-blocking)
+"""
+
+import pytest
+from datetime import date, time
+
+from tests.conftest import auth_header
+
+from app.models.calendar import (
+    SchedulePlan,
+    SlotStatus,
+    SlotType,
+    EventType,
+    TimeSlot,
+    DayOfWeek,
+)
+from app.models.minimum_workload import MinimumWorkload
+from app.services.minimum_workload import (
+    _date_range_for_period,
+    _sum_real_slots,
+    _sum_virtual_slots,
+    check_workload_warnings,
+    compute_scheduled_minutes,
+    get_workload_warnings_for_date,
+    get_workload_config,
+)
+from app.schemas.calendar import WorkloadWarningItem
+
+
+# ---------------------------------------------------------------------------
+# Unit: _date_range_for_period
+# ---------------------------------------------------------------------------
+
+class TestDateRangeForPeriod:
+    def test_daily(self):
+        d = date(2026, 3, 15)  # Sunday
+        start, end = _date_range_for_period("daily", d)
+        assert start == end == d
+
+    def test_weekly_midweek(self):
+        d = date(2026, 3, 18)  # Wednesday
+        start, end = _date_range_for_period("weekly", d)
+        assert start == date(2026, 3, 16)  # Monday
+        assert end == date(2026, 3, 22)    # Sunday
+
+    def test_weekly_monday(self):
+        d = date(2026, 3, 16)  # Monday
+        start, end = _date_range_for_period("weekly", d)
+        assert start == date(2026, 3, 16)
+        assert end == date(2026, 3, 22)
+
+    def test_weekly_sunday(self):
+        d = date(2026, 3, 22)  # Sunday
+        start, end = _date_range_for_period("weekly", d)
+        assert start == date(2026, 3, 16)
+        assert end == date(2026, 3, 22)
+
+    def test_monthly(self):
+        d = date(2026, 3, 15)
+        start, end = _date_range_for_period("monthly", d)
+        assert start == date(2026, 3, 1)
+        assert end == date(2026, 3, 31)
+
+    def test_monthly_february(self):
+        d = date(2026, 2, 10)
+        start, end = _date_range_for_period("monthly", d)
+        assert start == date(2026, 2, 1)
+        assert end == date(2026, 2, 28)
+
+    def test_monthly_december(self):
+        d = date(2026, 12, 25)
+        start, end = _date_range_for_period("monthly", d)
+        assert start == date(2026, 12, 1)
+        assert end == date(2026, 12, 31)
+
+    def test_yearly(self):
+        d = date(2026, 6, 15)
+        start, end = _date_range_for_period("yearly", d)
+        assert start == date(2026, 1, 1)
+        assert end == date(2026, 12, 31)
+
+    def test_unknown_period_raises(self):
+        with pytest.raises(ValueError, match="Unknown period"):
+            _date_range_for_period("hourly", date(2026, 1, 1))
+
+
+# ---------------------------------------------------------------------------
+# Unit: check_workload_warnings (pure comparison, no DB)
+# ---------------------------------------------------------------------------
+
+class TestCheckWorkloadWarnings:
+    """Test the comparison logic with pre-computed scheduled_minutes."""
+
+    def test_no_warnings_when_all_zero_config(self, db, seed):
+        """Default config (all zeros) never triggers warnings."""
+        scheduled = {
+            "daily": {"work": 0, "on_call": 0, "entertainment": 0},
+            "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+            "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+            "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+        }
+        warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled)
+        assert warnings == []
+
+    def test_warning_when_below_threshold(self, db, seed):
+        """Setting a threshold higher than scheduled triggers a warning."""
+        # Set daily work minimum to 60 min
+        cfg = MinimumWorkload(
+            user_id=seed["admin_user"].id,
+            config={
+                "daily": {"work": 60, "on_call": 0, "entertainment": 0},
+                "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        )
+        db.add(cfg)
+        db.commit()
+
+        scheduled = {
+            "daily": {"work": 30, "on_call": 0, "entertainment": 0},
+            "weekly": {"work": 100, "on_call": 0, "entertainment": 0},
+            "monthly": {"work": 400, "on_call": 0, "entertainment": 0},
+            "yearly": {"work": 5000, "on_call": 0, "entertainment": 0},
+        }
+        warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled)
+        assert len(warnings) == 1
+        w = warnings[0]
+        assert w.period == "daily"
+        assert w.category == "work"
+        assert w.current_minutes == 30
+        assert w.minimum_minutes == 60
+        assert w.shortfall_minutes == 30
+
+    def test_no_warning_when_meeting_threshold(self, db, seed):
+        cfg = MinimumWorkload(
+            user_id=seed["admin_user"].id,
+            config={
+                "daily": {"work": 30, "on_call": 0, "entertainment": 0},
+                "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        )
+        db.add(cfg)
+        db.commit()
+
+        scheduled = {
+            "daily": {"work": 30, "on_call": 0, "entertainment": 0},
+            "weekly": {"work": 100, "on_call": 0, "entertainment": 0},
+            "monthly": {"work": 400, "on_call": 0, "entertainment": 0},
+            "yearly": {"work": 5000, "on_call": 0, "entertainment": 0},
+        }
+        warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled)
+        assert warnings == []
+
+    def test_multiple_warnings_across_periods_and_categories(self, db, seed):
+        cfg = MinimumWorkload(
+            user_id=seed["admin_user"].id,
+            config={
+                "daily": {"work": 50, "on_call": 20, "entertainment": 0},
+                "weekly": {"work": 300, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        )
+        db.add(cfg)
+        db.commit()
+
+        scheduled = {
+            "daily": {"work": 10, "on_call": 5, "entertainment": 0},
+            "weekly": {"work": 100, "on_call": 0, "entertainment": 0},
+            "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+            "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+        }
+        warnings = check_workload_warnings(db, seed["admin_user"].id, scheduled)
+        assert len(warnings) == 3
+        periods_cats = {(w.period, w.category) for w in warnings}
+        assert ("daily", "work") in periods_cats
+        assert ("daily", "on_call") in periods_cats
+        assert ("weekly", "work") in periods_cats
+
+
+# ---------------------------------------------------------------------------
+# Integration: _sum_real_slots
+# ---------------------------------------------------------------------------
+
+class TestSumRealSlots:
+    def test_sums_work_slots(self, db, seed):
+        """Real work slots are summed correctly."""
+        user_id = seed["admin_user"].id
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=30,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=20,
+            scheduled_at=time(10, 0), status=SlotStatus.FINISHED,
+        ))
+        db.commit()
+
+        totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals["work"] == 50
+        assert totals["on_call"] == 0
+        assert totals["entertainment"] == 0
+
+    def test_excludes_skipped_and_aborted(self, db, seed):
+        user_id = seed["admin_user"].id
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=30,
+            scheduled_at=time(9, 0), status=SlotStatus.SKIPPED,
+        ))
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=20,
+            scheduled_at=time(10, 0), status=SlotStatus.ABORTED,
+        ))
+        db.commit()
+
+        totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals["work"] == 0
+
+    def test_excludes_system_slots(self, db, seed):
+        user_id = seed["admin_user"].id
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.SYSTEM, estimated_duration=10,
+            scheduled_at=time(8, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.commit()
+
+        totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals == {"work": 0, "on_call": 0, "entertainment": 0}
+
+    def test_sums_across_date_range(self, db, seed):
+        user_id = seed["admin_user"].id
+        for day in [15, 16, 17]:
+            db.add(TimeSlot(
+                user_id=user_id, date=date(2026, 3, day),
+                slot_type=SlotType.WORK, estimated_duration=10,
+                scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+            ))
+        db.commit()
+
+        totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 17))
+        assert totals["work"] == 30
+
+    def test_multiple_categories(self, db, seed):
+        user_id = seed["admin_user"].id
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=25,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.ON_CALL, estimated_duration=15,
+            scheduled_at=time(10, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.ENTERTAINMENT, estimated_duration=10,
+            scheduled_at=time(11, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.commit()
+
+        totals = _sum_real_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals == {"work": 25, "on_call": 15, "entertainment": 10}
+
+
+# ---------------------------------------------------------------------------
+# Integration: _sum_virtual_slots
+# ---------------------------------------------------------------------------
+
+class TestSumVirtualSlots:
+    def test_sums_virtual_plan_slots(self, db, seed):
+        """Virtual slots from an active plan are counted."""
+        user_id = seed["admin_user"].id
+        plan = SchedulePlan(
+            user_id=user_id,
+            slot_type=SlotType.WORK,
+            estimated_duration=40,
+            at_time=time(9, 0),
+            on_day=DayOfWeek.SUN,  # 2026-03-15 is a Sunday
+            is_active=True,
+        )
+        db.add(plan)
+        db.commit()
+
+        totals = _sum_virtual_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals["work"] == 40
+
+    def test_skips_materialized_plan_slots(self, db, seed):
+        """If a plan slot is already materialized, it shouldn't be double-counted."""
+        user_id = seed["admin_user"].id
+        plan = SchedulePlan(
+            user_id=user_id,
+            slot_type=SlotType.WORK,
+            estimated_duration=40,
+            at_time=time(9, 0),
+            on_day=DayOfWeek.SUN,
+            is_active=True,
+        )
+        db.add(plan)
+        db.flush()
+
+        # Materialize it
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=40,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+            plan_id=plan.id,
+        ))
+        db.commit()
+
+        totals = _sum_virtual_slots(db, user_id, date(2026, 3, 15), date(2026, 3, 15))
+        assert totals["work"] == 0  # Already materialized, not double-counted
+
+
+# ---------------------------------------------------------------------------
+# Integration: compute_scheduled_minutes
+# ---------------------------------------------------------------------------
+
+class TestComputeScheduledMinutes:
+    def test_combines_real_and_virtual(self, db, seed):
+        user_id = seed["admin_user"].id
+
+        # Real slot on the 15th
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=20,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+        ))
+
+        # Plan that fires every day
+        plan = SchedulePlan(
+            user_id=user_id,
+            slot_type=SlotType.ON_CALL,
+            estimated_duration=10,
+            at_time=time(14, 0),
+            is_active=True,
+        )
+        db.add(plan)
+        db.commit()
+
+        result = compute_scheduled_minutes(db, user_id, date(2026, 3, 15))
+
+        # Daily: 20 work (real) + 10 on_call (virtual)
+        assert result["daily"]["work"] == 20
+        assert result["daily"]["on_call"] == 10
+
+        # Weekly: the real slot + virtual slots for every day in the week
+        # 2026-03-15 is Sunday → week is Mon 2026-03-09 to Sun 2026-03-15
+        assert result["weekly"]["work"] == 20
+        assert result["weekly"]["on_call"] >= 10  # At least the one day
+
+
+# ---------------------------------------------------------------------------
+# Integration: get_workload_warnings_for_date (end-to-end)
+# ---------------------------------------------------------------------------
+
+class TestGetWorkloadWarningsForDate:
+    def test_returns_warnings_when_below_threshold(self, db, seed):
+        user_id = seed["admin_user"].id
+
+        # Set daily work minimum to 60 min
+        db.add(MinimumWorkload(
+            user_id=user_id,
+            config={
+                "daily": {"work": 60, "on_call": 0, "entertainment": 0},
+                "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        ))
+
+        # Only 30 min of work scheduled
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=30,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.commit()
+
+        warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15))
+        assert len(warnings) >= 1
+        daily_work = [w for w in warnings if w.period == "daily" and w.category == "work"]
+        assert len(daily_work) == 1
+        assert daily_work[0].shortfall_minutes == 30
+
+    def test_no_warnings_when_above_threshold(self, db, seed):
+        user_id = seed["admin_user"].id
+
+        db.add(MinimumWorkload(
+            user_id=user_id,
+            config={
+                "daily": {"work": 30, "on_call": 0, "entertainment": 0},
+                "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        ))
+
+        db.add(TimeSlot(
+            user_id=user_id, date=date(2026, 3, 15),
+            slot_type=SlotType.WORK, estimated_duration=45,
+            scheduled_at=time(9, 0), status=SlotStatus.NOT_STARTED,
+        ))
+        db.commit()
+
+        warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15))
+        daily_work = [w for w in warnings if w.period == "daily" and w.category == "work"]
+        assert len(daily_work) == 0
+
+    def test_warning_data_structure(self, db, seed):
+        """Ensure warnings contain all required fields with correct types."""
+        user_id = seed["admin_user"].id
+
+        db.add(MinimumWorkload(
+            user_id=user_id,
+            config={
+                "daily": {"work": 100, "on_call": 0, "entertainment": 0},
+                "weekly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "monthly": {"work": 0, "on_call": 0, "entertainment": 0},
+                "yearly": {"work": 0, "on_call": 0, "entertainment": 0},
+            },
+        ))
+        db.commit()
+
+        warnings = get_workload_warnings_for_date(db, user_id, date(2026, 3, 15))
+        assert len(warnings) >= 1
+        w = warnings[0]
+        assert isinstance(w, WorkloadWarningItem)
+        assert isinstance(w.period, str)
+        assert isinstance(w.category, str)
+        assert isinstance(w.current_minutes, int)
+        assert isinstance(w.minimum_minutes, int)
+        assert isinstance(w.shortfall_minutes, int)
+        assert isinstance(w.message, str)
+        assert w.shortfall_minutes == w.minimum_minutes - w.current_minutes
--- a/tests/test_slot_immutability.py
+++ b/tests/test_slot_immutability.py
@@ -0,0 +1,234 @@
+"""Tests for past-slot immutability rules (BE-CAL-008).
+
+Tests cover:
+  - Editing a past real slot is forbidden
+  - Cancelling a past real slot is forbidden
+  - Editing a past virtual slot is forbidden
+  - Cancelling a past virtual slot is forbidden
+  - Editing/cancelling today's slots is allowed
+  - Editing/cancelling future slots is allowed
+  - Plan-edit / plan-cancel do not retroactively affect past materialized slots
+"""
+
+import pytest
+from datetime import date, time
+
+from app.models.calendar import (
+    SchedulePlan,
+    SlotStatus,
+    SlotType,
+    TimeSlot,
+    DayOfWeek,
+)
+from app.services.slot_immutability import (
+    ImmutableSlotError,
+    guard_edit_real_slot,
+    guard_cancel_real_slot,
+    guard_edit_virtual_slot,
+    guard_cancel_virtual_slot,
+    get_past_materialized_slot_ids,
+    guard_plan_edit_no_past_retroaction,
+    guard_plan_cancel_no_past_retroaction,
+)
+from app.services.plan_slot import make_virtual_slot_id
+
+
+TODAY = date(2026, 3, 31)
+YESTERDAY = date(2026, 3, 30)
+LAST_WEEK = date(2026, 3, 24)
+TOMORROW = date(2026, 4, 1)
+
+
+# ---------------------------------------------------------------------------
+# Helper
+# ---------------------------------------------------------------------------
+
+def _make_slot(db, seed, slot_date, plan_id=None):
+    """Create and return a real TimeSlot."""
+    slot = TimeSlot(
+        user_id=seed["admin_user"].id,
+        date=slot_date,
+        slot_type=SlotType.WORK,
+        estimated_duration=30,
+        scheduled_at=time(9, 0),
+        status=SlotStatus.NOT_STARTED,
+        plan_id=plan_id,
+    )
+    db.add(slot)
+    db.flush()
+    return slot
+
+
+# ---------------------------------------------------------------------------
+# Real slot: edit
+# ---------------------------------------------------------------------------
+
+class TestGuardEditRealSlot:
+    def test_past_slot_raises(self, db, seed):
+        slot = _make_slot(db, seed, YESTERDAY)
+        db.commit()
+        with pytest.raises(ImmutableSlotError, match="Cannot edit"):
+            guard_edit_real_slot(db, slot, today=TODAY)
+
+    def test_today_slot_allowed(self, db, seed):
+        slot = _make_slot(db, seed, TODAY)
+        db.commit()
+        # Should not raise
+        guard_edit_real_slot(db, slot, today=TODAY)
+
+    def test_future_slot_allowed(self, db, seed):
+        slot = _make_slot(db, seed, TOMORROW)
+        db.commit()
+        guard_edit_real_slot(db, slot, today=TODAY)
+
+
+# ---------------------------------------------------------------------------
+# Real slot: cancel
+# ---------------------------------------------------------------------------
+
+class TestGuardCancelRealSlot:
+    def test_past_slot_raises(self, db, seed):
+        slot = _make_slot(db, seed, YESTERDAY)
+        db.commit()
+        with pytest.raises(ImmutableSlotError, match="Cannot cancel"):
+            guard_cancel_real_slot(db, slot, today=TODAY)
+
+    def test_today_slot_allowed(self, db, seed):
+        slot = _make_slot(db, seed, TODAY)
+        db.commit()
+        guard_cancel_real_slot(db, slot, today=TODAY)
+
+    def test_future_slot_allowed(self, db, seed):
+        slot = _make_slot(db, seed, TOMORROW)
+        db.commit()
+        guard_cancel_real_slot(db, slot, today=TODAY)
+
+
+# ---------------------------------------------------------------------------
+# Virtual slot: edit
+# ---------------------------------------------------------------------------
+
+class TestGuardEditVirtualSlot:
+    def test_past_virtual_raises(self):
+        vid = make_virtual_slot_id(1, YESTERDAY)
+        with pytest.raises(ImmutableSlotError, match="Cannot edit"):
+            guard_edit_virtual_slot(vid, today=TODAY)
+
+    def test_today_virtual_allowed(self):
+        vid = make_virtual_slot_id(1, TODAY)
+        guard_edit_virtual_slot(vid, today=TODAY)
+
+    def test_future_virtual_allowed(self):
+        vid = make_virtual_slot_id(1, TOMORROW)
+        guard_edit_virtual_slot(vid, today=TODAY)
+
+    def test_invalid_virtual_id_raises_value_error(self):
+        with pytest.raises(ValueError, match="Invalid virtual slot id"):
+            guard_edit_virtual_slot("bad-id", today=TODAY)
+
+
+# ---------------------------------------------------------------------------
+# Virtual slot: cancel
+# ---------------------------------------------------------------------------
+
+class TestGuardCancelVirtualSlot:
+    def test_past_virtual_raises(self):
+        vid = make_virtual_slot_id(1, YESTERDAY)
+        with pytest.raises(ImmutableSlotError, match="Cannot cancel"):
+            guard_cancel_virtual_slot(vid, today=TODAY)
+
+    def test_today_virtual_allowed(self):
+        vid = make_virtual_slot_id(1, TODAY)
+        guard_cancel_virtual_slot(vid, today=TODAY)
+
+    def test_future_virtual_allowed(self):
+        vid = make_virtual_slot_id(1, TOMORROW)
+        guard_cancel_virtual_slot(vid, today=TODAY)
+
+
+# ---------------------------------------------------------------------------
+# Plan retroaction: past materialized slots are protected
+# ---------------------------------------------------------------------------
+
+class TestPlanNoRetroaction:
+    def _make_plan_with_slots(self, db, seed):
+        """Create a plan with materialized slots in the past, today, and future."""
+        user_id = seed["admin_user"].id
+        plan = SchedulePlan(
+            user_id=user_id,
+            slot_type=SlotType.WORK,
+            estimated_duration=30,
+            at_time=time(9, 0),
+            is_active=True,
+        )
+        db.add(plan)
+        db.flush()
+
+        past_slot = _make_slot(db, seed, LAST_WEEK, plan_id=plan.id)
+        yesterday_slot = _make_slot(db, seed, YESTERDAY, plan_id=plan.id)
+        today_slot = _make_slot(db, seed, TODAY, plan_id=plan.id)
+        future_slot = _make_slot(db, seed, TOMORROW, plan_id=plan.id)
+        db.commit()
+        return plan, past_slot, yesterday_slot, today_slot, future_slot
+
+    def test_get_past_materialized_slot_ids(self, db, seed):
+        plan, past_slot, yesterday_slot, today_slot, future_slot = (
+            self._make_plan_with_slots(db, seed)
+        )
+        past_ids = get_past_materialized_slot_ids(db, plan.id, today=TODAY)
+        assert set(past_ids) == {past_slot.id, yesterday_slot.id}
+        assert today_slot.id not in past_ids
+        assert future_slot.id not in past_ids
+
+    def test_guard_plan_edit_returns_protected_ids(self, db, seed):
+        plan, past_slot, yesterday_slot, _, _ = (
+            self._make_plan_with_slots(db, seed)
+        )
+        protected = guard_plan_edit_no_past_retroaction(db, plan.id, today=TODAY)
+        assert set(protected) == {past_slot.id, yesterday_slot.id}
+
+    def test_guard_plan_cancel_returns_protected_ids(self, db, seed):
+        plan, past_slot, yesterday_slot, _, _ = (
+            self._make_plan_with_slots(db, seed)
+        )
+        protected = guard_plan_cancel_no_past_retroaction(db, plan.id, today=TODAY)
+        assert set(protected) == {past_slot.id, yesterday_slot.id}
+
+    def test_no_past_slots_returns_empty(self, db, seed):
+        """If all materialized slots are today or later, no past IDs returned."""
+        user_id = seed["admin_user"].id
+        plan = SchedulePlan(
+            user_id=user_id,
+            slot_type=SlotType.WORK,
+            estimated_duration=30,
+            at_time=time(9, 0),
+            is_active=True,
+        )
+        db.add(plan)
+        db.flush()
+        _make_slot(db, seed, TODAY, plan_id=plan.id)
+        _make_slot(db, seed, TOMORROW, plan_id=plan.id)
+        db.commit()
+
+        past_ids = get_past_materialized_slot_ids(db, plan.id, today=TODAY)
+        assert past_ids == []
+
+
+# ---------------------------------------------------------------------------
+# ImmutableSlotError attributes
+# ---------------------------------------------------------------------------
+
+class TestImmutableSlotError:
+    def test_error_attributes(self):
+        err = ImmutableSlotError(YESTERDAY, "edit", detail="test detail")
+        assert err.slot_date == YESTERDAY
+        assert err.operation == "edit"
+        assert err.detail == "test detail"
+        assert "Cannot edit" in str(err)
+        assert "2026-03-30" in str(err)
+        assert "test detail" in str(err)
+
+    def test_error_without_detail(self):
+        err = ImmutableSlotError(YESTERDAY, "cancel")
+        assert "Cannot cancel" in str(err)
+        assert "test detail" not in str(err)