クォータが復旧した瞬間、15本のジョブをどれだけ再実行すべきか ― launchdスロット選別バッチの設計

前回、Wikiのsecret-scan同期を直した話を書きました。今回は毛色を変えて、claude-quota-guard.py として組んだクォータのサーキットブレーカーの続きです。あのスクリプトはクォータ枯渇を検知して全ジョブを止めるところで話が終わっていましたが、実際に運用するとその先に別の問題があります。サーキットが閉じた後、スキップされた分をどうするかです。

困りごと：サーキットが閉じても、スキップされたジョブは戻ってこない

claude-quota-guard.py のrun_jobは、サーキットが開いている間にジョブが起動すると即座にexit 0で返し、ログにマーカーだけ残します。

print(
    "CLAUDE_QUOTA_JOB_SKIPPED "
    f"job={label} reason={status['reason']} remaining={status['remaining_seconds']}s ts={now()}",
    file=sys.stderr,
)
return 0

このガードは~/Library/LaunchAgents/*.plistのうち15本にかかっています。コード中のコメントにある通りです。

# 🔴 2026-08-21: circuit が開くと全ジョブが一律で止まるため、消費の大半を占める
# 返信/エンゲージ系がクォータを使い切った巻き添えで「投稿」まで停止していた。

問題は、サーキットが閉じてopen_untilが過ぎても、launchdは次にそのジョブのStartCalendarIntervalが来るまで何もしてくれないことです。朝9:00のジョブがクォータでスキップされ、10:00に復旧しても、次のスロットが翌朝9:00なら丸1日分の実行機会が失われます。この「復旧後に何を、どれだけ再実行するか」を決めるのがquota-catchup.pyです。

設計：3段の絞り込みで候補を決める

quota-catchup.pyはfind_candidatesで全plistを舐め、3つの条件をANDで通したものだけを再実行候補にします。

def find_candidates(...) -> list[Candidate]:
    candidates: list[Candidate] = []
    for plist_path in sorted(launch_agents.glob("*.plist")):
        plist = load_plist(plist_path)
        if not plist or not is_guarded(plist):
            continue
        ...
        if label in already_kicked or not latest_marker_is_today_skip(output_paths, label, today):
            continue
        if calendar_slot_passed(plist, now):
            candidates.append(Candidate(label, plist_path))
    return candidates

二重起動を防ぐ：is_guarded

再実行対象は「claude-quota-guard.py経由で起動している」ジョブに限定します。判定はProgramArgumentsの文字列一致だけです。

def is_guarded(plist: dict) -> bool:
    arguments = plist.get("ProgramArguments")
    return isinstance(arguments, list) and any("claude-quota-guard" in str(value) for value in arguments)

ガードを経由していないplistまで拾うと、クォータと無関係な普通のcronジョブを勝手にkickしてしまいます。

「過ぎたスロットがあるか」を決める：calendar_slot_passed

ここが今回の主役です。StartCalendarIntervalは1つのジョブに複数スロットを持てます。実例がcom.shun.daily-brief.plistです。

<key>StartCalendarInterval</key>
<array>
    <dict><key>Hour</key><integer>8</integer><key>Minute</key><integer>0</integer></dict>
    <dict><key>Hour</key><integer>10</integer><key>Minute</key><integer>30</integer></dict>
</array>

8:00の枠がクォータでスキップされても、10:30にまた同じジョブが自然に走ります。だから「過ぎたスロットが1つでもあれば再実行対象」であり、「未来のスロットが残っているから待て」ではありません。逆にStartInterval（例：30分おき）のジョブは放っておいても次のインターバルで自然に再実行されるので、catchupの対象にする必要がありません。

def calendar_slot_passed(plist: dict, now: datetime) -> bool:
    """True only for calendar-only jobs with at least one past slot today."""
    if "StartInterval" in plist:
        return False
    raw_entries = plist.get("StartCalendarInterval")
    if isinstance(raw_entries, dict):
        entries = [raw_entries]
    elif isinstance(raw_entries, list):
        entries = raw_entries
    else:
        return False

    saw_today_slot = False
    for entry in entries:
        if not isinstance(entry, dict) or "Hour" not in entry:
            return False
        if not runs_today(entry, now):
            continue
        try:
            scheduled = now.replace(
                hour=int(entry["Hour"]),
                minute=int(entry.get("Minute", 0)),
                second=0, microsecond=0,
            )
        except (TypeError, ValueError):
            return False
        if scheduled > now:
            continue
        saw_today_slot = True
    return saw_today_slot

全スロットを最後まで舐めてsaw_today_slotをORで積むのがポイントです。もし「最初に見つかったスロットで判定を返す」ような書き方をしていたら、スロットの並び順次第で結果が変わるバグを踏みます。例えば[{9:00}, {18:00}]という並びで13:00に判定する場合、正しくは9:00が過去なので対象ですが、ループを最後のエントリだけで判定していたら18:00が未来という理由で対象外に誤判定します。

実測から調整したタイムアウト：JOB_TIMEOUT_SECONDS

再実行はkick_and_waitでlaunchctl kickstartしてから、launchdの管理下から外れる（＝終了する）までポーリングします。このタイムアウトは最初1200秒でしたが、実測で不十分でした。

# 実測(2026-08-21): affameba-gen 等の claude 生成レーンは 20 分を超える。
# 1200s だと「待つのをやめて次を kick」するだけで前のジョブは生き続け、
# runbook が要求する直列 kick が崩れて重い生成が重なる（load 40 超の二次被害）。
JOB_TIMEOUT_SECONDS = 2700
# timeout 時は待つのをやめるだけでなく実際に止める。ここを殺さないと直列性が保てない。
JOB_KILL_GRACE_SECONDS = 30

「待つのをやめる」と「ジョブを止める」は別物だ、というのがここの学びです。前者だけだと、タイムアウトした古いジョブがバックグラウンドで生き続けたまま次の候補をkickしてしまい、生成が重なってload 40超という二次被害が出ました。なのでterminate_jobでSIGTERM→SIGKILLの順に実際に殺し、直列性を担保しています。

def terminate_job(domain_label: str, label: str) -> None:
    """timeout したジョブを実際に止める。次の kick と重ならせないための直列性の担保。"""
    for signal_name in ("SIGTERM", "SIGKILL"):
        subprocess.run(["launchctl", "kill", signal_name, domain_label], capture_output=True, check=False)
        deadline = time.monotonic() + JOB_KILL_GRACE_SECONDS
        while time.monotonic() < deadline:
            pid, _ = launchctl_status(label)
            if pid is None:
                return
            time.sleep(2)

テストで境界条件を潰す：test_quota_catchup.py

このスクリプトには18個のテストケースがあります（unittestで17、pytestスタイルの関数が1）。個人の自動化スクリプトにここまでテストを書くのはやり過ぎに見えるかもしれませんが、実際に効いたケースを挙げます。

候補ゼロの日にclaudeを叩かない

いちばん実害が大きいのはこれです。run関数のコメントにその理由が書いてあります。

# 拾うものが無い日に probe を撃つと、30分おきに claude -p を1日48回空撃ちして
# クォータを削る（このジョブが防ごうとしている事故そのものを起こす）。
# 候補が出た時だけ回復を確認する。
if not candidates:
    return [], 0, 0

これを守るテストがtest_no_candidates_skips_probeです。

def test_no_candidates_skips_probe(self):
    self.add_job(
        error_text=f"CLAUDE_QUOTA_JOB_RAN job=com.lily.test exit=0 ts={int(self.now.timestamp())}",
    )
    probe = Mock(return_value=True)
    kicker = Mock()
    result = quota_catchup.run(
        dry_run=False, now=self.now, state_path=self.state, catchup_path=self.catchup,
        launch_agents=self.agents, log_path=self.root / "result.log", probe=probe, kicker=kicker,
    )
    self.assertEqual(result, ([], 0, 0))
    probe.assert_not_called()
    kicker.assert_not_called()

「クォータ復旧を確認するためのバッチが、確認作業自体でクォータを削る」というのは、サーキットブレーカーを作った側なら絶対に踏みたくない自己矛盾です。probe.assert_not_called()という1行が、それを機械的に保証しています。

今日のスキップは対象、3日前は対象外

def test_today_skip_is_candidate(self):
    label = self.add_job()
    self.assertEqual([item.label for item in self.candidates()], [label])

def test_old_skip_is_not_candidate(self):
    self.add_job(mtime=self.now - timedelta(days=3))
    self.assertEqual(self.candidates(), [])

latest_marker_is_today_skipは正規表現でログからts=を拾い、日付が今日かどうかを見ています。3日前のSKIPPEDログが今日また拾われたら、過去の失敗を毎日再実行し続けるゾンビ状態になります。

過去スロットがあれば未来スロットが残っていても対象

def test_past_slot_makes_candidate_even_if_later_slot_is_future(self):
    self.add_job(schedule=[{"Hour": 9, "Minute": 0}, {"Hour": 18, "Minute": 0}])
    self.assertEqual([item.label for item in self.candidates()], ["com.lily.test"])

先述したcalendar_slot_passedのOR判定を、実際のdaily-brief型スケジュール（複数スロット）で固定しているテストです。

マーカーの順序を取り違えない

ログにはSKIPPEDとRANが混在します。「最新のマーカーがどちらか」を見誤ると、実は成功済みのジョブを二重にkickしたり、逆に成功後の別スキップを見落としたりします。

def test_ran_marker_after_skip_in_same_log_is_not_candidate(self):
    label = "com.lily.mixed"
    today = int(self.now.timestamp())
    self.add_job(
        label,
        error_text=(
            f"CLAUDE_QUOTA_JOB_SKIPPED job={label} reason=quota remaining=1s ts={today - 1}\n"
            f"CLAUDE_QUOTA_JOB_RAN job={label} exit=0 ts={today}"
        ),
    )
    self.assertEqual(self.candidates(), [])

SKIPPEDの後にRANがあれば「結局あとで成功した」ので対象外、という時系列判定をlatest_job_markerがreversed(lines)で担保しています。

このスクリプトの単体テストに共通する狙いは「賢いロジックの正しさ」ではなく「素朴に書いたら間違える境界を先に固定する」ことです。calendar_slot_passedのOR判定、マーカーの新旧判定、候補ゼロ時のprobe抑制――どれも実装1行の書き方次第で結果が反転する箇所で、動かしてみるまで気づきにくいものばかりです。個人用の自動化スクリプトにpytestを書く価値は、レビュアーを説得するためではなく、半年後の自分が仕様を変えたときに同じ間違いを踏まないためにあります。

踏んだ落とし穴

• 1200秒タイムアウトで20分超のジョブが打ち切られる → 実測から2700秒に拡張、JOB_KILL_GRACE_SECONDS=30でSIGTERM→SIGKILLの猶予を確保
• タイムアウトで「待つのをやめる」だけだと直列性が崩れる → terminate_jobで実際にkillしないと、前のジョブが生き続けたまま次をkickしてload 40超の二次被害が出る
• 候補ゼロの日にprobeを撃つと、防ごうとしている事故そのものが起きる → 候補がある時だけprobe_claude()を呼ぶ
• StartIntervalジョブまで拾うと不要な再実行になる → StartIntervalがあれば無条件でcalendar_slot_passedはFalse
• SKIPPED/RANのマーカー順序を「存在するかどうか」だけで見ると誤判定する → reversed(lines)で最新のマーカーだけを見る
• --dry-runのたびにclaudeを叩くと確認コストが実行コストになる → dry-runはprobeを呼ばず候補列挙だけで返す

まとめ

• クォータのサーキットブレーカーは「止める」で終わりではなく、閉じた後に何を再実行するかまで設計しないと復旧が翌日待ちになる
• 再実行候補はis_guarded / latest_marker_is_today_skip / calendar_slot_passedの3段AND。過ぎたスロットが1つでもあれば対象、未来スロットの有無は無関係
• タイムアウトは実測（20分超のジョブ実在）から決め、打ち切り時は実際にkillして直列性を守る
• 候補ゼロならprobeすら呼ばない。復旧確認のためのバッチが自分でクォータを削る自己矛盾を避ける
• 個人スクリプトの単体テストは、素朴な実装だと反転しがちな境界（スロット順序・マーカー新旧・ゼロ件時の副作用）を先に固定しておくためにある

次回は、このサーキット本体――claude-quota-guard.pyが「上限メッセージ」と「記事本文にたまたま同じ語がある成功ケース」をどう区別しているか、を書く予定です。

---

Lily（@bokuwalily）― 個人開発者。Claude Code で自動化基盤を組みながら、iOSアプリやWebサービスを量産しています

• AIで「寝てても回る仕組み」を作って月120万にした話は noteの有料記事 に💰
• OSS: github.com/bokuwalily 🐙
• 最新情報・お問い合わせは X @bokuwalily へ🌍
• AI導入・自動化の相談と実装テンプレ7本の配布は 公式LINE から💬

皆さんの ❤️ やシェアが励みになります！