diff --git a/pyproject.toml b/pyproject.toml index d6366e0d..1c8254b0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ requires-python = ">=3.14" dependencies = [ "anyio", "authlib>=1.7.0", + "croniter==6.2.4", "fastapi>=0.140", "fastapi-jsonrpc>=4.0.0", "fastmcp>=4", @@ -20,6 +21,7 @@ dependencies = [ "pyyaml>=6.0.3", "referencing>=0.37", "typer>=0.24.2", + "tzdata>=2026.3", "uvicorn>=0.46.0", ] diff --git a/src/wf_scheduling/__init__.py b/src/wf_scheduling/__init__.py new file mode 100644 index 00000000..09efc149 --- /dev/null +++ b/src/wf_scheduling/__init__.py @@ -0,0 +1,8 @@ +"""Deployment scheduling package. + +Thin scheduling layer over the durable run lifecycle: calendar iteration +(croniter owns DST resolution), occurrence identities, schedule stores, +polling, recovery, ownership, and the opt-in server lifecycle. +""" + +__all__: list[str] = [] diff --git a/src/wf_scheduling/calendar.py b/src/wf_scheduling/calendar.py new file mode 100644 index 00000000..15636eb0 --- /dev/null +++ b/src/wf_scheduling/calendar.py @@ -0,0 +1,160 @@ +"""Thin croniter occurrence-source adapter. + +croniter owns calendar behavior, including DST resolution for nonexistent +and repeated local times. The adapter converts the query instant into the +schedule's named zone, asks croniter for the next/previous occurrence, and +converts the result to UTC. It applies no calendar correction of its own. + +Iteration is exclusive in both directions (a query from exactly a due +instant returns the neighboring occurrence), so callers query forward from +the last-consumed instant and compare catch-up results against the same +watermark: a due occurrence is admitted exactly once. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Protocol +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from croniter import CroniterBadCronError, CroniterBadDateError +from croniter import croniter as _Croniter + +UTC = timezone.utc + + +class InvalidScheduleDefinitionError(ValueError): + """A schedule definition is rejected (bad cron, bad zone, naive time).""" + + +class ScheduleExhaustedError(RuntimeError): + """Calendar search exhausted: an impossible schedule with no occurrence.""" + + +class OccurrenceSource(Protocol): + """Due-instant source over aware datetimes, UTC at the boundary.""" + + def next_after(self, instant: datetime) -> datetime | None: + """Return the first occurrence strictly after ``instant``.""" + ... + + def prev_before(self, instant: datetime) -> datetime | None: + """Return the latest occurrence strictly before ``instant``.""" + ... + + +def _require_aware(instant: datetime, *, op: str) -> datetime: + if instant.tzinfo is None or instant.utcoffset() is None: + raise ValueError(f"{op} requires an aware datetime with an offset") + return instant + + +def _ensure_utc_progress( + result_utc: datetime, bound_utc: datetime, *, forward: bool, op: str +) -> None: + if forward and not result_utc > bound_utc: + raise RuntimeError(f"{op} produced a non-progressing result") + if not forward and not result_utc < bound_utc: + raise RuntimeError(f"{op} produced a non-progressing result") + + +class CronSource: + """Five-field Unix-cron source in one explicit IANA time zone. + + Dialect: numeric ``0`` and ``7`` both mean Sunday (not Monday-first), + and day-of-month/day-of-week matching uses croniter's standard + ``day_or=True`` (Unix OR), exposed explicitly. + """ + + def __init__(self, expression: str, timezone_name: str) -> None: + fields = expression.split() + if len(fields) != 5: + raise InvalidScheduleDefinitionError( + f"cron expression must have exactly 5 fields, got {len(fields)}: " + f"{expression!r}" + ) + try: + zone = ZoneInfo(timezone_name) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise InvalidScheduleDefinitionError( + f"invalid time zone {timezone_name!r}" + ) from exc + try: + _Croniter(expression, datetime(2026, 1, 1, tzinfo=zone), day_or=True) + except CroniterBadCronError as exc: + raise InvalidScheduleDefinitionError( + f"invalid cron expression {expression!r}: {exc}" + ) from exc + self._expression = expression + self._timezone_name = timezone_name + self._zone = zone + + @property + def expression(self) -> str: + """Return the five-field cron expression.""" + return self._expression + + @property + def timezone_name(self) -> str: + """Return the IANA time-zone name.""" + return self._timezone_name + + def next_after(self, instant: datetime) -> datetime | None: + """Return the first occurrence strictly after ``instant`` in UTC.""" + _require_aware(instant, op="next_after") + zoned = instant.astimezone(self._zone) + try: + iterator = _Croniter(self._expression, zoned, day_or=True) + result = iterator.get_next(datetime) + except CroniterBadDateError as exc: + raise ScheduleExhaustedError( + f"cron schedule {self._expression!r} is exhausted: {exc}" + ) from exc + as_utc = result.astimezone(UTC) + _ensure_utc_progress( + as_utc, instant.astimezone(UTC), forward=True, op="next_after" + ) + return as_utc + + def prev_before(self, instant: datetime) -> datetime | None: + """Return the latest occurrence strictly before ``instant`` in UTC.""" + _require_aware(instant, op="prev_before") + zoned = instant.astimezone(self._zone) + try: + iterator = _Croniter(self._expression, zoned, day_or=True) + result = iterator.get_prev(datetime) + except CroniterBadDateError as exc: + raise ScheduleExhaustedError( + f"cron schedule {self._expression!r} is exhausted: {exc}" + ) from exc + as_utc = result.astimezone(UTC) + _ensure_utc_progress( + as_utc, instant.astimezone(UTC), forward=False, op="prev_before" + ) + return as_utc + + +class OneShotSource: + """Single UTC instant source; naive timestamps are rejected.""" + + def __init__(self, at: datetime) -> None: + if at.tzinfo is None or at.utcoffset() is None: + raise InvalidScheduleDefinitionError( + "OneShotSource requires an aware datetime with an offset" + ) + self._at = at.astimezone(UTC) + + @property + def at(self) -> datetime: + """Return the one-shot instant in UTC.""" + return self._at + + def next_after(self, instant: datetime) -> datetime | None: + """Return the instant if strictly after, else None.""" + _require_aware(instant, op="next_after") + return self._at if instant.astimezone(UTC) < self._at else None + + def prev_before(self, instant: datetime) -> datetime | None: + """Return the instant if strictly before, else None.""" + _require_aware(instant, op="prev_before") + return self._at if instant.astimezone(UTC) > self._at else None diff --git a/tests/scheduling/__init__.py b/tests/scheduling/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/scheduling/test_calendar_adapter.py b/tests/scheduling/test_calendar_adapter.py new file mode 100644 index 00000000..7e90e023 --- /dev/null +++ b/tests/scheduling/test_calendar_adapter.py @@ -0,0 +1,300 @@ +"""Port of the calendar-probe Part A contract to the thin adapter. + +Policy: croniter owns calendar calculation, including DST resolution. +The adapter converts now into the schedule zone, queries, converts back +to UTC, and applies no correction of its own. See +docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md and the +implementation plan Gate 2. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from importlib.metadata import version as _pkg_version +from zoneinfo import ZoneInfo + +import pytest + +from wf_scheduling.calendar import ( + CronSource, + InvalidScheduleDefinitionError, + OneShotSource, + ScheduleExhaustedError, +) + +UTC = timezone.utc +HCMC = ZoneInfo("Asia/Ho_Chi_Minh") +NYC = ZoneInfo("America/New_York") + + +def utc(*args: int) -> datetime: + return datetime(*args, tzinfo=UTC) + + +def test_versions_pinned() -> None: + assert _pkg_version("croniter") == "6.2.4" + + +def test_utc_daily_next_is_strictly_increasing_and_unique() -> None: + src = CronSource("30 9 * * *", "UTC") + seen: set[datetime] = set() + prev: datetime | None = None + cursor = utc(2026, 9, 1, 0, 0, 0) + for _ in range(10): + nxt = src.next_after(cursor) + assert nxt is not None + assert nxt.tzinfo is not None + assert nxt.utcoffset() == timedelta(0) + assert nxt not in seen + seen.add(nxt) + if prev is not None: + assert nxt > prev + prev = nxt + cursor = nxt + assert len(seen) == 10 + + +def test_hcmc_daily_converts_and_hourly_counts() -> None: + src = CronSource("30 9 * * *", "Asia/Ho_Chi_Minh") + nxt = src.next_after(datetime(2026, 9, 1, 0, 0, tzinfo=HCMC)) + assert nxt is not None + assert (nxt.hour, nxt.minute) == (2, 30) + assert nxt.date().isoformat() == "2026-09-01" + for _label, start in ( + ("march", utc(2026, 3, 7, 17, 30, 0)), + ("november", utc(2026, 10, 31, 17, 30, 0)), + ): + hourly = CronSource("0 * * * *", "Asia/Ho_Chi_Minh") + cursor = start + count = 0 + while True: + hit = hourly.next_after(cursor) + assert hit is not None + if hit >= start + timedelta(hours=48): + break + count += 1 + assert count < 60 + cursor = hit + assert count == 48 + + +def test_expression_forms_wildcard_step_list_range() -> None: + cases = { + "* * * * *": (datetime(2026, 9, 8, 0, 0, tzinfo=UTC), "2026-09-08T00:01:00"), + "5/15 * * * *": ( + datetime(2026, 9, 8, 0, 0, tzinfo=UTC), + "2026-09-08T00:05:00", + ), + "0,30 1-2 * * *": ( + datetime(2026, 9, 8, 0, 0, tzinfo=UTC), + "2026-09-08T01:00:00", + ), + "*/20 1-3 * * *": ( + datetime(2026, 9, 8, 1, 50, tzinfo=UTC), + "2026-09-08T02:00:00", + ), + } + for expr, (start, want) in cases.items(): + src = CronSource(expr, "UTC") + nxt = src.next_after(start) + assert nxt is not None + assert nxt.isoformat() == want + "+00:00" + + +def test_weekday_names_and_numbers_unix() -> None: + monday = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + cases = { + "0 12 * * 0": 6, + "0 12 * * 7": 6, + "0 12 * * 1": 0, + "0 12 * * mon": 0, + "0 12 * * sun": 6, + } + for expr, want_wd in cases.items(): + nxt = CronSource(expr, "UTC").next_after(monday) + assert nxt is not None + assert nxt.weekday() == want_wd, expr + assert CronSource("0 12 * * 0", "UTC").next_after(monday) == CronSource( + "0 12 * * sun", "UTC" + ).next_after(monday) + + +def test_dom_dow_day_or_true_is_selected() -> None: + start = datetime(2026, 9, 1, 0, 0, tzinfo=UTC) + src = CronSource("0 12 13 * fri", "UTC") + hits: list[str] = [] + cursor = start + for _ in range(4): + nxt = src.next_after(cursor) + assert nxt is not None + hits.append(nxt.date().isoformat()) + cursor = nxt + assert hits == ["2026-09-04", "2026-09-11", "2026-09-13", "2026-09-18"] + + +def test_zone_conversion_is_the_adapters_job() -> None: + src = CronSource("30 9 * * *", "Asia/Ho_Chi_Minh") + nxt = src.next_after(datetime(2026, 9, 1, 0, 0, tzinfo=HCMC)) + assert nxt is not None + assert nxt.tzinfo is not None + assert nxt.utcoffset() == timedelta(0) + + +def test_gap_nonexistent_wall_times_resolve_forward() -> None: + src = CronSource("30 2 * * *", "America/New_York") + first = src.next_after(datetime(2026, 3, 7, 12, 0, tzinfo=NYC)) + assert first is not None + assert first.isoformat() == "2026-03-08T07:00:00+00:00" + following = src.next_after(first) + assert following is not None + assert following.isoformat() == "2026-03-09T06:30:00+00:00" + + +def test_gap_per_minute_stream_jumps_forward() -> None: + src = CronSource("* * * * *", "America/New_York") + cursor = datetime(2026, 3, 8, 0, 0, tzinfo=NYC) + seq: list[datetime] = [] + for _ in range(300): + nxt = src.next_after(cursor) + assert nxt is not None + seq.append(nxt) + cursor = nxt + assert all(b > a for a, b in zip(seq, seq[1:])) + assert len(set(seq)) == len(seq) + gap_day = [d for d in seq if d.astimezone(NYC).date().isoformat() == "2026-03-08"] + assert all(d.astimezone(NYC).hour != 2 for d in gap_day) + + +def test_gap_backward_queries_return_library_resolution() -> None: + src = CronSource("30 2 * * *", "America/New_York") + cases = { + "2026-03-08T12:00": "2026-03-08T07:00:00+00:00", + "2026-03-09T00:00": "2026-03-08T07:00:00+00:00", + "2026-03-07T12:00": "2026-03-07T07:30:00+00:00", + "2026-03-09T12:00": "2026-03-09T06:30:00+00:00", + } + for start_iso, want in cases.items(): + start = datetime.fromisoformat(start_iso).replace(tzinfo=NYC) + got = src.prev_before(start) + assert got is not None + assert got.isoformat() == want, start_iso + + +def test_fold_repeated_times_are_distinct_utc() -> None: + src = CronSource("30 1 * * *", "America/New_York") + first = src.next_after(datetime(2026, 10, 31, 12, 0, tzinfo=NYC)) + assert first is not None + assert first == utc(2026, 11, 1, 5, 30) + assert src.next_after(first) == utc(2026, 11, 1, 6, 30) + + +def test_fold_per_minute_unique_increasing() -> None: + src = CronSource("* * * * *", "America/New_York") + cursor = datetime(2026, 10, 31, 20, 0, tzinfo=NYC) + seq: list[datetime] = [] + for _ in range(560): + nxt = src.next_after(cursor) + assert nxt is not None + seq.append(nxt) + cursor = nxt + assert all(b > a for a, b in zip(seq, seq[1:])) + assert len(set(seq)) == len(seq) + iso = {d.isoformat() for d in seq} + assert "2026-11-01T05:30:00+00:00" in iso + assert "2026-11-01T06:30:00+00:00" in iso + + +def test_latest_missed_after_years_of_downtime_is_bounded() -> None: + import time + + src = CronSource("* * * * *", "UTC") + now = utc(2026, 9, 8, 12, 0, 0) + t0 = time.perf_counter() + latest = src.prev_before(now) + elapsed = time.perf_counter() - t0 + assert latest is not None + assert elapsed < 5 + assert latest <= now + assert (now - latest) < timedelta(minutes=2) + nyc_src = CronSource("* * * * *", "America/New_York") + latest_nyc = nyc_src.prev_before(datetime(2026, 9, 8, 12, 0, tzinfo=NYC)) + assert latest_nyc is not None + assert latest_nyc.isoformat() == "2026-09-08T15:59:00+00:00" + + +def test_iteration_is_exclusive_both_directions() -> None: + # Exclusive get_next/get_prev alone does not include an occurrence equal + # to now: the adapter AND scheduler must handle exact due-time inclusion + # via the last-consumed watermark, not by assuming inclusivity. + src = CronSource("0 13 * * *", "UTC") + due = utc(2026, 9, 8, 13, 0, 0) + fwd = src.next_after(due) + back = src.prev_before(due) + assert fwd == utc(2026, 9, 9, 13, 0, 0) + assert back is not None + assert (back.year, back.month, back.day, back.hour, back.minute) == ( + 2026, + 9, + 7, + 13, + 0, + ) + just_before = src.next_after(due - timedelta(seconds=1)) + assert just_before == due + # Exact due-time inclusion through the adapter: prev_before just after due + # returns the due instant itself. + just_after = src.prev_before(due + timedelta(seconds=1)) + assert just_after == due + + +def test_exact_due_time_inclusion_through_scheduler_watermark() -> None: + # A scheduler querying forward from the last-consumed instant and comparing + # catch-up results against the same watermark admits a due occurrence + # exactly once: forward from consumed (exclusive) finds due, and due is + # not <= consumed so it is eligible. Querying forward from due itself + # must not return due again. + src = CronSource("0 13 * * *", "UTC") + due = utc(2026, 9, 8, 13, 0, 0) + consumed = utc(2026, 9, 7, 13, 0, 0) + candidate = src.next_after(consumed) + assert candidate == due + assert candidate is not None and candidate > consumed + assert src.next_after(due) != due + + +def test_impossible_schedule_maps_to_exhausted() -> None: + import time + + src = CronSource("0 12 30 2 *", "UTC") + start = time.perf_counter() + with pytest.raises(ScheduleExhaustedError): + src.next_after(utc(2026, 1, 1)) + assert time.perf_counter() - start < 5 + + +def test_bad_expressions_rejected_naive_rejected() -> None: + with pytest.raises(InvalidScheduleDefinitionError): + CronSource("nonsense", "UTC") + with pytest.raises(InvalidScheduleDefinitionError): + CronSource("* * * *", "UTC") + with pytest.raises(InvalidScheduleDefinitionError): + CronSource("* * * * * *", "UTC") + with pytest.raises(InvalidScheduleDefinitionError): + CronSource("* * * * *", "No/Such_Zone") + with pytest.raises(ValueError, match="aware"): + CronSource("* * * * *", "UTC").next_after(datetime(2026, 9, 8, 12, 0)) + with pytest.raises(ValueError, match="aware"): + CronSource("* * * * *", "UTC").prev_before(datetime(2026, 9, 8, 12, 0)) + + +def test_oneshot_requires_offset_and_returns_utc() -> None: + at = datetime(2026, 9, 8, 12, 0, tzinfo=timezone(timedelta(hours=2))) + src = OneShotSource(at) + assert src.next_after(utc(2026, 9, 8, 9, 0)) == utc(2026, 9, 8, 10, 0) + assert src.next_after(utc(2026, 9, 8, 10, 0)) is None + assert src.prev_before(utc(2026, 9, 8, 11, 0)) == utc(2026, 9, 8, 10, 0) + assert src.prev_before(utc(2026, 9, 8, 10, 0)) is None + with pytest.raises(InvalidScheduleDefinitionError): + OneShotSource(datetime(2026, 9, 8, 12, 0)) + with pytest.raises(ValueError, match="aware"): + src.next_after(datetime(2026, 9, 8, 12, 0)) diff --git a/uv.lock b/uv.lock index f11eca57..20daf1b4 100644 --- a/uv.lock +++ b/uv.lock @@ -235,6 +235,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ae/8c/469afb6465b853afff216f9528ffda78a915ff880ed58813ba4faf4ba0b6/contourpy-1.3.3-cp314-cp314t-win_arm64.whl", hash = "sha256:b7448cb5a725bb1e35ce88771b86fba35ef418952474492cf7c764059933ff8b", size = 203831, upload-time = "2025-07-26T12:02:51.449Z" }, ] +[[package]] +name = "croniter" +version = "6.2.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/37/57/2e2a65aee2a70483cb28e2b7e15a072d00a523207593b44400d4717bb100/croniter-6.2.4.tar.gz", hash = "sha256:fc124f751b1b04805c2a04b061898b436b45ab2320b045e1e052ea895de65189", size = 166267, upload-time = "2026-07-10T09:52:59.955Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cd/ba/d678e5bd329646ca51d3c92addbc77804e86d21f4b6b6a027218e6abb010/croniter-6.2.4-py3-none-any.whl", hash = "sha256:8ef3d544107a5c05a150a2d78f8bf5a8eb9c5c4d93405a736b824109574e3f4d", size = 46677, upload-time = "2026-07-10T09:52:58.425Z" }, +] + [[package]] name = "cryptography" version = "50.0.1" @@ -814,6 +826,7 @@ source = { editable = "." } dependencies = [ { name = "anyio" }, { name = "authlib" }, + { name = "croniter" }, { name = "fastapi" }, { name = "fastapi-jsonrpc" }, { name = "fastmcp" }, @@ -826,6 +839,7 @@ dependencies = [ { name = "pyyaml" }, { name = "referencing" }, { name = "typer" }, + { name = "tzdata" }, { name = "uvicorn" }, ] @@ -843,6 +857,7 @@ dev = [ requires-dist = [ { name = "anyio" }, { name = "authlib", specifier = ">=1.7.0" }, + { name = "croniter", specifier = "==6.2.4" }, { name = "fastapi", specifier = ">=0.140" }, { name = "fastapi-jsonrpc", specifier = ">=4.0.0" }, { name = "fastmcp", specifier = ">=4" }, @@ -855,6 +870,7 @@ requires-dist = [ { name = "pyyaml", specifier = ">=6.0.3" }, { name = "referencing", specifier = ">=0.37" }, { name = "typer", specifier = ">=0.24.2" }, + { name = "tzdata", specifier = ">=2026.3" }, { name = "uvicorn", specifier = ">=0.46.0" }, ] @@ -1750,6 +1766,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" }, ] +[[package]] +name = "tzdata" +version = "2026.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/92/ff/5a28bdfd8c3ebec42564ac7d0e54ca3db65044a9314a97f9564fa7a1e926/tzdata-2026.3.tar.gz", hash = "sha256:4a1518b8993086a7982523e071643f3c0e5f213e75b21318e78bcabfff9d1415", size = 198674, upload-time = "2026-07-10T08:50:37.887Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/6d/b53b99a9f2766d095985947a5782f1702cabb129a34f7a802d7197af832f/tzdata-2026.3-py2.py3-none-any.whl", hash = "sha256:dc096730c87af6cab1b171c9d532be840741ff5d459015e7f6947bd7d7e54931", size = 348168, upload-time = "2026-07-10T08:50:36.46Z" }, +] + [[package]] name = "uncalled-for" version = "0.4.0"