← all builds

Dead Man's Switch

monitoringreliabilityalerting
Download .md

A sanitised reference implementation. Drop it into Claude Code, point it at your own scheduled jobs, adapt the windows. Generic on purpose, with no business specifics and no real schedules.

The problem

A scheduled job can fail two different ways. It can stop running at all, which is the easy case to catch. Or it can keep running, keep reporting success, and quietly stop doing the one thing it exists to do, storing something. A watcher that only asks "did it run" misses the second kind completely, and that kind is the one that does real damage, because everything looks fine right up until someone needs the data that was never actually saved.

The idea

Give every watched job two separate clocks, not one. The first tracks the last time it succeeded at all. The second, usually a longer window, tracks the last time it actually produced something, a stored row, a written file. A job can be well within its first window and still be flagged, because the second one has run out. Every watched job also carries its own short hint, written in advance, about what to actually go and check if it goes quiet, so the first response to an alert is a look in the right place, not a cold start.

for each job: last success ──> within window? ──yes──> last output ──> within
                  │no                                   quiet window?  │no
                  v                                          │yes      v
                STALE                                        OK      QUIET

Minimal reference implementation

# dead_mans_switch.py, reference only. Swap in your own job list and log source.
from datetime import timedelta

# name, success window, quiet window, hint. The quiet window is usually longer
# than the success window, for a job that only expects new data every so often.
WATCHED_JOBS = (
    ("example_job", timedelta(hours=2), timedelta(days=3),
     "check the job's own log for an error line before assuming it is dead"),
)

def judge(last_success, window, now, woke_recently=False):
    """A job that has never once succeeded is always stale. A job waking from
    sleep gets one cycle of grace before it is called stale, never longer."""
    if last_success is not None and now - last_success <= window:
        return "ok"
    if last_success is not None and woke_recently:
        return "grace"
    return "stale"

def check(job_state, now, woke_recently=False):
    """job_state: {name: (last_success, last_output)}. Returns one finding per
    watched job. Never raises; a job it cannot read is its own finding."""
    findings = []
    for name, window, quiet_window, hint in WATCHED_JOBS:
        last_success, last_output = job_state.get(name, (None, None))
        state = judge(last_success, window, now, woke_recently)
        quiet = judge(last_output, quiet_window, now, woke_recently) == "stale"
        findings.append({
            "name": name, "state": state, "quiet": quiet,
            "flagged": state == "stale" or quiet,
            "hint": hint if quiet else None,
        })
    return findings

Why it works

Adapt it

Built for my own set of scheduled jobs, runs at the start of every working session, and this is the generalised version. Take it, point it at your own job list, tell me what you would add.