pobblebonk

the clock and the notebook for an agent

pobblebonk adds callbacks, durable lists, and per-reader notes to honker. Schedules, queued work, list items, retries, and notes share one SQLite file.

There is no scheduler daemon. Call Pob.tick() on a heartbeat, and each tick asks honker for due fires and runs their callbacks. pobblebonk.heartbeat installs that recurring call through cron, launchd, or Task Scheduler.

Install

uv add pobblebonk

Python 3.12 or later is required.

Schedule a callback

Register a callback, add its schedule, then call tick. The callback return value becomes a note.

import time

pob = Pob()
beats = []

@pob.on('heartbeat')
def beat(fire):
    beats.append(fire.fire_at)
    return f'beat {len(beats)} at {fire.fire_at}'

pob.add('heartbeat', every='1s')
for _ in range(3):
    time.sleep(1.05)
    pob.tick()

beats

[1788060622, 1788060623, 1788060624]

fire.fire_at is the scheduled boundary, not the time the callback happened. The one-second gaps show that the cadence held.

[b-a for a, b in zip(beats, beats[1:])]

[1, 1]

drain returns the notes a reader has not seen. Each reader has an independent cursor.

pob.drain('leela')

[{'title': 'heartbeat', 'body': 'beat 1 at 1788060622', 'ref': 1, 'used': 0, 'offset': 1}, {'title': 'heartbeat', 'body': 'beat 2 at 1788060623', 'ref': 2, 'used': 0, 'offset': 2}, {'title': 'heartbeat', 'body': 'beat 3 at 1788060624', 'ref': 3, 'used': 0, 'offset': 3}]

pob.drain('leela'), len(pob.drain('phone'))

([], 3)

Run ordinary Python

A callback is a Python function. It can call a library, write a file, or run a command.

import subprocess

pob2 = Pob()

@pob2.on('git roundup')
def roundup(fire):
    out = subprocess.run(['git', 'log', '--oneline', '--since=30 days ago'],
                         capture_output=True, text=True).stdout.strip().splitlines()
    return f'{len(out)} commits in the last 30 days'

pob2.add('git roundup', cron='0 17 * * *')      # 5pm daily
time.sleep(1.05)
pob2.tick(at=int(time.time()) + 86400).ran[0].result

'17 commits in the last 30 days'

Give a callback a durable list

push adds an item to a named list. A schedule with needs runs only when that list is not empty. Its callback receives the open items as fire.list.

The callback marks its items used when it returns. If it raises, the items remain open for the retry. A key makes repeated open items idempotent.

pob3 = Pob()

@pob3.on('cart')
def cart(fire):
    return 'added: ' + ', '.join(i.text for i in fire.list)

pob3.add('cart', cron='0 20 * * 3', needs='shopping')     # 8pm on Wednesdays
pob3.push('shopping', 'yoghurt, the greek one', key='img_2201.heic')
pob3.push('shopping', 'yoghurt, the greek one', key='img_2201.heic')   # the same photo twice
pob3.push('shopping', 'oat milk')
pob3.items('shopping').attrgot('text')

['yoghurt, the greek one', 'oat milk']

time.sleep(1.05)
got = pob3.tick(at=int(time.time()) + 7*86400).ran[0]
got.status, got.result, got.used

('ok', 'added: yoghurt, the greek one, oat milk', 2)

When the list is empty, the next fire is skipped. It is not a callback failure and produces no note.

time.sleep(1.05)
nxt = pob3.tick(at=int(time.time()) + 14*86400).ran[0]
nxt.status, nxt.why

('skipped', 'the shopping list is empty')

Use a model callback

A model is another callable dependency. The documentation build does not run this example because it needs a LiteRT model.

import rishi

pob4 = Pob()

@pob4.on('news')
def digest(fire):
    chat = rishi.Chat('gpt-4.1')
    topics = ', '.join(i.text for i in fire.list)
    return rishi.resp_text(chat(f'Name one thing worth reading about each of: {topics}. One line each.'))

pob4.add('news', every='1s', needs='interests')
pob4.push('interests', 'sanskrit grammar')
pob4.push('interests', 'sqlite internals')
time.sleep(1.05)
pob4.tick().ran[0].result.split('\n')

['- **Sanskrit grammar:** The concept of *Sandhi*—how sounds combine at word boundaries—reveals the language’s precision and flexibility.',
 '- **SQLite internals:** The B-tree structure in SQLite’s storage engine is central to how it efficiently manages and retrieves data on disk.']

Operate schedules

Use pause, resume, update, and drop to maintain schedules. update changes only the fields you pass. drop also unregisters the callback in the current process.

pob4.pause('news')
pob4.update('news', cron='30 7 * * *', retries=5)
pob4.resume('news')
pob4.drop('news')

True

# Queue due fires without running callbacks.
queued = pob.tick(run=False)

# Run queued fires separately, with a bounded batch.
results = pob.work(worker='scheduler', limit=100)

Failures and missed fires

A fire retries with exponential backoff when its callback raises. The default attempt budget is three. Set retries on Pob or on one schedule. After the final attempt, the fire is dead-lettered and a note records the error.

catchup='once' keeps the latest fire missed while the machine was off. This is the default. catchup='all' keeps every missed fire, and a positive integer keeps that many of the most recent. start= sets the first fire; without it the cadence picks the next boundary from now.

Use tick(run=False) when scheduling and callback execution belong in separate processes. It queues due fires without running them. work claims and runs queued fires.

pob = Pob('~/.pobblebonk/pob.db', retries=3)
pob.add('roundup', cron='0 17 * * *', catchup='once', retries=5)

RuntimeError: Database error: unable to open database file: ~/.pobblebonk/pob.db
---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
Cell In[13], line 2
      1 #| eval: false
----> 2 pob = Pob('~/.pobblebonk/pob.db', retries=3)
      3 pob.add('roundup', cron='0 17 * * *', catchup='once', retries=5)

File ~/code/pobblebonk/pobblebonk/core.py:65, in Pob.__init__(self, path, db, retries, base)
     58 def __init__(self,
     59              path=None,          # the file; None makes a temporary one
     60              db=None,            # or a honker database something else already opened
     61              retries:int=RETRIES,# attempts a failing fire gets before it is dead-lettered
     62              base:float=BASE):   # seconds before the first retry; doubles per attempt
     63     # honker watches `PRAGMA data_version` on a file, so `:memory:` is not an option
     64     if db is None and path is None: path = Path(mkdtemp())/'pob.db'
---> 65     self.db = db if db is not None else honker.open(str(path))
     66     self.retries, self.base = max(1, int(retries)), float(base)
     67     self.q, self.stream = self.db.queue(FIRES), self.db.stream(NOTES)

File ~/code/pobblebonk/.venv/lib/python3.13/site-packages/honker/_honker.py:1433, in open(path, max_readers, watcher_backend, watcher_poll_interval_ms)
   1412 def open(
   1413     path: str,
   1414     max_readers: int = 8,
   1415     watcher_backend: Optional[str] = None,
   1416     watcher_poll_interval_ms: Optional[int] = None,
   1417 ) -> Database:
   1418     """Open a Honker database at `path`.
   1419 
   1420     `watcher_backend` selects the update-detection strategy:
   (...)   1431     when lower idle CPU matters more than lowest-latency wakeups.
   1432     """
-> 1433     return Database(_core_open(
   1434         path,
   1435         max_readers=max_readers,
   1436         watcher_backend=watcher_backend,
   1437         watcher_poll_interval_ms=watcher_poll_interval_ms,
   1438     ))

File ~/code/pobblebonk/.venv/lib/python3.13/site-packages/honker/_honker.py:23, in _core_open(path, max_readers, watcher_backend, watcher_poll_interval_ms)
     21 def _core_open(path, max_readers, watcher_backend=None, watcher_poll_interval_ms=None):
     22     from honker._honker_native import open as _open
---> 23     return _open(
     24         path,
     25         max_readers=max_readers,
     26         watcher_backend=watcher_backend,
     27         watcher_poll_interval_ms=watcher_poll_interval_ms,
     28     )

RuntimeError: Database error: unable to open database file: ~/.pobblebonk/pob.db

Run the tick on a heartbeat

The application owns one Pob database and one tick script. The script registers every callback before it calls tick.

# tick.py
from pobblebonk.core import Pob

pob = Pob('/absolute/path/to/pob.db')

@pob.on('cart')
def cart(fire):
    return 'added: ' + ', '.join(item.text for item in fire.list)

if __name__ == '__main__': print(pob.tick())

Run the script once before scheduling it. This catches import and path errors directly.

/absolute/path/to/uv run --directory /absolute/path/to/app python tick.py

Install one machine heartbeat after the script works. The command must use absolute paths because schedulers have a small environment.

from pobblebonk.heartbeat import install, installed, uninstall

install('/absolute/path/to/uv run --directory /absolute/path/to/app python tick.py', every=300)
installed()

install creates ~/.pobblebonk/pobblebonk.sh and schedules it every five minutes. every is seconds and defaults to 60. Portable values are whole minutes that divide an hour. Output is appended to ~/.pobblebonk/pobblebonk.log. On Linux it installs a cron line. On macOS it installs a LaunchAgent. On Windows it installs a Task Scheduler job.

The heartbeat only calls tick. Schedule times, retries, pauses, and catch-up policy stay in SQLite.

Inspect the log when a callback does not run. Remove the machine job when the application no longer needs it.

from pathlib import Path

print(Path.home().joinpath('.pobblebonk/pobblebonk.log').read_text())
uninstall()
assert installed() is None

The Linux backend needs uv add 'pobblebonk[cron]'.

Share an existing database

Pass an open honker database to keep pobblebonk data beside another application.

import honker

db = honker.open('app.db')
pob = Pob(db=db)

Develop

uv sync --group dev
uv run nbdev-export
uv run --extra cron nbdev-test
uv run nbdev-clean

nbs/00_core.ipynb contains schedules, lists, notes, and tick execution. nbs/01_heartbeat.ipynb contains the machine heartbeat. nbs/index.ipynb generates this README.