スケジュール生成が無音で不発に終わったことを、外側から気づく仕組み

# heartbeat.sh — 生成ジョブの末尾で呼ぶ。判断はせず、起きた事実だけを打刻する
# 使い方: heartbeat.sh <job_id> <repo_dir> <expected_min_articles>
set -uo pipefail
 
JOB_ID="$1"
REPO_DIR="$2"
EXPECTED_MIN="${3:-1}"
LEDGER="${HEARTBEAT_LEDGER:-$HOME/ops/heartbeat.jsonl}"
mkdir -p "$(dirname "$LEDGER")"
 
# 地の事実: 直近1コミットで追加された ja 記事の数を数える（自己カウントしない）
cd "$REPO_DIR" || exit 0
ADDED=$(git show --name-only --pretty=format: HEAD 2>/dev/null \
  | grep -c '^content/articles/ja/.*\.mdx$')
HEAD_SHA=$(git rev-parse --short HEAD 2>/dev/null || echo "nohead")
TS=$(TZ=Asia/Tokyo date +%Y-%m-%dT%H:%M:%S%z)
 
printf '{"job":"%s","ts":"%s","sha":"%s","added":%s,"expected_min":%s}\n' \
  "$JOB_ID" "$TS" "$HEAD_SHA" "${ADDED:-0}" "$EXPECTED_MIN" >> "$LEDGER"

#!/usr/bin/env python3
# watchdog.py — 期待されたジョブの鼓動と、リポジトリの地の事実を突き合わせる
import json, subprocess, sys
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo
 
JST = ZoneInfo("Asia/Tokyo")
LEDGER = Path.home() / "ops" / "heartbeat.jsonl"
 
# 曜日ごとの期待集合（0=月 .. 6=日）。休止日はここで宣言して誤検知を防ぐ
EXPECTED_BY_WEEKDAY = {
    0: {"claudelab-premium", "gemilab-premium", "antigravitylab-premium", "rorklab-premium"},
    # 平日は4サイト共通。土日は別集合に差し替える運用も同じ形で書ける
}
for wd in range(1, 5):
    EXPECTED_BY_WEEKDAY[wd] = EXPECTED_BY_WEEKDAY[0]
EXPECTED_BY_WEEKDAY[5] = {"weekend-content"}
EXPECTED_BY_WEEKDAY[6] = {"weekend-content"}
 
REPO_OF = {
    "claudelab-premium": Path.home() / "repos" / "claudelab.net",
    "gemilab-premium": Path.home() / "repos" / "gemilab.net",
    "antigravitylab-premium": Path.home() / "repos" / "antigravitylab.net",
    "rorklab-premium": Path.home() / "repos" / "rorklab.net",
}
 
def today_jst():
    return datetime.now(JST).date()
 
def load_today_beats():
    beats = {}
    if not LEDGER.exists():
        return beats
    for line in LEDGER.read_text(encoding="utf-8").splitlines():
        try:
            rec = json.loads(line)
            ts = datetime.fromisoformat(rec["ts"])
        except (json.JSONDecodeError, KeyError, ValueError):
            continue
        if ts.astimezone(JST).date() == today_jst():
            beats[rec["job"]] = rec  # 同一ジョブの最後の打刻を採用
    return beats
 
def ground_truth_added(repo: Path) -> int:
    """ウォッチドッグ自身が数える。台帳の self-count は信用しない。"""
    if not (repo / ".git").exists():
        return -1  # クローンが無い＝検証不能。これも異常として扱う
    since = today_jst().isoformat()
    out = subprocess.run(
        ["git", "-C", str(repo), "log", "--since", since,
         "--name-only", "--pretty=format:"],
        capture_output=True, text=True
    ).stdout
    return sum(
        1 for ln in out.splitlines()
        if ln.startswith("content/articles/ja/") and ln.endswith(".mdx")
    )
 
def main():
    weekday = today_jst().weekday()
    expected = EXPECTED_BY_WEEKDAY.get(weekday, set())
    beats = load_today_beats()
 
    missing, hollow = [], []
    for job in sorted(expected):
        rec = beats.get(job)
        if rec is None:
            missing.append(job)          # 鼓動が無い＝完全な沈黙
            continue
        repo = REPO_OF.get(job)
        truth = ground_truth_added(repo) if repo else int(rec.get("added", 0))
        if truth <= 0:
            hollow.append((job, rec.get("added", 0), truth))  # 打刻はあるが地に無い
 
    if not missing and not hollow:
        print(f"[OK] {today_jst()} 期待 {len(expected)} 件すべて成果を確認")
        return 0
 
    if missing:
        print(f"[ALERT] 無音の不発（鼓動なし）: {', '.join(missing)}")
    for job, claimed, truth in hollow:
        print(f"[ALERT] 空打ち（自己申告={claimed} / 実測={truth}）: {job}")
    return 1
 
if __name__ == "__main__":
    sys.exit(main())