monitor

Folder watches checked between turns, with a standing review that runs when files change.

This module covers two kinds of work that happen outside the turn you are typing in. A folder review reads the changes someone else made to a watched folder and reports on them at the start of your next turn. The beat, at the end of the page, runs scheduled work on a timer even when no session is open.

Folder reviews

Say a colleague, a script or another agent is editing files in a folder you care about, and you want the agent to look over each change. You ask it to watch the folder with instructions such as “check each change for correctness”. From then on, whenever files under that folder change, a reviewer reads the diff, follows the instructions, and its report is waiting for you in the next turn.

A watch is that standing request. The model opens one with shalya’s watch tool, as watch(target, kind='folder', instructions=..., pattern=...), where pattern is a list of globs that limits which files count. The tool stores the watch as a row in the vault, beside the page, search and reminder watches, so list_watches and cancel_watch see it too. Monitors is the side that does the reviewing. Monitors.sync copies the vault’s folder rows into this session, and Monitors.check calls it first, so a watch the model opens is picked up on the next check. A host without a vault offers the model no way to open a folder watch, but code can still open one with Monitors.add.

A new watch records the folder as it stands as its baseline, and only changes after that are reviewed. The reviewer gets the watch’s instructions and the changes, and nothing from the conversation that created the watch, so the instructions have to say everything. add refuses a watch without them.

Each completed review goes into a queue, and the agent adds the queued reviews to the next prompt it sends, through review_notice. The review is also noted to the user and filed in memory under one key per watch, so a newer review of the same folder replaces the older one. The agent runs Monitors.check in a background thread between turns.

Change detection

snapshot records the text of every readable file under the watched folder that matches the pattern, and changed compares two snapshots and returns the files added, edited or removed. A file larger than SNAP_MAX_BYTES is recorded by its size alone, so a change to it shows as a changed size with no diff. report turns the changes into a short list of files with their added and removed line counts, followed by the unified diffs, clipped to REVIEW_MAX_CHARS, and summarise gives the one-line count such as 1 added, 1 edited.

A watch’s settle window, 20 seconds by default, is the least time between two reviews of one folder. Changes that arrive inside the window wait and go into the next review together, which keeps a burst of saves from producing a burst of reviews. check(force=True) looks anyway.

Reviewer

The reviewer is a read-only sub-agent: a second agent with the session’s read tools and none that write, briefed with REVIEW_SP, and limited to REVIEW_MAX_STEPS steps. It cannot edit the watched folder, so it describes any fix instead of making it. Without a model to run it on, the change is recorded with status unreviewed. The baseline moves forward after every attempt, including a failed one, so a change is never reviewed twice.

---

source

secs

def secs(
    every
):

Seconds from '30s', '5m', '1h', or a number.

---

source

changed

def changed(
    before, after
):

{path: (before, after)} for every file added, edited or removed between two snapshots.

---

source

snapshot

def snapshot(
    host, folder, pattern:str=''
):

{path: text} for every watched file as it is now.

---

source

files_under

def files_under(
    host, folder, pattern:str=''
):

Returns readable files under folder matching pattern.

---

source

report

def report(
    changes, folder:str='', mx:int=24000
):

Summarize changed files and clip their unified diffs.

---

source

summarise

def summarise(
    changes
):

One line: how many files were added, edited and removed.

---

source

review_notice

def review_notice(
    recs, mx:int=24000
):

Reviews since the last turn, as a prompt block.

---

source

review_prompt

def review_prompt(
    instructions, changes, folder:str=''
):

The question a reviewer gets: the instructions, then the changes.

---

source

FolderWatch

def FolderWatch(
    folder, # folder or file to watch, inside the roots
    instructions, # the reviewer's brief, verbatim
    pattern:str='', # globs to limit it to; empty for all
    settle:str='20s', # least time between two reviews
    note:str='', # why it is being watched
):

One monitored folder: its target, instructions and last snapshot.

---

source

Monitors

def Monitors(
    host, get_backend:NoneType=None, # callable -> review backend, or None
    get_tools:NoneType=None, # callable -> the reviewer's read tools
    on_review:NoneType=None, # callable(record) per completed review
    log_dir:NoneType=None, # callable -> folder for transcripts and logs
):

Folder watches and reviews: check looks, drain hands reviews to the next turn.

---

source

Monitors.check

def check(
    force:bool=False, # look even inside a watch's settle window
    block:bool=True, # wait for a running check, not return `None`
):

One record per changed folder, or None if another check runs and block is off.

---

source

Monitors.sync

def sync():

Mirror the host’s folder watches: add new vault rows, drop removed ones.

from ramabana.testing import FakeBackend, MemHost, SPEC

The example watches a two-file project held in memory by MemHost, with FakeBackend standing in for the reviewer’s model; it answers every question with sub answer. Nothing has changed right after add, so the first check returns nothing. After one file is edited, the check returns one record holding the summary, the review and the diff.

h = MemHost({'/proj/a.py': 'def a(): return 1\n', '/proj/notes.md': 'hello\n'})
m = Monitors(h, get_backend=lambda: FakeBackend(SPEC))
w = m.add('/proj', 'Review each change for correctness.')
test_eq(len(w.snap), 2)
test_eq(m.check(), [])

h.write('/proj/a.py', 'def a(): return 2\n')
rec, = m.check()
test_eq((rec['status'], rec['files'], rec['summary']), ('ok', 1, '1 edited'))
test_eq(rec['review'], 'sub answer')
assert '-def a(): return 1' in rec['changes'] and '+def a(): return 2' in rec['changes']

A file added inside the settle window waits, so the plain check returns nothing and force=True reviews it. A removed file shows as removed. Each check also queued its record, and drain hands them all over once, which is what the agent does before its next turn.

h.write('/proj/b.py', 'def b(): pass\n')
test_eq(m.check(), [])
rec, = m.check(force=True)
test_eq((rec['summary'], rec['status']), ('1 added', 'ok'))

del h.files['/proj/notes.md']
test_eq(m.check(force=True)[0]['summary'], '1 removed')

test_eq(len(m.drain()), 3)
test_eq(m.drain(), [])
test_eq(review_notice([]), '')

test_eq(_globs('*.py, *.ipynb'), ['*.py', '*.ipynb'])
test_eq(summarise({'a': ('', 'x'), 'b': ('x', 'y')}), '1 added, 1 edited')
test_fail(lambda: m.add('/proj', '  '), contains='needs instructions')

check(block=False) returns None at once when another check is already running, instead of waiting for it. The agent’s background thread uses the blocking form.

h.write('/proj/a.py', 'def a(): return 3\n')
m.checking.acquire()
try: test_eq(m.check(block=False), None)
finally: m.checking.release()
test_eq(m.check(force=True)[0]['summary'], '1 edited')

The same flow through the model’s tools. VaultLike is a stand-in for a vault host, which stores watches as rows and keeps notes in memory. The watch tool opens a folder watch, the next check picks it up through sync and takes its baseline, an edit produces a review filed under the key folder-review:w1, and cancel_watch makes the watch disappear from Monitors on the next check.

from ramabana.tools import MemoryHost, WatchHost, watch_tools

class VaultLike(MemHost, MemoryHost, WatchHost):
    "A host whose watches table is a list and whose memory is the notes it was handed."
    def __init__(self, files): super().__init__(files); self.rows, self.notes = [], []
    def watch(self, target, action='remind', every='1d', note=None, **params):
        w = dict(id=f'w{len(self.rows)+1}', kind=action, target=target, every=86400, note=note, runs=0, last_status='', **params)
        self.rows.append(w); return w
    def watches(self, due_only=False): return list(self.rows)
    def unwatch(self, wid): self.rows = [r for r in self.rows if r['id'] != wid]; return True
    def poll(self): return {'checked': 0, 'ran': 0, 'results': []}
    def remember(self, text, title=None, tags=(), key=''): self.notes.append((title, key, text)); return dict(doc_id='d1', title=title)
    def memory_search(self, query, limit=8): return []
    def memory_tree(self, document=''): return []
    def memory_read(self, node_id): return {}
    def memory_forget(self, doc_id): return False
    watch_actions = ('url', 'remind', 'web', 'folder')

vh = VaultLike({'/proj/a.py': 'one\n'})
vm = Monitors(vh, get_backend=lambda: FakeBackend(SPEC))
wt = {t.__name__: t for t in watch_tools(vh)}
print(wt['watch']('/proj', kind='folder', instructions='Review each change.'))
test_eq(vm.check(force=True), [])
test_eq([w.id for w in vm.all()], ['w1'])
vh.write('/proj/a.py', 'two\n')
rec, = vm.check(force=True)
test_eq((rec['watch_id'], rec['review']), ('w1', 'sub answer'))
test_eq(vh.notes[-1][:2], ('folder review: proj: 1 edited', 'folder-review:w1'))
test_eq(wt['cancel_watch']('w1'), 'cancelled w1')
test_eq(vm.check(), [])
test_eq(vm.all(), [])

watching folder /proj as w1, every 1d

The beat

Some work should run on a timer whether or not anyone has a session open: a nightly job that learns from the day’s sessions, say. Ramabana hands the timing to pobblebonk, a small scheduler that installs one job with the operating system’s own scheduler (cron, launchd or the Windows task scheduler) and keeps its schedules, the work they fired, and the notes that work left in a SQLite database.

A schedule is a name, an interval, and a callback that runs when the schedule fires. A beat is one pass over the schedules that runs every one that has come due. A note is a short title and body that a callback leaves for whoever reads next. Several readers can share one database, and each reader sees each note once.

This module holds Ramabana’s end of that arrangement. on_tick(name) registers the callback for a schedule from Python. pob opens the shared database, or returns None when pobblebonk is not installed. beat_notes takes the notes a reader has not seen yet, and beat_notice wraps them in a <beat> block, which the agent adds to its next prompt. Each session reads under its own reader name, so every open session sees every note. The packages that define recurring work own their schedules. Ramabana only runs them.

---

source

beat_notice

def beat_notice(
    notes, mx:int=24000
):

Beat notes since the last look, as a prompt block.

---

source

beat_notes

def beat_notes(
    db, reader:str='ramabana', limit:int=20
):

Beat notes reader has not read; empty on no beat or any error.

---

source

pob

def pob(
    path:NoneType=None
):

The database the beat and this session share, or None without pobblebonk.

---

source

pob_path

def pob_path():

---

source

on_tick

def on_tick(
    name
):

Register the callback a fire of schedule name runs.

tick is the ramabana-tick command. Run plainly, it registers the callbacks, runs one beat, prints how many schedules ran of those that fired, and exits. With no schedules registered it does nothing and still exits successfully. --install hands ramabana-tick --quiet to pobblebonk to run every --every seconds, and --uninstall removes it. beat_command builds that command line, quoting the path for the platform’s shell, and heartbeat returns pobblebonk’s scheduler module or None.

launchd and cron start a job on /usr/bin:/bin, which holds neither claude in ~/.local/bin nor Homebrew’s git-lfs. Off Windows, beat_command therefore runs the beat through /usr/bin/env with the PATH of the shell that installed it, and --install prints that PATH. Only PATH is carried: a key exported in a shell profile does not reach the beat. A tool installed in a new folder after --install is found once ramabana-tick --install runs again. Task Scheduler on Windows gives a job the user’s environment, so the Windows command is unchanged.

An installed package registers a callback through a ramabana.ticks entry point (TICK_GROUP). The entry point’s name is the schedule’s name and its value is the callback. The beat imports the package only when that schedule fires, so a slow or broken package costs nothing until then. A package that fails to import fails that one fire, and pobblebonk retries it or leaves a note. An on_tick callback with the same name replaces the entry point’s. Dhrona registers its nightly job this way:

[project.entry-points."ramabana.ticks"]
dhrona = "dhrona.improve:nightly"

---

source

tick

def tick(
    db:str='', # the shared database; pobblebonk's own by default
    quiet:bool=False, # say nothing on success
    install:bool=False, # schedule this command instead of running one beat
    uninstall:bool=False, # stop the scheduled beat
    every:int=60, # seconds between installed beats, whole minutes
):

One beat: run the schedules that are due and leave what they found as notes.

---

source

heartbeat

def heartbeat():

pobblebonk’s scheduler seam, or None when it is not installed.

---

source

beat_command

def beat_command(
    exe, path:NoneType=None
):

exe --quiet as the launcher’s shell reads it, quoted; off Windows it runs on path, the installing PATH by default.

if sys.platform != 'win32':
    assert beat_command('/a b/ramabana-tick', path="/x y/bin:/it's/$HOME:/usr/bin") == "/usr/bin/env PATH='/x y/bin:/it'\"'\"'s/$HOME:/usr/bin' '/a b/ramabana-tick' --quiet"
    assert beat_command('/x/ramabana-tick', path='') == '/x/ramabana-tick --quiet'

The example registers a schedule named demo that fires every second, waits for it to come due, and runs one beat against a database in a temporary folder. The callback’s return value becomes the note that beat_notes reads back.

import contextlib, io, tempfile, time

@on_tick('demo')
def _demo(fire): return 'the beat ran this'

db = str(Path(tempfile.mkdtemp())/'pob.db')
pob(db).add('demo', every='1s')
time.sleep(1.2)
out = io.StringIO()
with contextlib.redirect_stdout(out): test_eq(tick.__wrapped__(db=db), 0)
test_eq(out.getvalue().strip(), '1 ran of 1 fired')
test_eq(beat_notes(pob(db)), ['demo: the beat ran this'])
del TICKS['demo']

Updating packages:   0%|          | 0/31 [00:00<?, ?pkg/s]

parse files from /home/runner/work/ramabana/ramabana:   0%|          | 0/103 [00:00<?, ?it/s]

parse files from /home/runner/work/ramabana/ramabana:   2%|1         | 2/103 [00:00<00:08, 11.88it/s]/opt/hostedtoolcache/Python/3.12.15/x64/lib/python3.12/site-packages/usearch/__init__.py:131: UserWarning: Will download `usearch_sqlite` binary from GitHub.
  warnings.warn("Will download `usearch_sqlite` binary from GitHub.", UserWarning)


parse files from /home/runner/work/ramabana/ramabana:   6%|5         | 6/103 [00:00<00:04, 19.57it/s]

parse files from /home/runner/work/ramabana/ramabana:  17%|#6        | 17/103 [00:00<00:01, 49.85it/s]

parse files from /home/runner/work/ramabana/ramabana:  38%|###7      | 39/103 [00:00<00:00, 104.09it/s]

Notes can also be written straight to the database with note(title, body). A note with an empty body reads as its title alone:

import tempfile
beat = pob(Path(tempfile.mkdtemp())/'pob.db')
beat.note('watches', '2 of 5 fired')
beat.note('nightly', '')
beat_notes(beat)

['watches: 2 of 5 fired', 'nightly']

Once a reader has drained the notes, it gets nothing more, while another reader still sees all of them. beat_notes returns an empty list when there is no database, or when reading it fails, so a locked database never stops a turn.

test_eq(beat_notes(beat), [])
test_eq(beat_notes(beat, reader='another'), ['watches: 2 of 5 fired', 'nightly'])
test_eq(beat_notes(None), [])

class _Broken:
    def drain(self, reader, limit=20): raise RuntimeError('the database is locked')
test_eq(beat_notes(_Broken()), [])