スケジュール実行エージェントの指示書ドリフト — 定義・文書・実態の乖離を検知する三層整合の設計

#!/usr/bin/env bash
# require_ref.sh — 参照ファイルの欠落・空読みをドリフトログに残す
DRIFT_LOG="${DRIFT_LOG:-$HOME/.agent-drift.log}"
 
require_ref() {
  local path="$1"
  if [ ! -s "$path" ]; then
    printf '%s missing-or-empty %s\n' "$(date -Is)" "$path" >> "$DRIFT_LOG"
    echo "⚠️ 参照欠落: $path" >&2
    return 1
  fi
  cat "$path"
}
 
# 手順書内では cat の代わりに必ずこれを通す
require_ref "data/crash-thresholds.yaml" || exit 1

#!/usr/bin/env python3
"""drift_check.py — 定義(tasks.yaml)と文書・参照の乖離を検査する"""
import pathlib
import sys
import yaml
 
ROOT = pathlib.Path(__file__).resolve().parent
 
# 文書側で使ってよい頻度語彙と、cron 式との整合条件
FREQ_RULES = {
    "毎日": lambda c: c.split()[4] == "*",
    "毎週": lambda c: c.split()[4] != "*" and "," not in c.split()[4],
    "週 2 回": lambda c: c.split()[4].count(",") == 1,
}
 
def load_tasks():
    with open(ROOT / "tasks.yaml", encoding="utf-8") as f:
        return yaml.safe_load(f)["tasks"]
 
def check(tasks):
    problems = []
    all_docs = {p.resolve() for p in (ROOT / "docs" / "agents").glob("*.md")}
    used_docs = set()
 
    for t in tasks:
        doc = (ROOT / t["doc"]).resolve()
        used_docs.add(doc)
        if not doc.is_file():
            problems.append(f"[zombie] {t['name']}: 文書が存在しません -> {t['doc']}")
            continue
        text = doc.read_text(encoding="utf-8")
        for word, ok in FREQ_RULES.items():
            if word in text and not ok(t["schedule"]):
                problems.append(
                    f"[freq] {t['name']}: 文書は「{word}」/ 定義は {t['schedule']}")
        for ref in t.get("refs", []):
            rp = ROOT / ref
            if not rp.is_file() or rp.stat().st_size == 0:
                problems.append(f"[ref] {t['name']}: 参照が欠落または空 -> {ref}")
 
    for orphan in sorted(all_docs - used_docs):
        problems.append(f"[orphan] 定義から参照されない文書: {orphan.name}")
    return problems
 
if __name__ == "__main__":
    found = check(load_tasks())
    for line in found:
        print(line)
    sys.exit(1 if found else 0)

# 定義上アクティブなタスクの成果物の鮮度を一覧する
python3 - << 'EOF'
import pathlib, time, yaml
 
tasks = yaml.safe_load(open("tasks.yaml", encoding="utf-8"))["tasks"]
now = time.time()
for t in tasks:
    out = pathlib.Path("outputs") / t["name"]
    files = sorted(out.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not files:
        print(f"[stale] {t['name']}: 成果物がありません")
        continue
    age_h = (now - files[0].stat().st_mtime) / 3600
    flag = "[stale]" if age_h > 48 else "[ok]   "
    print(f"{flag} {t['name']}: 最終成果物 {age_h:.0f} 時間前")
EOF

# 毎朝の検査をログに積む（これ自体もスケジュール実行）
{ date -I; python3 drift_check.py; echo "---"; } >> drift-history.log