Add Cron dependency for cron-style task scheduling

Introduces a new Cron dependency that extends Perpetual to support
wall-clock scheduled tasks using cron expressions. Unlike Perpetual
which uses relative intervals, Cron schedules tasks at exact times
(e.g., "0 9 * * 1" for Mondays at 9 AM).

Key changes:
- Add Cron class with croniter integration for expression parsing
- Support standard 5-field cron syntax and Vixie keywords (@daily, etc.)
- Automatic scheduling at worker startup (automatic=True by default)
- Fix Perpetual.__aenter__ to preserve the automatic flag

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: Omar Abdelkader

pyproject.toml

@@ -24,6 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "cloudpickle>=3.1.1",
+    "croniter>=6",
     "exceptiongroup>=1.2.0; python_version < '3.11'",
     "taskgroup>=0.2.2; python_version < '3.11'",
     "fakeredis[lua]>=2.32.1",
@@ -69,6 +70,7 @@ dev = [
     "pytest-timeout>=2.4.0",
     "pytest-xdist>=3.6.1",
     "ruff>=0.14.14",
+    "types-croniter>=6",
 ]
 
 docs = [

src/docket/__init__.py

@@ -12,6 +12,7 @@
 from .annotations import Logged
 from .dependencies import (
     ConcurrencyLimit,
+    Cron,
     CurrentDocket,
     CurrentExecution,
     CurrentWorker,
@@ -36,6 +37,7 @@
     "__version__",
     "Agenda",
     "ConcurrencyLimit",
+    "Cron",
     "CurrentDocket",
     "CurrentExecution",
     "CurrentWorker",

src/docket/dependencies/__init__.py

@@ -16,6 +16,7 @@
     format_duration,
 )
 from ._concurrency import ConcurrencyBlocked, ConcurrencyLimit
+from ._cron import Cron
 from ._contextual import (
     CurrentDocket,
     CurrentExecution,
@@ -74,6 +75,7 @@
     "AdmissionBlocked",
     "ConcurrencyBlocked",
     "ConcurrencyLimit",
+    "Cron",
     "Perpetual",
     "Progress",
     "Timeout",

src/docket/dependencies/_cron.py

@@ -0,0 +1,86 @@
+"""Cron-style scheduling dependency."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from croniter import croniter
+
+from ._perpetual import Perpetual
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ._base import TaskOutcome
+    from ..execution import Execution
+
+
+class Cron(Perpetual):
+    """Declare a task that should run on a cron schedule. Cron tasks are automatically
+    rescheduled for the next matching time after they finish (whether they succeed or
+    fail). By default, a cron task is scheduled at worker startup with `automatic=True`.
+
+    Unlike `Perpetual` which schedules based on intervals from the current time, `Cron`
+    schedules based on wall-clock time, ensuring tasks run at consistent times regardless
+    of execution duration or delays.
+
+    Supports standard cron expressions and Vixie cron-style keywords (@daily, @hourly, etc.)
+    via the croniter library.
+
+    Example:
+
+    ```python
+    @task
+    async def weekly_report(cron: Cron = Cron("0 9 * * 1")) -> None:
+        # Runs every Monday at 9:00 AM UTC
+        ...
+
+    @task
+    async def daily_cleanup(cron: Cron = Cron("@daily")) -> None:
+        # Runs every day at midnight UTC
+        ...
+    ```
+    """
+
+    expression: str
+
+    _croniter: croniter[datetime]
+
+    def __init__(
+        self,
+        expression: str,
+        automatic: bool = True,
+    ) -> None:
+        """
+        Args:
+            expression: A cron expression string. Supports:
+                - Standard 5-field syntax: "minute hour day month weekday"
+                  (e.g., "0 9 * * 1" for Mondays at 9 AM)
+                - Vixie cron keywords: @yearly, @annually, @monthly, @weekly,
+                  @daily, @midnight, @hourly
+            automatic: If set, this task will be automatically scheduled during worker
+                startup and continually through the worker's lifespan. This ensures
+                that the task will always be scheduled despite crashes and other
+                adverse conditions. Automatic tasks must not require any arguments.
+        """
+        super().__init__(automatic=automatic)
+        self.expression = expression
+        self._croniter = croniter(expression, datetime.now(timezone.utc), datetime)
+
+    async def __aenter__(self) -> Cron:
+        execution = self.execution.get()
+        cron = Cron(expression=self.expression, automatic=self.automatic)
+        cron.args = execution.args
+        cron.kwargs = execution.kwargs
+        return cron
+
+    def get_next(self) -> datetime:
+        return self._croniter.get_next()
+
+    async def on_complete(self, execution: Execution, outcome: TaskOutcome) -> bool:
+        """Handle completion by scheduling the next execution at the exact cron time.
+
+        This overrides Perpetual's on_complete to ensure we hit the exact wall-clock
+        time rather than adjusting for task duration.
+        """
+        self.at(self.get_next())
+        return await super().on_complete(execution, outcome)

src/docket/dependencies/_perpetual.py

@@ -61,7 +61,7 @@ def __init__(
 
     async def __aenter__(self) -> Perpetual:
         execution = self.execution.get()
-        perpetual = Perpetual(every=self.every)
+        perpetual = Perpetual(every=self.every, automatic=self.automatic)
         perpetual.args = execution.args
         perpetual.kwargs = execution.kwargs
         return perpetual

src/docket/worker.py

@@ -42,6 +42,7 @@
 from .dependencies import (
     AdmissionBlocked,
     CompletionHandler,
+    Cron,
     CurrentExecution,
     Dependency,
     FailedDependency,
@@ -763,15 +764,14 @@ async def _schedule_all_automatic_perpetual_tasks(self) -> None:
                         perpetual = get_single_dependency_parameter_of_type(
                             task_function, Perpetual
                         )
-                        if perpetual is None:
-                            continue
-
-                        if not perpetual.automatic:
-                            continue
-
-                        key = task_function.__name__
-
-                        await self.docket.add(task_function, key=key)()
+                        if perpetual is not None and perpetual.automatic:
+                            key = task_function.__name__
+                            when = (
+                                perpetual.get_next()
+                                if isinstance(perpetual, Cron)
+                                else None
+                            )
+                            await self.docket.add(task_function, when=when, key=key)()
             except LockError:  # pragma: no cover
                 return
 

tests/fundamentals/test_cron.py

@@ -0,0 +1,119 @@
+"""Tests for Cron dependency (cron-style scheduled tasks)."""
+
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from docket import Docket, Worker
+from docket.dependencies import Cron
+
+
+async def test_cron_task_reschedules_itself(docket: Docket, worker: Worker):
+    """Cron tasks automatically reschedule after each execution."""
+    runs = 0
+
+    async def my_cron_task(cron: Cron = Cron("0 9 * * *", automatic=False)):
+        nonlocal runs
+        runs += 1
+
+    # Patch get_next to return a time 10ms in the future (instead of waiting for 9 AM)
+    with patch.object(
+        Cron,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(my_cron_task)()
+        await worker.run_at_most({execution.key: 3})
+
+    assert runs == 3
+
+
+async def test_cron_tasks_are_automatically_scheduled(docket: Docket, worker: Worker):
+    """Cron tasks with automatic=True are scheduled at worker startup."""
+    calls = 0
+
+    async def my_automatic_cron(
+        cron: Cron = Cron("0 0 * * *"),
+    ):  # automatic=True is default
+        nonlocal calls
+        calls += 1
+
+    docket.register(my_automatic_cron)
+
+    with patch.object(
+        Cron,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        await worker.run_at_most({"my_automatic_cron": 2})
+
+    assert calls == 2
+
+
+async def test_cron_tasks_continue_after_errors(docket: Docket, worker: Worker):
+    """Cron tasks keep rescheduling even when they raise exceptions."""
+    calls = 0
+
+    async def flaky_cron_task(cron: Cron = Cron("0 * * * *", automatic=False)):
+        nonlocal calls
+        calls += 1
+        raise ValueError("Task failed!")
+
+    with patch.object(
+        Cron,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(flaky_cron_task)()
+        await worker.run_at_most({execution.key: 3})
+
+    assert calls == 3
+
+
+async def test_cron_tasks_can_cancel_themselves(docket: Docket, worker: Worker):
+    """A cron task can stop rescheduling by calling cron.cancel()."""
+    calls = 0
+
+    async def limited_cron_task(cron: Cron = Cron("0 * * * *", automatic=False)):
+        nonlocal calls
+        calls += 1
+        if calls >= 3:
+            cron.cancel()
+
+    with patch.object(
+        Cron,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        await docket.add(limited_cron_task)()
+        await worker.run_until_finished()
+
+    assert calls == 3
+
+
+async def test_cron_supports_vixie_keywords(docket: Docket, worker: Worker):
+    """Cron supports Vixie cron keywords like @daily, @weekly, @hourly."""
+    runs = 0
+
+    # @daily is equivalent to "0 0 * * *" (midnight every day)
+    async def daily_task(cron: Cron = Cron("@daily", automatic=False)):
+        nonlocal runs
+        runs += 1
+
+    with patch.object(
+        Cron,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(daily_task)()
+        await worker.run_at_most({execution.key: 1})
+
+    assert runs == 1
+
+
+def test_cron_get_next_returns_future_time():
+    """Cron.get_next() returns a datetime in the future via croniter."""
+    cron = Cron("* * * * *", automatic=False)  # Every minute
+    next_time = cron.get_next()
+
+    assert isinstance(next_time, datetime)
+    assert next_time > datetime.now(timezone.utc)