Cron Scheduling skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Cron Scheduling"
  short_description: "Scheduled jobs that run once, on time, visibly"
  default_prompt: "$cron-scheduling Add a nightly job that emails overdue invoice reminders at 8am Berlin time"

#!/usr/bin/env python3
"""Explain a 5-field cron expression: next run times in a time zone, plus DST hazards.

Usage:
  python3 cron_next.py "30 2 * * *" --tz Europe/London --count 5
  python3 cron_next.py "@hourly" --tz UTC --fail-on-dst

Supports *, lists (1,5), ranges (1-5), steps (*/15, 10-50/10), month and weekday names
(jan, mon), 7 = Sunday, and @yearly/@annually/@monthly/@weekly/@daily/@midnight/@hourly.
Standard cron semantics: when both day-of-month and day-of-week are restricted, a day
matches if EITHER matches.

Exit codes: 0 ok, 1 DST hazard found with --fail-on-dst, 2 invalid expression, zone,
or a schedule that never fires.
Standard library only (zoneinfo needs system tz data or the tzdata package).
"""
import argparse
import sys
from datetime import date, datetime, timedelta, timezone
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

MACROS = {"@yearly": "0 0 1 1 *", "@annually": "0 0 1 1 *", "@monthly": "0 0 1 * *",
          "@weekly": "0 0 * * 0", "@daily": "0 0 * * *", "@midnight": "0 0 * * *", "@hourly": "0 * * * *"}
MONTHS = {m: i for i, m in enumerate("jan feb mar apr may jun jul aug sep oct nov dec".split(), 1)}
DAYS = {d: i for i, d in enumerate("sun mon tue wed thu fri sat".split())}
FIELDS = [("minute", 0, 59, {}), ("hour", 0, 23, {}), ("day-of-month", 1, 31, {}),
          ("month", 1, 12, MONTHS), ("day-of-week", 0, 7, DAYS)]


def parse_field(text: str, name: str, lo: int, hi: int, names: dict) -> tuple[set[int], bool]:
    values: set[int] = set()
    for part in text.lower().split(","):
        step = 1
        if "/" in part:
            part, step_s = part.split("/", 1)
            if not step_s.isdigit() or int(step_s) == 0:
                raise ValueError(f"{name}: bad step '{step_s}'")
            step = int(step_s)
        if part == "*":
            start, end = lo, hi
        elif "-" in part:
            a, b = part.split("-", 1)
            start, end = conv(a, name, names), conv(b, name, names)
        else:
            start = conv(part, name, names)
            end = hi if step > 1 else start
        if not (lo <= start <= hi and lo <= end <= hi) or start > end:
            raise ValueError(f"{name}: '{part}' outside {lo}-{hi} or reversed")
        values.update(range(start, end + 1, step))
    return values, text != "*"


def conv(token: str, name: str, names: dict) -> int:
    if token in names:
        return names[token]
    if token.isdigit():
        return int(token)
    raise ValueError(f"{name}: cannot parse '{token}'")


def parse(expr: str):
    expr = MACROS.get(expr.strip().lower(), expr)
    parts = expr.split()
    if len(parts) != 5:
        raise ValueError(f"expected 5 fields (min hour dom month dow), got {len(parts)}: '{expr}'")
    parsed = [parse_field(p, *spec) for p, spec in zip(parts, FIELDS)]
    dow, dow_restricted = parsed[4]
    if 7 in dow:
        dow = (dow - {7}) | {0}
    return parsed[0][0], parsed[1][0], parsed[2], parsed[3][0], (dow, dow_restricted)


def day_matches(d: date, dom, months, dow) -> bool:
    if d.month not in months:
        return False
    dom_set, dom_r = dom
    dow_set, dow_r = dow
    in_dom = d.day in dom_set
    in_dow = (d.isoweekday() % 7) in dow_set
    if dom_r and dow_r:
        return in_dom or in_dow
    return in_dom and in_dow


def classify(naive: datetime, tz: ZoneInfo) -> str:
    """Return 'ok', 'gap' (wall time does not exist) or 'fold' (wall time occurs twice)."""
    a = naive.replace(tzinfo=tz, fold=0)
    b = naive.replace(tzinfo=tz, fold=1)
    if a.utcoffset() == b.utcoffset():
        return "ok"
    roundtrip = a.astimezone(timezone.utc).astimezone(tz).replace(tzinfo=None)
    return "gap" if roundtrip != naive else "fold"


def iterate(spec, tz: ZoneInfo, start: datetime, days: int):
    minutes, hours, dom, months, dow = spec
    d0 = start.date()
    for offset in range(days):
        d = d0 + timedelta(days=offset)
        if not day_matches(d, dom, months, dow):
            continue
        for h in sorted(hours):
            for m in sorted(minutes):
                naive = datetime(d.year, d.month, d.day, h, m)
                if naive <= start:
                    continue
                yield naive, classify(naive, tz)


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("expr")
    p.add_argument("--tz", default="UTC", help="IANA zone the scheduler evaluates cron in (default UTC)")
    p.add_argument("--count", type=int, default=5)
    p.add_argument("--from", dest="start", help="start time, ISO format, local to --tz (default now)")
    p.add_argument("--fail-on-dst", action="store_true", help="exit 1 if any run in the next year hits a DST gap or overlap")
    args = p.parse_args()
    try:
        tz = ZoneInfo(args.tz)
        spec = parse(args.expr)
    except (ValueError, ZoneInfoNotFoundError) as exc:
        print(f"error: {exc}")
        return 2
    start = datetime.fromisoformat(args.start) if args.start else datetime.now(tz).replace(tzinfo=None, second=0, microsecond=0)

    print(f"expr: {args.expr}   zone: {args.tz}")
    shown = 0
    for naive, kind in iterate(spec, tz, start, days=366 * 5):
        if kind == "gap":
            continue
        aware = naive.replace(tzinfo=tz, fold=0)
        utc = aware.astimezone(timezone.utc)
        note = "  (wall time occurs twice: may run twice)" if kind == "fold" else ""
        print(f"  {aware.isoformat()}  =  {utc.strftime('%Y-%m-%dT%H:%MZ')}{note}")
        shown += 1
        if shown >= args.count:
            break
    if shown == 0:
        print("  no runs in the next 5 years (impossible date such as Feb 30?)")
        return 2

    hazards = [(n, k) for n, k in iterate(spec, tz, start, days=366) if k != "ok"]
    if hazards:
        print("\nDST hazards in the next 12 months:")
        for naive, kind in hazards[:10]:
            what = "does not exist (skipped or shifted, scheduler-dependent)" if kind == "gap" else "is ambiguous (may run twice or once)"
            print(f"  {naive.isoformat(timespec='minutes')} local {what}")
        if len(hazards) > 10:
            print(f"  ... {len(hazards) - 10} more")
        print("Fix: schedule in UTC, or move the job outside the local 01:00-03:00 window.")
        return 1 if args.fail_on_dst else 0
    print("\nNo DST hazards in the next 12 months.")
    return 0


if __name__ == "__main__":
    sys.exit(main())