heartbeat

what wakes the tick on this machine

There is no scheduler daemon. Pob.tick runs only when an outside process calls it. This module installs that recurring call in the machine scheduler.

The machine heartbeat knows only its interval. pobblebonk keeps schedule times, pauses, retries, catch-up policy, and list requirements in SQLite. Mirroring those values into the machine scheduler would create a second schedule definition.

CronBeat writes a crontab line. LaunchdBeat writes a LaunchAgent. TaskBeat calls schtasks. They share the Beat interface, and beat() selects the implementation for this machine.

Three machines, three schedulers. CronBeat writes a crontab line. LaunchdBeat writes a LaunchAgent, which is what macOS actually wants. TaskBeat calls schtasks. They share the Beat interface, and beat() picks.

One tag names everything: the launcher file, its log, and the job in whichever scheduler. Change it and you have a second, independent heartbeat, which is how one machine ticks two databases.

every is the heartbeat interval in seconds. It defaults to 60. Portable intervals are whole minutes that divide an hour: 60, 120, 300, 600, 900, 1200, 1800, or 3600 seconds. cron and Task Scheduler set that common limit.

---

source

every_mins

def every_mins(
    every:int=60
)->int:

Whole minutes between portable heartbeat runs.

The launcher

Every backend schedules one launcher file. cron, launchd, and schtasks disagree about command quoting. schtasks /tr also limits command length. A launcher gives each scheduler one path and gives the user a command they can run directly.

cmd is a shell command such as uv run --directory ~/app python -m app.tick. The launcher writes it without escaping. >> appends each result to the log. newline='' preserves \r\n in the Windows file. Without it, Python writes \r\r\n, which cmd reads as a blank line.

---

source

launcher

def launcher(
    cmd:str, # the command one beat runs, as you would type it
    dirn:NoneType=None, # where to write it; `~/.pobblebonk` by default
    tag:str='pobblebonk', # names this file and its log
    win:bool=None, # write a Windows `.cmd`; this platform decides by default
)->Path:

Write the one file every backend schedules, and give back its path.

The point of writing a file rather than a command string is that the file runs. So run it:

import subprocess
from tempfile import mkdtemp

d = Path(mkdtemp())
sh = launcher('/bin/echo ticked', d, tag='demo')

test_eq(sh.name, 'demo.sh')
test_eq(subprocess.run([str(sh)]).returncode, 0)
test_eq((d/'demo.log').read_text(), 'ticked\n')

# A second beat appends. A log holding only the most recent tick would be useless the one
# morning you go looking for what happened at 3am.
subprocess.run([str(sh)])
test_eq((d/'demo.log').read_text().count('ticked'), 2)

# The same command, told it is on Windows: a different file, and a shell that has never heard
# of `#!`.
cmd = launcher('py -m app.tick', d, tag='demo', win=True)
test_eq(cmd.name, 'demo.cmd')
test_eq(cmd.read_text().splitlines()[0], '@echo off')

What a scheduler has to do

install replaces the job for a tag. installed returns its launcher path. uninstall removes it. Replacement prevents two processes from racing for the same fires.

Each backend uses an operating-system identity: a crontab comment, a plist filename, or schtasks /f.

---

source

Beat

def Beat(
    *args, **kwargs
):

One machine’s way of running a file at a portable interval.

install replaces what tag installed before. installed reads back what is really scheduled, not what we hoped, so it is the honest way to answer “is this thing on”.

cron

A user has one crontab, which can contain unrelated jobs. python-crontab finds this job by its trailing comment and preserves the other entries. It is available through the pobblebonk[cron] extra.

tabfile redirects these operations to another crontab. The tests use a temporary file.

---

source

CronBeat

def CronBeat(
    tabfile:NoneType=None, # a crontab file to edit; None is the live user crontab
):

A line in the user crontab. Linux, and anywhere without something better.

---

source

cron_line

def cron_line(
    every:int=60
)->str:

The portable cron expression for a heartbeat interval.

A crontab with somebody’s nightly backup already in it. Everything below is about that line surviving.

tabfile = d/'crontab'
tabfile.write_text('# nightly backup\n0 3 * * * /usr/bin/backup\n')

cron = CronBeat(tabfile)
cron.install('/home/k/.pobblebonk/pobblebonk.sh', every=120)

print(tabfile.read_text())

# nightly backup
0 3 * * * /usr/bin/backup
*/2 * * * * /home/k/.pobblebonk/pobblebonk.sh # pobblebonk

test_eq(cron.installed(), '/home/k/.pobblebonk/pobblebonk.sh')
assert '*/2 * * * * /home/k/.pobblebonk/pobblebonk.sh' in tabfile.read_text()
assert '0 3 * * * /usr/bin/backup' in tabfile.read_text()

# Installing again replaces the line rather than adding a second one.
cron.install('/home/k/.pobblebonk/pobblebonk.sh', every=120)
test_eq(tabfile.read_text().count('pobblebonk.sh'), 1)

# Removing ours removes only ours, and says whether there was anything there.
test_eq(cron.uninstall(), True)
test_eq(cron.installed(), None)
test_eq(cron.uninstall(), False)
assert '0 3 * * * /usr/bin/backup' in tabfile.read_text()

launchd

macOS cron jobs run without Full Disk Access. A callback that reads ~/Documents can therefore fail. cron also drops times missed while the machine sleeps. launchd runs one missed StartInterval after waking, which lets catchup process missed fires.

A LaunchAgent is a plist in ~/Library/LaunchAgents. Its filename is its identity. Installation writes the plist, runs bootout for the previous job, then runs bootstrap for the replacement.

_run calls a scheduler binary. With check=True, a missing binary or rejected command raises an exception. This prevents a failed installation from appearing successful. Output is decoded here because schtasks returns UTF-16.

---

source

LaunchdBeat

def LaunchdBeat(
    dirn:NoneType=None, # where agents live; `~/Library/LaunchAgents` by default
):

A LaunchAgent. What macOS wants, and it survives a sleeping laptop where cron does not.

---

source

plist

def plist(
    script, # the launcher to run
    tag:str='pobblebonk', # the launchd label, and the plist filename
    every:int=60, # seconds between beats
)->bytes:

The job description launchd reads: run script every every seconds.

launchd rejects invalid plists and ignores unknown keys. The checks parse the plist and inspect the launchd key names.

got = plistlib.loads(plist('/home/k/.pobblebonk/pobblebonk.sh', every=120))
test_eq(got['Label'], 'pobblebonk')
test_eq(got['ProgramArguments'], ['/home/k/.pobblebonk/pobblebonk.sh'])
test_eq(got['StartInterval'], 120)

Reading an agent back and removing it are plain file operations, so they run anywhere. Only the launchctl call in install needs a Mac, and it is the one line here that this notebook cannot check.

agents = d/'LaunchAgents'
agents.mkdir()
(agents/'pobblebonk.plist').write_bytes(plist('/home/k/.pobblebonk/pobblebonk.sh'))

launchd = LaunchdBeat(agents)
test_eq(launchd.installed(), '/home/k/.pobblebonk/pobblebonk.sh')
test_eq(launchd.uninstall(), True)
test_eq(launchd.installed(), None)
test_eq(launchd.uninstall(), False)

Task Scheduler

schtasks /f replaces a task with the same name. /sc minute /mo N runs it every N minutes. The launcher path stays unquoted because subprocess quotes arguments that contain spaces.

/query /xml reads the installed command without depending on the display language. The XML is UTF-16, uses a namespace, and can quote the command.

---

source

TaskBeat

def TaskBeat(
    *args, **kwargs
):

A Windows Task Scheduler entry, through schtasks.

---

source

task_args

def task_args(
    script, # the launcher to run
    tag:str='pobblebonk', # the task name
    every:int=60, # seconds between beats
)->list:

The schtasks command that creates the task. /f is what makes a second install replace the first.

The Windows checks inspect the schtasks command without running it. /f proves installation replaces the existing task.

args = task_args(r'C:\Users\k\.pobblebonk\pobblebonk.cmd', every=120)
assert '/f' in args
test_eq(args[args.index('/tn')+1], 'pobblebonk')
test_eq(args[args.index('/mo')+1], '2')
test_eq(args[-1], r'C:\Users\k\.pobblebonk\pobblebonk.cmd')

# What `schtasks /query /xml ONE` gives back: UTF-16, namespaced, and the command quoted whether
# or not it needed to be.
xml = '''<?xml version="1.0" encoding="UTF-16"?>
<Task version="1.2" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
  <Triggers><TimeTrigger><Repetition><Interval>PT1M</Interval></Repetition></TimeTrigger></Triggers>
  <Actions Context="Author">
    <Exec><Command>"C:\\Users\\k\\.pobblebonk\\pobblebonk.cmd"</Command></Exec>
  </Actions>
</Task>'''

test_eq(_task_command(_text(xml.encode('utf-16'))), 'C:\\Users\\k\\.pobblebonk\\pobblebonk.cmd')

Installing one

beat selects the scheduler for the current platform. install writes the launcher, schedules it, and returns its path.

on accepts a specific Beat. Tests use it to exercise the public path without changing the machine scheduler.

---

source

uninstall

def uninstall(
    tag:str='pobblebonk', on:Beat=None
)->bool:

Stop the beat. False if there was nothing scheduled.

---

source

installed

def installed(
    tag:str='pobblebonk', on:Beat=None
)->str:

The launcher tag runs now, or None.

---

source

install

def install(
    cmd:str, # the command one beat runs
    dirn:NoneType=None, # where the launcher goes; `~/.pobblebonk` by default
    tag:str='pobblebonk', # names the launcher, its log, and the scheduled job
    on:Beat=None, # a scheduler of your own; this machine's by default
    every:int=60, # seconds between portable heartbeat runs
)->Path:

Write the launcher for cmd and schedule it every every seconds.

---

source

beat

def beat(
    platform:str=None
)->Beat:

The scheduler this machine actually has.

test_eq(type(beat('linux')), CronBeat)
test_eq(type(beat('darwin')), LaunchdBeat)
test_eq(type(beat('win32')), TaskBeat)

The macOS smoke test installs a real LaunchAgent under pobblebonk-demo. It uses the one-minute default so the check finishes quickly. /bin/date appends each timestamp to ~/.pobblebonk/pobblebonk-demo.log. These cells change the machine scheduler, so nbdev does not run them automatically.

def launchd_smoke(tag='pobblebonk-demo'):
    import time
    from datetime import datetime

    on = LaunchdBeat()
    script = HOME/f'{tag}.sh'
    log = script.with_suffix('.log')
    uninstall(tag, on=on)
    script.unlink(missing_ok=True)
    log.unlink(missing_ok=True)
    try:
        script = install('/bin/date', tag=tag, on=on)
        test_eq(installed(tag, on=on), str(script))
        for _ in range(75):
            if log.exists() and (lines := log.read_text().splitlines()): break
            time.sleep(1)
        else: raise AssertionError(f'no timestamp reached {log}')
        datetime.strptime(lines[-1], '%a %b %d %H:%M:%S %Z %Y')
        return lines[-1]
    finally:
        uninstall(tag, on=on)
        script.unlink(missing_ok=True)
        log.unlink(missing_ok=True)

stamp = launchd_smoke()
print(stamp)

Sun Aug 30 13:29:52 AEST 2026

demo_tag = 'pobblebonk-demo'
demo = LaunchdBeat()
test_eq(installed(demo_tag, on=demo), None)
test_eq(demo._p(demo_tag).exists(), False)
test_eq((HOME/f'{demo_tag}.sh').exists(), False)
test_eq((HOME/f'{demo_tag}.log').exists(), False)

What this does not do

This module does not run a tick or interpret schedules. It installs one launcher that calls the tick at the configured heartbeat interval. Schedule timing stays in the database for pause, retries, and catchup.

Two ticks can overlap when a callback takes longer than the heartbeat interval. The heartbeat does not prevent this. A database lock must serialize ticks on every supported platform.