エージェントの定期実行で、失敗より先に「走らなかった」を疑うようになりました

# 実測: 同じ判定でも、パイプを一本挟むと結果が逆になります
$ grep -q NOPE /etc/hostname | head -1; echo $?
0                                  # 失敗したのに 0
 
$ set -o pipefail; grep -q NOPE /etc/hostname | head -1; echo $?
1                                  # pipefail を立てれば拾えます
 
$ grep -q NOPE /etc/hostname; echo $?
1                                  # そもそもパイプを挟まないのが確実です

# 悪い例: <<EOF（クォートなし）
cat << EOF > ledger_bad.txt
id=$RUN_ID status=OK started=`date -u +%s` note=`echo INJECTED`
EOF
 
# 出力（記録したかった文字列ではなく、実行結果が入ります）
# id=run-2026-09-06 status=OK started=1788657291 note=INJECTED

# 良い例: <<'EOF'（クォートあり）＋ 後から置換
cat << 'EOF' > ledger_good.txt
id=__RUN_ID__ status=OK started=__TS__ note=none
EOF
sed -i "s/__RUN_ID__/$RUN_ID/; s/__TS__/$(date -u +%s)/" ledger_good.txt
 
# 出力
# id=run-2026-09-06 status=OK started=1757116800 note=none

# audit.py — 期待発火表と実行台帳を突合し、欠落した実行を出します
from datetime import datetime, timedelta
import json, sys
from expect import expected
 
GRACE = timedelta(minutes=20)      # 起動遅延の許容幅（決め方は次節）
 
def load_ledger(path):
    """1行1レコードの JSONL を (開始時刻, レコード) の昇順で返します。"""
    rows = []
    for line in open(path, encoding="utf-8"):
        line = line.strip()
        if not line:
            continue
        r = json.loads(line)
        rows.append((datetime.fromisoformat(r["started_at"]), r))
    return sorted(rows)
 
def audit(schedules, ledger_path, days):
    rows = load_ledger(ledger_path)
    missing, matched = [], 0
    for job, expr in schedules.items():
        seen = [t for t, r in rows if r["job"] == job]
        for day in days:
            for exp in expected(expr, day):
                if exp > datetime.now():       # まだ来ていない枠は欠落にしません
                    continue
                if any(abs(t - exp) <= GRACE for t in seen):
                    matched += 1
                else:
                    missing.append((job, exp))
    return matched, missing
 
if __name__ == "__main__":
    schedules = {
        "wallpaper-category": "30 4,16 * * *",
        "reference-refresh":  "0 7 * * *",
    }
    days = [datetime(2026, 9, 4), datetime(2026, 9, 5), datetime(2026, 9, 6)]
    matched, missing = audit(schedules, "ledger.jsonl", days)
    print(f"照合できた実行: {matched}")
    print(f"欠落した実行  : {len(missing)}")
    for job, t in missing:
        print(f"  MISSING {job} @ {t:%Y-%m-%d %H:%M}")
    sys.exit(1 if missing else 0)   # 欠落があれば非ゼロで終えます

# 期待時刻とのずれを一覧します（許容幅を決める前に必ず測ります）
python3 - << 'EOF'
import json
from datetime import datetime
for line in open("ledger.jsonl", encoding="utf-8"):
    r = json.loads(line)
    t = datetime.fromisoformat(r["started_at"])
    delay = t.minute * 60 + t.second - 30 * 60     # 04:30 起点の秒数
    print(f'{r["job"]:20} {t:%m-%d %H:%M:%S}  遅れ {delay:>4}s')
EOF

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
 
utc = datetime(2026, 9, 6, 4, 31, 12, tzinfo=ZoneInfo("UTC"))       # 台帳の1行
jst = datetime(2026, 9, 6, 4, 30,     tzinfo=ZoneInfo("Asia/Tokyo")) # 期待表の1枠
 
print(abs(utc - jst))                       # 9:01:12
print(abs(utc - jst) <= timedelta(minutes=20))   # False — 全件が欠落に見えます
 
# タイムゾーンを落として比べた場合
print(abs(utc.replace(tzinfo=None) - jst.replace(tzinfo=None)))   # 0:01:12

# 期待表の側もタイムゾーンを明示して組みます
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Tokyo")
days = [datetime(2026, 9, 4, tzinfo=TZ), datetime(2026, 9, 5, tzinfo=TZ)]