Health Checks and Heartbeats for Long-Running CLIs

The worker has been "running" for three days. systemctl status says active, docker ps says up, the process is in ps. It has also not processed a single event since Tuesday, because a network call hung without a timeout and the loop never came back. Or the nightly sync has not failed in a month — because the timer was disabled during a migration and it has not run in a month. A process that exists is not the same as a process that is doing its job, and nothing notices the difference unless you give it a way to. This guide adds a heartbeat to a long-running CLI command, a health subcommand that turns it into an exit code for probes and scripts, and a dead-man switch for scheduled jobs whose failure mode is not running at all. It belongs to the long-running and watch-mode topic.

Prerequisites

• Python 3.10+, Typer and httpx (for the dead-man switch ping).
• A long-running loop, like the one on the topic overview, or a scheduled command as in running a CLI on a schedule with cron and systemd.
• A state directory for the heartbeat file, from storing app data with platformdirs.

Three questions, three checks

"Is it healthy?" hides three different questions, answered by different mechanisms and acted on by different things:

Liveness asks whether the loop is still making progress. If not, the right response is to restart the process — which is what Docker's HEALTHCHECK and Kubernetes liveness probes do. Readiness asks whether the process can do useful work right now — whether its dependencies are reachable — and is used to hold off traffic or raise an alert rather than restart. The dead-man switch asks whether a scheduled job ran at all, which no in-process check can answer, because a job that never starts never reports anything.

The recipe: a heartbeat

The simplest liveness signal is a small file that the loop rewrites after every unit of work. Its age is the answer to "when did this last make progress?", and anything that can read a file — a health command, a probe, a monitoring agent — can check it.

# src/mytool/heartbeat.py
from __future__ import annotations

import json
import os
import tempfile
import time
from dataclasses import asdict, dataclass
from pathlib import Path


@dataclass(frozen=True)
class Beat:
    at: float            # wall-clock seconds since the epoch
    pid: int
    processed: int       # total units of work so far
    last_error: str | None = None


def write_beat(path: Path, processed: int, last_error: str | None = None) -> None:
    """Atomically replace the heartbeat file."""
    path.parent.mkdir(parents=True, exist_ok=True)
    beat = Beat(at=time.time(), pid=os.getpid(), processed=processed, last_error=last_error)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".heartbeat.")
    with os.fdopen(fd, "w", encoding="utf-8") as fh:
        json.dump(asdict(beat), fh)
    os.replace(tmp, path)


def read_beat(path: Path) -> Beat | None:
    try:
        return Beat(**json.loads(path.read_text(encoding="utf-8")))
    except (FileNotFoundError, json.JSONDecodeError, TypeError):
        return None


def check(path: Path, max_age: float, now: float | None = None) -> tuple[bool, str]:
    beat = read_beat(path)
    if beat is None:
        return False, "no heartbeat yet"
    age = (now if now is not None else time.time()) - beat.at
    detail = f"last heartbeat {age:.0f}s ago (processed {beat.processed:,})"
    if beat.last_error:
        detail += f"; last error: {beat.last_error}"
    return age <= max_age, detail

The heartbeat is written with the same temporary-file-and-rename pattern as any state file, so a reader never sees a half-written JSON document — see writing files atomically in Python CLIs. Recording a running count of processed units and the last error turns the heartbeat into a useful status page as well as a liveness signal.

The health command

# src/mytool/cli.py
import time
from pathlib import Path

import httpx
import typer

from mytool.heartbeat import check, write_beat

app = typer.Typer()
HEARTBEAT = Path.home() / ".local/state/mytool/heartbeat.json"


@app.callback()
def main() -> None:
    """Event worker."""


@app.command()
def work(interval: float = 5.0, once: bool = False) -> None:
    """Process events until stopped."""
    processed, last_error = 0, None
    while True:
        try:
            processed += process_batch()
            last_error = None
        except Exception as exc:
            last_error = f"{type(exc).__name__}: {exc}"
        write_beat(HEARTBEAT, processed, last_error)     # after every unit, success or not
        if once:
            break
        time.sleep(interval)


@app.command()
def health(max_age: float = typer.Option(120, help="Seconds before the heartbeat is stale.")) -> None:
    """Exit 0 if the worker made progress recently, 1 otherwise."""
    ok, detail = check(HEARTBEAT, max_age)
    typer.echo(f"{'ok' if ok else 'stale'}: {detail}")
    raise typer.Exit(0 if ok else 1)


@app.command()
def nightly(ping_url: str = typer.Option(None, envvar="MYTOOL_PING_URL")) -> None:
    """A scheduled job that reports success to a dead-man switch."""
    count = process_batch()
    typer.echo(f"processed {count} items", err=True)
    if ping_url:
        try:
            httpx.get(ping_url, timeout=10)
        except httpx.HTTPError as exc:
            typer.echo(f"warning: could not report success: {exc}", err=True)


def process_batch() -> int:
    return 0   # stand-in for real work


if __name__ == "__main__":
    app()

Writing the heartbeat after every unit, including failed ones, is deliberate: a loop that keeps failing but keeps trying is alive (and the last_error field says why it is unhappy), whereas a loop that stops writing heartbeats is stuck. Choose max_age as a few multiples of the longest normal interval between beats, so one slow batch does not trigger a restart.

Wiring it to probes

The health command's exit code is its whole API, which makes it usable by anything that runs commands:

# Dockerfile
HEALTHCHECK --interval=30s --timeout=5s --start-period=60s --retries=3 \
  CMD ["mytool", "health", "--max-age", "120"]

# Kubernetes container spec
livenessProbe:
  exec:
    command: ["mytool", "health", "--max-age", "120"]
  initialDelaySeconds: 60
  periodSeconds: 30
  failureThreshold: 3

Keep the health command fast and dependency-free: it runs every thirty seconds, often in a resource-limited container, and a health check that imports your whole application or calls the network adds load and new ways to fail. Reading one small file is ideal. If startup time matters here, the techniques in lazy-loading subcommands for faster startup keep mytool health from importing the heavy parts of the tool.

The dead-man switch

A heartbeat catches a job that is running but stuck. It cannot catch a job that is not running — a cron entry deleted during a migration, a timer never re-enabled, a machine that was decommissioned. For scheduled jobs, the reliable pattern inverts the check: at the end of each successful run, the job pings an external service, and that service alerts when an expected ping does not arrive. Hosted services such as Healthchecks.io and Cronitor work this way, as do Prometheus Pushgateway plus an alert on a stale timestamp, and most uptime monitors' "heartbeat" check types.

The nightly command above does it in three lines: an optional ping URL from the environment, a short timeout, and a warning rather than a failure if the ping itself cannot be sent — the job succeeded, and failing it because the monitor was unreachable would be wrong. Only ping on success; a failed run that pings anyway defeats the purpose.

UX considerations

• Human-readable and machine-readable at once. One line of text for the person reading probe logs, and the exit code for the probe. Add --json if other tooling wants the details.
• Include context in the output. "stale: last heartbeat 9m ago; last error: ConnectTimeout" often diagnoses the problem without opening another log.
• Report the heartbeat in status. When a person runs mytool status, show the last heartbeat and last scheduled run. Humans notice stale timestamps faster than monitors get configured.
• Do not restart on readiness. If the API the worker depends on is down, restarting the worker changes nothing. Record dependency failures in last_error and alert on them; reserve restarts for genuine lack of progress.
• Start-up grace. Configure probes with an initial delay, or have health treat "no heartbeat yet within N seconds of process start" as healthy, so slow start-ups are not killed in a loop.

Testing the behaviour

The health logic is pure — a file and a clock — so it tests cleanly with an injected now:

# tests/test_health.py
import time

from typer.testing import CliRunner

from mytool import cli
from mytool.heartbeat import check, read_beat, write_beat

runner = CliRunner()


def test_fresh_heartbeat_is_healthy(tmp_path):
    hb = tmp_path / "heartbeat.json"
    write_beat(hb, processed=10)
    ok, detail = check(hb, max_age=60)
    assert ok and "processed 10" in detail


def test_stale_heartbeat(tmp_path):
    hb = tmp_path / "heartbeat.json"
    write_beat(hb, processed=10)
    ok, detail = check(hb, max_age=60, now=time.time() + 600)
    assert not ok


def test_missing_or_corrupt(tmp_path):
    hb = tmp_path / "heartbeat.json"
    assert check(hb, 60) == (False, "no heartbeat yet")
    hb.write_text("{not json")
    assert read_beat(hb) is None


def test_errors_are_reported_but_still_alive(tmp_path):
    hb = tmp_path / "heartbeat.json"
    write_beat(hb, processed=3, last_error="ConnectTimeout: api")
    ok, detail = check(hb, max_age=60)
    assert ok and "ConnectTimeout" in detail


def test_health_command_exit_codes(tmp_path, monkeypatch):
    monkeypatch.setattr(cli, "HEARTBEAT", tmp_path / "hb.json")
    assert runner.invoke(cli.app, ["health"]).exit_code == 1
    runner.invoke(cli.app, ["work", "--once"])
    result = runner.invoke(cli.app, ["health", "--max-age", "60"])
    assert result.exit_code == 0 and result.output.startswith("ok")

For the dead-man switch, replace the ping with an httpx.MockTransport or patch httpx.get and assert it is called exactly once on success and never on failure.

Conclusion

A long-running command should be able to answer "are you making progress?" cheaply and truthfully. Write an atomic heartbeat after every unit of work, expose it through a fast health command whose exit code probes can use, keep readiness problems in the heartbeat's error field rather than triggering restarts, and give scheduled jobs a dead-man switch so that silence itself raises the alarm. The cost is a few dozen lines; the payoff is never again discovering that a "running" worker stopped working days ago.

Frequently asked questions

Related

• Up: Long-running and watch-mode CLIs
• Handling SIGTERM and graceful shutdown
• Running a CLI on a schedule with cron and systemd
• Building a watch mode with watchfiles
• Choosing exit codes for CLI tools