KEY INSIGHT
Cron expressions provide a human-readable way to define recurring schedules, enabling the agent to run tasks at specific times without continuous polling.
Scheduled tasks require a mechanism to determine when they should run. Cron expressions provide a standard format for describing time-based schedules. The agent's scheduler evaluates cron expressions to determine the next run time for each task.
Cron Expression Format
A cron expression consists of five fields separated by spaces: minute, hour, day of month, month, and day of week. Each field accepts specific values, ranges, and special characters.
```
┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sunday=0)
│ │ │ │ │
* * * * *
```
Common patterns include: `0 9 * * 1-5` for 9 AM on weekdays, `*/15 * * * *` for every 15 minutes, and `0 0 1 * *` for midnight on the first of each month.
Cron Parser Implementation
```python
import re
from datetime import datetime, timedelta
from typing import Callable, Awaitable
class CronField:
"""Parse and match a single cron field."""
def __init__(self, value: str, min_val: int, max_val: int):
self.value = value
self.min_val = min_val
self.max_val = max_val
self._parse()
def _parse(self) -> None:
"""Parse the field into a set of matching values."""
self.values: set[int] = set()
self._ranges: list[tuple[int, int]] = []
for part in self.value.split(","):
if "/" in part:
# Handle step values: */15 or 1-30/5
range_part, step = part.split("/")
step = int(step)
if range_part == "*":
start, end = self.min_val, self.max_val
else:
start, end = self._parse_range(range_part)
for v in range(start, end + 1, step):
self.values.add(v)
elif "-" in part:
# Handle ranges: 1-5
start, end = self._parse_range(part)
self._ranges.append((start, end))
for v in range(start, end + 1):
self.values.add(v)
elif part == "*":
# All values
for v in range(self.min_val, self.max_val + 1):
self.values.add(v)
else:
# Single value
self.values.add(int(part))
def matches(self, value: int) -> bool:
"""Check if a value matches this field."""
if value in self.values:
return True
return any(start <= value <= end for start, end in self._ranges)
class CronSchedule:
"""Parse and evaluate cron expressions."""
def __init__(self, expression: str):
parts = expression.split()
if len(parts) != 5:
raise ValueError(f"Invalid cron expression: {expression}")
self.minute = CronField(parts[0], 0, 59)
self.hour = CronField(parts[1], 0, 23)
self.day_of_month = CronField(parts[2], 1, 31)
self.month = CronField(parts[3], 1, 12)
self.day_of_week = CronField(parts[4], 0, 6)
def next_run(self, after: datetime) -> datetime:
"""Calculate the next run time after the given datetime."""
# Implementation iterates forward minute by minute
# until all fields match
current = after.replace(second=0, microsecond=0) + timedelta(minutes=1)
max_attempts = 366 * 24 * 60 # Max 1 year ahead
for _ in range(max_attempts):
if self._matches(current):
return current
current += timedelta(minutes=1)
raise ValueError("No matching time found within one year")
def _matches(self, dt: datetime) -> bool:
"""Check if a datetime matches the schedule."""
return (
self.minute.matches(dt.minute) and
self.hour.matches(dt.hour) and
self.day_of_month.matches(dt.day) and
self.month.matches(dt.month) and
self.day_of_week.matches(dt.weekday())
)
```