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() from cron, launchd, or a systemd timer. Each tick asks honker for due fires and runs their callbacks.

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

[1787735375, 1787735376, 1787735377]

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 1787735375', 'ref': 1, 'used': 0, 'offset': 1}, {'title': 'heartbeat', 'body': 'beat 2 at 1787735376', 'ref': 2, 'used': 0, 'offset': 2}, {'title': 'heartbeat', 'body': 'beat 3 at 1787735377', '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

'7 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 Paninian system’s use of formal rules (sutras) and meta-rules (“meta-language”) is an early example of generative grammar, influencing modern linguistics.',
 '',
 '**SQLite internals:** SQLite’s “B-tree” storage and its zero-configuration, self-contained transactional engine reveal how complex data management can be achieved in a lightweight, embeddable database.']

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.

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)

Run the tick

The process that calls tick must register the callbacks first. A small script is enough.

# tick.py
from pobblebonk.core import Pob

pob = Pob('~/.pobblebonk/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 it once a minute. A tick with nothing due is one SQL call.

* * * * * cd ~/myapp && uv run python tick.py

Use a launchd StartInterval of 60 on macOS or a systemd timer with OnCalendar=minutely on Linux.

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 nbdev-test
uv run nbdev-clean

nbs/00_core.ipynb contains the implementation and tests. nbs/index.ipynb generates this README.