Cowork の定期タスクは、来週の土曜にも動きますか

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
"""スケジュールタスク台帳（TSV）を読み、定義の段階で分かる事故を先に見つける。
 
検出するもの:
  ONE-TIME  : 繰り返すつもりの系統に、一回きりの定義が混じっている
  MULTI     : 1本の cron が1日に複数回の着火を持っている（分けるべき形）
  GAP       : 系統ごとに「動いてほしい曜日」を、実際の着火が埋めていない
  DISABLED  : 無効になっているのに、止めた理由が台帳に書かれていない
"""
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):
    """cron の1フィールドを整数の集合に展開する（*, 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:
        # cron の曜日は 0 と 7 がどちらも日曜。Python の weekday() に揃える（月=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"5フィールドではありません: {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"]
    # 標準 cron: 日と曜日の両方が指定されていれば「どちらか」で一致する
    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):
    """その日の着火時刻（分単位）の一覧。"""
    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):
    """系統ごとに、有効な定期タスクで埋まっている曜日の集合を返す。"""
    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} は繰り返す想定ですが、一回きりの定義です（{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"1日に {per_day} 回の着火を1本で持っています（{r['schedule']}）。枠ごとに別タスクへ"))
        if not r["enabled"] and not r.get("note", "").strip():
            findings.append(("DISABLED", name, "無効ですが、止めた理由が台帳にありません"))
    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"動いてほしい曜日のうち {names} を、どの有効タスクも埋めていません"))
    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)} 件")
    sys.exit(1 if findings else 0)

[MULTI   ] wallpaper-category: 1日に 2 回の着火を1本で持っています（30 4,16 * * *）。枠ごとに別タスクへ
[ONE-TIME] monthly-check: 系統 monthly-check は繰り返す想定ですが、一回きりの定義です（2026-10-01T09:00）
[DISABLED] backup-fri: 無効ですが、止めた理由が台帳にありません
[GAP     ] monthly-check: 動いてほしい曜日のうち mon,tue,wed,thu,fri,sat,sun を、どの有効タスクも埋めていません
[GAP     ] weekend: 動いてほしい曜日のうち sun を、どの有効タスクも埋めていません
-- 5 件

#!/usr/bin/env python3
"""cron を書き換える前後の台帳を比べ、変更で新しく空いた曜日だけを示す。"""
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: この変更で空く曜日はありません")
        sys.exit(0)
    for stream, lost in holes.items():
        print(f"HOLE: {stream} から {','.join(DOW_NAMES[d] for d in lost)} が消えます")
    sys.exit(1)

$ python3 change_guard.py before.tsv after.tsv
HOLE: weekend から sun が消えます
$ echo $?
1