15 launchd Jobs and One Quota Circuit Breaker: Deciding What to Re-run Once the Circuit Closes

A circuit breaker that halts every job the moment your quota runs dry is only half the story. The harder question shows up afterward: once the circuit closes again, what do you do with everything it skipped? In my previous post I wrote about fixing the Wiki secret-scan sync. This time I'm switching gears and following up on claude-quota-guard.py, the quota circuit breaker I built. That script's story ended at "detect quota exhaustion, stop all jobs." In real operation, there's a whole second problem waiting past that point—and along the way it involved a 1200-second timeout that turned out to need 2700, and a load average that climbed past 40 when I got it wrong.

run_job in claude-quota-guard.py returns exit 0 immediately when a job starts while the circuit is open, leaving nothing but a marker in the log.

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

This guard sits in front of 15 of the plists under ~/Library/LaunchAgents/*.plist. As the comment in the code puts it:

The problem is that after the circuit closes and open_until has passed, launchd does nothing until that job's next StartCalendarInterval slot comes around. If the 9:00 AM job was skipped for quota reasons and the quota recovers at 10:00, but the next slot is 9:00 AM tomorrow, you've lost an entire day's execution. Deciding "what to re-run, and how much, after recovery" is the job of quota-catchup.py.

quota-catchup.py walks every plist in find_candidates and only treats a job as a re-run candidate if it passes three conditions ANDed together.

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

Re-run targets are limited to jobs that are "launched via claude-quota-guard.py." The check is nothing more than a string match on 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)

If you picked up plists that don't go through the guard, you'd end up kicking ordinary cron-style jobs that have nothing to do with the quota.

This is the star of the show. A single job can have multiple StartCalendarInterval slots. A real example is 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>

Even if the 8:00 slot gets skipped for quota, the same job naturally runs again at 10:30. So the rule is "if even one slot has passed, it's a re-run candidate," not "wait, because there's still a future slot." Conversely, StartInterval jobs (e.g., every 30 minutes) will naturally re-run on the next interval if you just leave them alone, so there's no need to make them catch-up targets.

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

The key is walking every slot to the end and accumulating saw_today_slot with an OR. If you'd written it as "return the verdict from the first slot you find," you'd hit a bug where the result depends on the order of the slots. For example, with the ordering [{9:00}, {18:00}] evaluated at 13:00, the correct answer is "candidate" because 9:00 is in the past—but if the loop only judged by the last entry, it would misclassify the job as not a candidate on the grounds that 18:00 is in the future.

A re-run goes through kick_and_wait, which runs launchctl kickstart and then polls until the job leaves launchd's management (i.e., exits). This timeout started at 1200 seconds, which turned out to be insufficient in practice.

JOB_TIMEOUT_SECONDS = 2700
JOB_KILL_GRACE_SECONDS = 30

The lesson here: "stop waiting" and "stop the job" are two different things. With only the former, the timed-out old job kept running in the background while the next candidate got kicked, the generation workloads piled up, and the load average went past 40 as collateral damage. So terminate_job actually kills the job—SIGTERM, then SIGKILL—to guarantee serial execution.

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)

This script has 18 test cases (17 in unittest, plus 1 pytest-style function). Writing this many tests for a personal automation script might look like overkill, so here are the cases where they actually earned their keep.

Don't call claude on days with zero candidates

This is the one with the biggest real-world cost. The comment in the run function explains why.

if not candidates:
    return [], 0, 0

The test that protects this is 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()

"The batch that checks whether the quota has recovered burns quota just by checking" is a self-contradiction that anyone who built a circuit breaker absolutely does not want to step into. That single line, probe.assert_not_called(), guarantees it mechanically.

Today's skip is a candidate; a skip from three days ago is not

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 uses a regex to pull ts= out of the log and checks whether the date is today. If a SKIPPED log from three days ago got picked up again today, you'd have a zombie state where past failures get re-run every single day.

A past slot makes it a candidate even if a future slot remains

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"])

This test pins the OR logic in calendar_slot_passed described above, using a real daily-brief-style schedule (multiple slots).

Don't get the marker order wrong

The log contains both SKIPPED and RAN. If you misjudge which one is the latest marker, you'll either double-kick a job that actually succeeded, or miss a separate skip that happened after a success.

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(), [])

If RAN comes after SKIPPED, the job "succeeded later after all" and is not a candidate. latest_job_marker guarantees this chronological judgment by scanning reversed(lines).

Note

The common goal of this script's unit tests is not "prove the clever logic is correct" but "pin down, ahead of time, the boundaries a naive implementation gets wrong." The OR logic in calendar_slot_passed, the old-vs-new marker judgment, suppressing the probe when there are zero candidates—each of these flips its result depending on how a single line is written, and none of them are easy to notice until you actually run it. The value of writing pytest for a personal automation script isn't to convince a reviewer; it's so that six months from now, when you change the spec, you don't step on the same mistake again.

JOB_KILL_GRACE_SECONDS=30 providing the SIGTERM→SIGKILL grace periodterminate_job, the previous job stays alive while the next one gets kicked, causing the load-over-40 collateral damageprobe_claude() when there are candidatesStartInterval is present, calendar_slot_passed is unconditionally Falsereversed(lines) --dry-run turns verification cost into execution costis_guarded / latest_marker_is_today_skip / Next time I plan to write about the circuit breaker itself—how claude-quota-guard.py distinguishes a genuine "limit reached" message from a successful run whose article body just happens to contain the same words.

How do you handle catch-up for scheduled jobs that got skipped in your own setup—do you re-run them, or just wait for the next slot?

*Written by Lily — I ship iOS apps and automate my content stack with Claude Code.

Follow along: Portfolio · X · GitHub*