Will Your Cowork Scheduled Task Still Run Next Saturday?

name	stream	kind	schedule	enabled	note
wallpaper-category	wallpaper-category	cron	30 4,16 * * *	true	
monthly-check	monthly-check	one-time	2026-10-01T09:00	true	
weekend-a	weekend	cron	0 1 * * 6	true	
weekend-b	weekend	cron	30 1 * * 6	true	
backup	backup	cron	15 9 * * 2,5	true	
backup-fri	backup	cron	15 9 * * 5	false	
effect-eval	effect-eval	one-time	2026-10-12T10:00	true	

stream	days	recurring
wallpaper-category	daily	true
monthly-check	daily	true
weekend	sat,sun	true
backup	tue,fri	true
effect-eval	-	false

#!/usr/bin/env python3
"""Read a scheduled-task ledger (TSV) and catch accidents that are visible in the definitions alone.
 
Detects:
  ONE-TIME  : a one-off definition inside a stream that is meant to recur
  MULTI     : a single cron that fires more than once a day (split it)
  GAP       : weekdays a stream should cover that no enabled task actually fires on
  DISABLED  : a disabled task with no reason written in the ledger
"""
import csv
import sys
from datetime import datetime, timedelta
 
DOW_NAMES = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
 
 
def expand_field(expr, lo, hi, is_dow=False):
    """Expand one cron field into a set of ints (*, a-b, a,b, */n, a-b/n)."""
    values = set()
    for part in expr.split(","):
        step = 1
        if "/" in part:
            part, step_s = part.split("/", 1)
            step = int(step_s)
        if part == "*":
            start, end = lo, hi
        elif "-" in part:
            a, b = part.split("-", 1)
            start, end = int(a), int(b)
        else:
            start = end = int(part)
        values.update(range(start, end + 1, step))
    if is_dow:
        # In cron, both 0 and 7 mean Sunday. Normalize to Python weekday() (Mon=0)
        values = {6 if v in (0, 7) else v - 1 for v in values}
    return values
 
 
def parse_cron(expr):
    fields = expr.split()
    if len(fields) != 5:
        raise ValueError(f"not a 5-field cron: {expr!r}")
    minute, hour, dom, month, dow = fields
    return {
        "minute": expand_field(minute, 0, 59),
        "hour": expand_field(hour, 0, 23),
        "dom": expand_field(dom, 1, 31),
        "month": expand_field(month, 1, 12),
        "dow": expand_field(dow, 0, 7, is_dow=True),
        "dom_any": dom == "*",
        "dow_any": dow == "*",
    }
 
 
def day_matches(c, day):
    if day.month not in c["month"]:
        return False
    dom_ok = day.day in c["dom"]
    dow_ok = day.weekday() in c["dow"]
    # Standard cron: if both day-of-month and day-of-week are set, EITHER one matches
    if c["dom_any"] or c["dow_any"]:
        return dom_ok and dow_ok
    return dom_ok or dow_ok
 
 
def fires_on(c, day):
    """Fire times on that day, in minutes after midnight."""
    if not day_matches(c, day):
        return []
    return sorted(h * 60 + m for h in c["hour"] for m in c["minute"])
 
 
def load_ledger(path):
    with open(path, newline="", encoding="utf-8") as f:
        rows = list(csv.DictReader(f, delimiter="\t"))
    for r in rows:
        r["enabled"] = r["enabled"].strip().lower() in ("1", "true", "yes", "on")
    return rows
 
 
def coverage(rows, start, days):
    """For each stream, the set of weekdays covered by enabled recurring tasks."""
    cov = {}
    for r in rows:
        cov.setdefault(r["stream"], set())
        if r["kind"] != "cron" or not r["enabled"]:
            continue
        c = parse_cron(r["schedule"])
        for i in range(days):
            day = start + timedelta(days=i)
            if fires_on(c, day):
                cov[r["stream"]].add(day.weekday())
    return cov
 
 
def expected_days(spec):
    spec = spec.strip().lower()
    if spec == "daily":
        return set(range(7))
    return {DOW_NAMES.index(s.strip()) for s in spec.split(",") if s.strip()}
 
 
def audit(rows, streams, start, days=14):
    findings = []
    for r in rows:
        name, stream = r["name"], r["stream"]
        recurring = streams.get(stream, {}).get("recurring", True)
        if r["kind"] == "one-time" and recurring:
            findings.append(("ONE-TIME", name,
                             f"stream {stream} is meant to recur, but this is a one-off definition ({r['schedule']})"))
        if r["kind"] == "cron":
            c = parse_cron(r["schedule"])
            per_day = len(c["hour"]) * len(c["minute"])
            if per_day > 1:
                findings.append(("MULTI", name,
                                 f"one task holds {per_day} fires per day ({r['schedule']}); give each slot its own task"))
        if not r["enabled"] and not r.get("note", "").strip():
            findings.append(("DISABLED", name, "disabled, but the ledger has no reason for stopping it"))
    cov = coverage(rows, start, days)
    for stream, spec in streams.items():
        if not spec.get("recurring", True):
            continue
        missing = expected_days(spec["days"]) - cov.get(stream, set())
        if missing:
            names = ",".join(DOW_NAMES[d] for d in sorted(missing))
            findings.append(("GAP", stream, f"no enabled task covers {names}, which this stream is expected to run on"))
    return findings
 
 
def load_streams(path):
    streams = {}
    with open(path, newline="", encoding="utf-8") as f:
        for r in csv.DictReader(f, delimiter="\t"):
            streams[r["stream"]] = {
                "days": r["days"],
                "recurring": r["recurring"].strip().lower() in ("1", "true", "yes"),
            }
    return streams
 
 
if __name__ == "__main__":
    if len(sys.argv) < 3:
        sys.exit("usage: schedule_audit.py tasks.tsv streams.tsv")
    rows = load_ledger(sys.argv[1])
    streams = load_streams(sys.argv[2])
    today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    findings = audit(rows, streams, today)
    for kind, target, msg in findings:
        print(f"[{kind:8}] {target}: {msg}")
    print(f"-- {len(findings)} finding(s)")
    sys.exit(1 if findings else 0)

[MULTI   ] wallpaper-category: one task holds 2 fires per day (30 4,16 * * *); give each slot its own task
[ONE-TIME] monthly-check: stream monthly-check is meant to recur, but this is a one-off definition (2026-10-01T09:00)
[DISABLED] backup-fri: disabled, but the ledger has no reason for stopping it
[GAP     ] monthly-check: no enabled task covers mon,tue,wed,thu,fri,sat,sun, which this stream is expected to run on
[GAP     ] weekend: no enabled task covers sun, which this stream is expected to run on
-- 5 finding(s)

#!/usr/bin/env python3
"""Compare ledgers before and after a cron edit and show only the weekdays the edit newly leaves empty."""
import sys
from datetime import datetime
from schedule_audit import DOW_NAMES, coverage, load_ledger
 
 
def new_holes(before_rows, after_rows, start, days=14):
    before = coverage(before_rows, start, days)
    after = coverage(after_rows, start, days)
    holes = {}
    for stream, days_before in before.items():
        lost = days_before - after.get(stream, set())
        if lost:
            holes[stream] = sorted(lost)
    return holes
 
 
if __name__ == "__main__":
    before_rows = load_ledger(sys.argv[1])
    after_rows = load_ledger(sys.argv[2])
    start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    holes = new_holes(before_rows, after_rows, start)
    if not holes:
        print("OK: this change leaves no weekday empty")
        sys.exit(0)
    for stream, lost in holes.items():
        print(f"HOLE: {stream} loses {','.join(DOW_NAMES[d] for d in lost)}")
    sys.exit(1)

$ python3 change_guard.py before.tsv after.tsv
HOLE: weekend loses sun
$ echo $?
1