Periodic background tasks, scheduled cron jobs, and long-running autonomous worker scripts share a common failure mode: silent death. When a background script crashes before it can write to a log file, runs into an unhandled system signal, or stops being triggered because a machine rebooted, standard monitoring tools that only parse logs often report nothing. In monitoring terminology, an absence of news is mistakenly treated as good news.

To solve this, reliable systems rely on a dead-man's switch. Instead of waiting for an explicit error message, the system tracks expected check-in intervals. If a job does not pulse within its designated window plus an allowed grace period, it is flagged as silent.

This tutorial walks through cron_agent_heartbeat_that_never_goes_silent_s.py, a self-contained Python script backed by SQLite that tracks pulses, logs runtimes, catches exit codes, detects overdue agents, and triggers notification commands.

The Problem: Silent Failure in Scheduled Tasks

Consider a cron task configured to sync database records every hour. If the process encounters a network partition, runs out of memory, or fails to start entirely, traditional log monitoring can leave you unaware for days.

Common issues include:

  1. Uncaught crashes: An unhandled error terminates the script before any application-level alerting logic can execute.
  2. Infinite hangs: A network call without timeouts freezes the worker indefinitely; subsequent jobs may either pile up or fail to start.
  3. Ghost schedules: An administrative change accidentally deletes a crontab entry or disables a system timer. The job simply stops running, and because no error is generated, no notification is fired.

To handle these scenarios, you need a mechanism that records when a task starts, logs when it completes, tracks its duration, and flags tasks that missed their expected deadlines.

Install & Usage

The script is entirely self-contained. It relies solely on the Python standard library (sqlite3, subprocess, argparse, datetime, shlex, json, and tempfile). No third-party packages need to be installed via pip.

To get started, download the source code, save it locally as cron_agent_heartbeat_that_never_goes_silent_s.py, and run it directly with your Python interpreter:

python cron_agent_heartbeat_that_never_goes_silent_s.py --help

By default, the script stores state in a local SQLite file located at ~/.agent_heartbeat.db.

You can integrate this tool into your workflows either by importing its functions into your Python automation scripts or by using it to wrap command executions.

Wrapping a Command in Python

When you want to execute a command while capturing its runtime, exit status, and alert triggers automatically, you can call execute_wrap:

import sqlite3 from cron_agent_heartbeat_that_never_goes_silent_s import ( DEFAULT_DB_PATH, execute_wrap, get_connection, ) conn = get_connection(DEFAULT_DB_PATH) # Run a Python script as an agent named "nightly-backup" # Expected interval: 86400 seconds (24h), grace period: 1800 seconds (30m) exit_code = execute_wrap( conn=conn, agent_name="nightly-backup", cmd_args=["python3", "/opt/scripts/backup.py"], interval_sec=86400, grace_sec=1800, alert_hook_cmd="echo 'Alert: {agent} failed with status {state}'", )

Inspecting Health Programmatically

To evaluate which agents are healthy, failing, or overdue:

from cron_agent_heartbeat_that_never_goes_silent_s import ( DEFAULT_DB_PATH, evaluate_agents, get_connection, ) conn = get_connection(DEFAULT_DB_PATH) health_report = evaluate_agents(conn) for agent in health_report: print( f"Agent: {agent['name']} | " f"State: {agent['state']} | " f"Overdue: {agent['seconds_overdue']}s" )

How It Works

The architecture of cron_agent_heartbeat_that_never_goes_silent_s.py centers on two SQLite tables: agents and heartbeat_history.

  • agents: Stores the registry of monitored tasks, their expected intervals, grace periods, alert hook templates, and running streaks of successes or failures.
  • heartbeat_history: Records an append-only log of every ping, including execution durations, exit codes, timestamps, and optional JSON payloads.

Deadline Evaluation and State Determination

The evaluate_agents function inspects every registered record in the SQLite database and calculates if the task has exceeded its execution window.

def evaluate_agents(conn: sqlite3.Connection) -> List[Dict[str, Any]]: """Scan all registered agents and classify current operational health.""" now_dt = datetime.datetime.now(datetime.timezone.utc) rows = conn.execute("SELECT * FROM agents ORDER BY name ASC").fetchall() results = [] for r in rows: name = r["name"] interval = r["expected_interval_sec"] grace = r["grace_sec"] last_ping_str = r["last_ping_utc"] last_status = r["last_status"] or "UNKNOWN" hook_cmd = r["alert_hook_cmd"] last_dt = parse_iso(last_ping_str) if last_dt is None: state = "NEVER_SEEN" seconds_overdue = 0.0 next_deadline_iso = None else: deadline = last_dt + datetime.timedelta(seconds=interval + grace) next_deadline_iso = deadline.replace(microsecond=0).isoformat() diff_sec = (now_dt - deadline).total_seconds() if diff_sec > 0: state = "SILENT" seconds_overdue = round(diff_sec, 1) elif last_status in ("FAIL", "FAILED", "ERROR"): state = "FAILING" seconds_overdue = 0.0 else: state = "HEALTHY" seconds_overdue = 0.0

The script evaluates the health into four clear states:

  • NEVER_SEEN: Registered in the database, but has never transmitted a ping.
  • HEALTHY: Checked in within interval + grace seconds and finished with a successful status.
  • FAILING: Checked in recently, but the latest run ended with an error status (FAIL, FAILED, or ERROR).
  • SILENT: The current time has surpassed the deadline (last_ping + interval_sec + grace_sec).

Execution Wrapping and Alert Dispatch

The execute_wrap function runs arbitrary CLI commands while shielding the caller from silent terminations. It updates the database status to RUNNING before execution, calculates the precise elapsed runtime using time.perf_counter(), records the final exit code, and invokes an alert hook if the exit code is non-zero.

def execute_wrap( conn: sqlite3.Connection, agent_name: str, cmd_args: List[str], interval_sec: int, grace_sec: int, alert_hook_cmd: Optional[str] = None, ) -> int: """Wrap command execution: logs start, records runtime, pings success/fail, triggers alerts.""" register_or_update_agent( conn, name=agent_name, interval_sec=interval_sec, grace_sec=grace_sec, alert_hook_cmd=alert_hook_cmd, ) record_heartbeat( conn=conn, name=agent_name, status="RUNNING", message=f"Started execution of: {' '.join(shlex.quote(a) for a in cmd_args)}", ) start_t = time.perf_counter() exit_code = 1 err_msg = "" try: proc = subprocess.run(cmd_args) exit_code = proc.returncode except FileNotFoundError: exit_code = 127 err_msg = f"Executable not found: {cmd_args[0]}" except Exception as exc: exit_code = 1 err_msg = str(exc) elapsed_ms = int((time.perf_counter() - start_t) * 1000) status = "OK" if exit_code == 0 else "FAIL" msg = f"Command completed with exit code {exit_code}" if not err_msg else err_msg res = record_heartbeat( conn=conn, name=agent_name, status=status, duration_ms=elapsed_ms, exit_code=exit_code, message=msg, ) if exit_code != 0 and alert_hook_cmd: agent_eval = { "name": agent_name, "state": "FAILING", "last_status": status, "seconds_overdue": 0.0, "last_ping_utc": res["timestamp_utc"], } trigger_alert_hook(alert_hook_cmd, agent_eval, dry_run=False) return exit_code

When an alert fires, trigger_alert_hook populates both template variables ({agent}, {state}, {overdue}) and environment variables (HB_AGENT_NAME, HB_AGENT_STATE, HB_LAST_STATUS, HB_SECONDS_OVERDUE, HB_LAST_PING), letting you route notifications to shell scripts, webhook pushers, or local loggers.

Limitations

While practical for standalone machines and single-host architectures, this implementation has deliberate design trade-offs:

  1. Local SQLite Concurrency: SQLite handles concurrent readers well, but concurrent writes lock the database file. If hundreds of jobs attempt to ping the watchdog in the exact same millisecond, you may encounter SQLite busy or lock errors.
  2. No Persistent Background Loop: The script does not run its own persistent daemon loop to detect silent tasks. It only evaluates whether a task is SILENT when evaluate_agents is executed. To catch overdue jobs, you must run an evaluation check on a separate schedule.
  3. Synchronous Alert Hooks: Alert hooks run via subprocess.run(shell=True) with a hardcoded 15-second timeout. If your alert command hangs on a slow network call, it will delay the execution wrapper until the timeout expires.
  4. Local Host Boundary: Because it defaults to writing to ~/.agent_heartbeat.db, agents running across multiple physical or virtual hosts cannot see each other's status unless they share the underlying file system.

FAQ

Does this tool require an external database server?

No. It uses Python's standard sqlite3 engine and writes by default to ~/.agent_heartbeat.db. No network connections, database servers, or port configurations are needed.

How does the watchdog know a job is silent versus failing?

A job is marked FAILING when it completes execution but reports an exit code other than 0 or a status such as FAIL or ERROR. It is marked SILENT when the time elapsed since its last ping exceeds its configured expected_interval_sec plus its grace_sec.

What environment variables are exposed to custom alert hooks?

When an alert command executes, trigger_alert_hook injects five environment variables into the subshell: HB_AGENT_NAME, HB_AGENT_STATE, HB_LAST_STATUS, HB_SECONDS_OVERDUE, and HB_LAST_PING. You can access these variables directly in shell scripts or reference the template placeholders {agent}, {state}, and {overdue} in the command string.

Full version and the rest of the toolkit: Cron/agent heartbeat that never goes silent s on PyForge. Product page: https://py-forge.com/products/cron-agent-heartbeat-that-never-goes-silent-s