feat: add monthly day-of-month recurrence with month-end clamp policy (#28)

This commit is contained in:
Joshua King
2026-09-14 11:12:33 -04:00
parent 23a52707af
commit 27e882f931
3 changed files with 345 additions and 5 deletions
+41 -2
View File
@@ -418,6 +418,39 @@ def _within_window(rule: ScheduleRule, target: datetime.date) -> bool:
return True
def _month_end(year: int, month: int) -> int:
"""The last calendar day of the given month (leap-aware)."""
if month == 12:
return 31
return (datetime.date(year, month + 1, 1)
- datetime.timedelta(days=1)).day
def _occurs_monthly_day(rule: ScheduleRule, target: datetime.date) -> bool:
"""MONTHLY_DAY evaluation with the documented month-end policy.
Policy: a rule for a day-of-month that does not exist in a given
month (the 31st in April, the 30th in February, the 29th of
February in a non-leap year) fires on the LAST day of that month
instead of being skipped — a monthly chore still happens monthly,
pinned to the month's end when its nominal date is absent.
The interval anchors on months ELAPSED from start_date's month: the
rule fires when the months between the anchor month and the target's
month form a multiple of the interval, keeping "every 2 months on
the 15th" aligned to the anchor month even when the firing lands on
a clamped month-end day.
"""
anchor = _parse_date(rule.start_date)
months_elapsed = (
(target.year - anchor.year) * 12 + (target.month - anchor.month)
)
if months_elapsed % rule.interval != 0:
return False
month_end = _month_end(target.year, target.month)
return target.day == min(rule.day_of_month, month_end)
def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
"""Return True when ``rule`` fires on ``target``.
@@ -428,7 +461,10 @@ def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
# Dispatch FIRST: an unimplemented shape must raise loudly even for
# dates outside the window, never silently evaluate to False.
shape = rule.rule_type
if shape not in (RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS):
if shape not in (
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
):
raise NotImplementedError(
f"occurs_on for {shape} lands with its own engine task "
"(Features 04 task sequence)"
@@ -438,6 +474,8 @@ def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
if shape is RuleType.DAILY:
offset = (target - _parse_date(rule.start_date)).days
return offset % rule.interval == 0
if shape is RuleType.MONTHLY_DAY:
return _occurs_monthly_day(rule, target)
# WEEKLY / CUSTOM_DAYS: whole-week anchoring from start_date, never
# ISO week numbers (Feature 04/05 guardrail: 53-week years flip ISO
# parity on January 1st; anchor arithmetic does not).
@@ -462,7 +500,8 @@ def occurrences_between(
rule.end_date] yields an empty list.
"""
if rule.rule_type not in (
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS
RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
RuleType.MONTHLY_DAY,
):
raise NotImplementedError(
f"occurrences_between for {rule.rule_type} lands with its own "