Async file-based IPC: one side drop()s a file, another side – possibly a different process, possibly a different machine reachable only via a synced/shared directory (Dropbox, rsync, an NFS mount, …) – pickup()s it later, no direct connection between the two required. Named after the spy tradecraft term. This is the “agent-loop file relay” deferred out of 02_remote.ipynb’s original scope, now built out on its own – and it subsumes an earlier hand-rolled prototype (cycle.sh/watch.sh, a bash-only prompt/response relay used to drive a Claude Code session from a Jupyter cell on a different machine) whose good ideas are generalized here: Maildir’s atomic-rename trick (drop()/pickup()), and blocking-wait-on-mtime-change (watch(), next_prompt()) – reimplemented in Python rather than shelled out to stat, because bash’s 1-second mtime resolution left a real race between “checked, nothing pending” and “started waiting for the next change” (found the hard way: a drop landing in that gap could be missed forever).
MACHINE_INFO.md
Session subdirectories live under one parent dir per machine, so that parent is a natural place to describe the machine itself – RAM, CPU count, GPU/VRAM if any – for a producer deciding which machine’s sessions to target. watch() writes this once per machine (skips it if the file’s already there, so it’s never overwritten with stale info by every session that starts up).
drop / pickup
One dead drop per message, in a shared directory. drop() uses Maildir’s write-temp-then-atomic-rename trick, so a concurrent pickup()/watch() never observes a half-written file.
def drop(dir:str, # directory to drop into (created if it doesn't exist) content:str, # the drop's body name:str=None, # filename to use; default is a sortable `{timestamp}-{uuid8}` so pickups see drops in creation order)->str:
Atomically write a dead drop into dir, for another process – possibly on another machine, via a synced/shared directory – to pickup() or watch() later. Writes to a temp file in the same directory first, then atomically renames it into place (Maildir’s trick), so a concurrent pickup/watch never sees a half-written drop. Returns the dropped file’s final path.
def pickup(dir:str, # directory to check for drops delete:bool=True, # remove the file after reading it, so each drop is picked up exactly once)->dict: # {'path':..., 'name':..., 'content':...}, or None if nothing's pending
Look for the oldest pending drop in dir (sorted by filename, so default drop() names – timestamp-prefixed – come out in creation order) and read it. Returns None if dir doesn’t exist or has no drops. Ignores hidden files – both its own in-progress temp files (.tmp-*) and stray dotfiles from other processes sharing the directory (e.g. an editor’s .foo.swp if something’s being edited in place there).
watch
Blocking loop over pickup(). Idles by polling os.stat().st_mtime (sub-second precision) instead of re-running pickup() on a fixed timer – a directory’s mtime changes when a file is added inside it, so this reacts to a new drop without necessarily waiting out a full poll_interval. The mtime baseline is captured before the emptiness check, not after, so a drop landing in between is still caught on the very next check rather than silently missed.
Session-scoped subdirectories: so multiple agents can watch() the same parent dir without clobbering each other’s drops, watch() doesn’t read dir directly – it creates (or reuses, if already present) a subdirectory of dir named after the current session id, and pickups happen from there. The session id is auto-detected (CLAUDE_CODE_SESSION_ID if set, else a random id generated once and cached for the life of the process) or can be passed explicitly via session_id=. drop()/pickup() themselves are unchanged – callers on the producing side need to know which session subdirectory to drop into (not yet wired up; watch() is the first half of this change).
def watch(dir:str, # parent directory for drops; a session-scoped subdirectory is created (or reused) under it -- see module docs callback:callable, # called as callback(drop) for each drop, where drop is the dict pickup() returns poll_interval:float=1.0, # seconds between mtime checks while dir is empty delete:bool=True, # remove each drop after callback returns (passed through to pickup) max_iters:int=None, # stop after this many drops-or-idle-checks (mainly for tests/scripted runs -- omit to run forever) session_id:str=None, # session id whose subdirectory to watch; default auto-detects (see _session_id))->int:
*Blocking loop: repeatedly pickup() drops from a session-scoped subdirectory of dir and pass each to callback; when that subdirectory is empty, blocks on its mtime change (see _await_change) instead of busy-polling pickup() itself. The subdirectory is named after session_id (or the auto-detected current session id if omitted), created if it doesn’t exist yet or reused as-is if it does – so multiple agents watching the same parent dir don’t see each other’s drops. Also writes MACHINE_INFO.md into dir itself the first time any session watches there (see _write_machine_info) – a no-op on later calls once it exists. Runs forever unless max_iters is given – stop it with Ctrl-C otherwise. Like slmn’s other blocking primitives, this is CLI/import-only, never exposed over MCP, since it never returns promptly. Returns the number of drops processed.*
list_sessions
The other half of session-scoped watch(): before a producer can drop() a prompt into a specific agent’s subdirectory, it needs to know which session subdirectories currently exist. list_sessions() just lists them.
def list_sessions(dir:str, # parent dead-drop directory (the same `dir` passed to watch()) require:tuple=None, # if given, only count a subdirectory as a session when it contains at least one of these subdirectories, e.g. ('inbox','outbox'))->list: # sorted list of session-id subdirectory names; [] if `dir` doesn't exist
List the session ids currently present as subdirectories of dir – i.e. sessions that have called watch(dir, …) here at least once – so a producer can pick one to drop() into. Sorted alphabetically. Ignores hidden/dotted entries and non-directories. require exists because ‘every subdirectory’ is too loose once a dead-drop root holds anything else: the single-file workflow keeps its own prompts/ and responses/ directories right there, and those would otherwise be offered as sessions you could address but never get an answer from. Passing require=(‘inbox’,‘outbox’) restricts the result to directories actually set up for the prompt/reply exchange.
Sessions: inbox and outbox
One machine can be running several agent sessions at once, each with its own context – a different project, a different notebook, a different train of thought. So a producer with a prompt to send needs to say which one it’s for, and get that session’s answer back and no one else’s. Each session gets a subdirectory of dir holding exactly two directories:
inbox/{prompt_id}/ – a prompt, dropped by the producer (drop_prompt)
outbox/{prompt_id}.md – the reply, streamed back by the responder (reply_stream)
Nothing is ever moved, renamed or deleted. Both halves stay where they were written, so the directories are their own permanent record of every exchange, and “what still needs answering” is a set difference computed on demand (pending_prompts = inbox minus outbox) rather than state maintained across directories – which can drift out of sync, or be left wrong by a crash at exactly the wrong instant.
The two sides are shaped differently, on purpose. A prompt is a bundle: a directory built in a hidden temp dir and renamed into place in one atomic step (drop_bundle, extending drop()’s Maildir trick from a file to a directory), because the producer has the whole thing at once and it may carry attachments alongside the text, and a watcher must never see it half-written. A reply is a plain file appended to as it’s composed (reply_stream), because it has to be readable while it’s still being written – someone is watching it arrive. An atomically-renamed bundle is invisible until it’s finished, which is exactly right for the prompt and useless for the reply.
That leaves the reply with no atomic “it’s complete now” moment, so completion is marked in-band instead: reply_stream appends DONE_MARKER when its block exits normally and deliberately not when it exits via an exception, so a reply cut short by a crash stays visibly unfinished rather than reading as a short complete answer. follow_file/follow_reply yield chunks as they land and stop at that marker.
Bundle directories are named by a prompt_id (see _prompt_id) – {timestamp}-{slug}-{rand}, e.g. 20260721T143205-hellow-a3f9c2 – so listings sort chronologically (the timestamp leads, same reasoning as drop()’s default filename) while staying human-skimmable (ls shows a snippet of the prompt) and unique. The reply reuses its prompt’s id as its filename, which is what lets the two sides match up with no parsing.
A responder just loops on await_prompt() and answers with reply_stream(). Waiting for prompts is itself what announces the session: await_prompt() creates the two directories on the way in, and until they exist list_sessions(require=('inbox','outbox')) can’t see the session and a producer has nowhere to address a prompt. So starting the watch is all it takes to become reachable – print the session id it’s serving and the other side can pick it by name.
def read_bundle( path:str, # a bundle directory, as written by drop_bundle)->dict: # {filename: content}, text decoded where possible, else raw bytes
Read every file in a bundle directory into {filename: content}, decoding as text where possible and falling back to raw bytes for anything that isn’t (e.g. an image attachment).
def drop_bundle(dir:str, # directory to drop the bundle into (created if it doesn't exist) files:dict, # {filename: content}, content str or bytes bundle_id:str=None, # bundle directory name; auto-generated (see _prompt_id) from the first str-valued file's content if omitted)->str: # path to the bundle directory
Atomically drop a multi-file bundle – e.g. a prompt’s text plus an image attachment – as a single directory under dir. Writes every file into a hidden temp directory first, then atomically renames the whole directory into place (extends drop()’s Maildir trick to a whole bundle), so a concurrent pickup/watch never sees a partially-written bundle. Returns the bundle’s final directory path.
def drop_prompt(dir:str, # parent dead-drop directory session_id:str, # which session's inbox to drop into files:dict, # {filename: content}, content str or bytes -- see drop_bundle)->str: # the generated prompt_id (the bundle's directory name)
Drop a multi-file prompt bundle into a session’s inbox/ (see drop_bundle). Returns the prompt_id, which names the reply file to watch for in outbox/ (see follow_reply). The prompt is a bundle – a directory renamed into place in one step – because the producer has the whole thing at once and it may carry attachments; the reply deliberately isn’t (see reply_stream).
next_prompt
The other reusable piece of the old prototype: a human types a prompt into a shared file (e.g. from a Jupyter cell on a different machine), possibly multi-line, and marks it submitted with a ---SEND----style line so a mid-typing autosave doesn’t false-trigger. cycle.sh implemented this with an awk block-extraction pass and a .sentcount sidecar file; next_prompt() is the same idea generalized (any path, any marker regex, no hardcoded filenames) and reimplemented in Python, reusing _await_change for the blocking wait.
The response side doesn’t need a new primitive – once a response is composed, drop() it to whatever file/directory the other end is watching.
def next_prompt( path:str, # path to a growing, marker-delimited prompt log (e.g. a file a human edits in Jupyter/an editor and appends a SEND line to when done typing) poll_interval:float=1.0, # seconds between mtime checks marker:str='^\\s*-+\\s*send\\s*-+\\s*$', # case-insensitive regex marking the end of a submitted block; a dash-padded SEND line by default)->str:
*Block until path gains a new complete marker-delimited block, then return that block’s text – the newest multi-line message the other side submitted. Waits for the file to actually change (see _await_change), then re-checks the marker count, so an unrelated resave or a still-typing autosave doesn’t false-trigger. Remembers how many markers it’s already consumed in a sidecar state file ({path}.dd_state), so repeated calls – e.g. one per turn of a conversation loop – never re-fire on old content.*
stream
Write a dead drop incrementally instead of all at once – e.g. as an LLM streams tokens. One file, plain appends, flushed immediately so a tail -f responses/{name} sees it live. Same name a plain drop() would use.
def follow_reply(dir:str, # parent dead-drop directory session_id:str, # the session that was prompted prompt_id:str, # the id returned by drop_prompt poll_interval:float=1.0, # seconds between checks name:str=None, # reply filename within outbox/; defaults to '{prompt_id}.md' (matches reply_stream)):
Follow the streamed reply to prompt_id as the responder writes it (see follow_file) – the producer-side counterpart to reply_stream(), and the last step of an exchange: drop_prompt() to send, follow_reply() to receive. Returns when the reply is terminated by DONE_MARKER; a caller that doesn’t care about seeing it arrive can simply join everything yielded.
def follow_file( path:str, # file to follow; it doesn't have to exist yet poll_interval:float=1.0, # seconds between checks):
Follow a file being streamed into by a writer (see stream/reply_stream), yielding each newly-appended stretch of text as it lands – tail -f as a generator. Returns once the content ends with DONE_MARKER, which is stripped and never yielded, so joining everything yielded gives exactly the reply. Waits for path to appear rather than requiring it upfront, since the reader normally starts before the responder has written anything. Polls instead of using inotify deliberately: the whole point of dead_drop is that the directory can be a synced/networked one, where filesystem change events don’t cross the boundary but mtime and size still do.
def read_prompt(dir:str, # parent dead-drop directory session_id:str, # session whose inbox/ holds the prompt prompt_id:str, # the bundle id, e.g. from pending_prompts)->dict: # {filename: content} -- see read_bundle
Read a prompt bundle out of a session’s inbox/ without* claiming, moving or deleting it – the bundle counterpart to pickup(), minus the consuming. A prompt is answered by writing to outbox/, never by being taken; leaving it in place is what lets pending_prompts() work as a plain set difference, and keeps the whole exchange readable after the fact.*
def pending_prompts(dir:str, # parent dead-drop directory session_id:str=None, # session to check; default auto-detects (see _session_id))->list: # sorted prompt ids in inbox/ with no reply in outbox/
The prompt ids sitting in a session’s inbox/ that have no reply in its outbox/ yet – i.e. the work still to do. This is the two-directory protocol’s entire bookkeeping: nothing is ever moved, renamed or deleted, so ‘what’s outstanding’ is a set difference computed on demand instead of state that has to be maintained correctly across directories (and can therefore drift out of sync, or be left wrong by a crash at the wrong instant). Both halves of an exchange stay where they were written, which also makes the directories their own permanent record. Note this reports unanswered, not unfinished: a reply file counts the moment it’s created, even mid-stream, so use this to find work to pick up – to judge whether a reply actually completed, check for DONE_MARKER (see follow_file).
def await_prompt(dir:str, # parent dead-drop directory session_id:str=None, # session to serve; default auto-detects (see _session_id) poll_interval:float=1.0, # seconds between checks while the inbox is quiet)->dict: # {'id':..., 'files':...} -- the oldest unanswered prompt
Block until this session has an unanswered prompt, then return the oldest one – the responder’s entry point, answering a producer’s drop_prompt(). Returns straight away if one is already waiting. Creates the session’s inbox/ and outbox/ first (and MACHINE_INFO.md alongside, if absent – same as watch()): waiting for prompts is exactly what announces a session, since until those two directories exist list_sessions(require=(‘inbox’,‘outbox’)) can’t see it and a producer has nowhere to address a prompt. The prompt itself stays in inbox/ (see read_prompt); what stops it counting as outstanding is a reply file appearing in outbox/, which reply_stream() creates the moment it starts writing – so an answer already in progress won’t be handed out twice. Idles on the inbox’s mtime rather than re-listing on a timer, with the baseline captured before* the check, so a prompt landing in that gap is caught on the next pass instead of waiting out a full interval (the same race watch() closes).*
def reply_stream(dir:str, # parent dead-drop directory session_id:str, # the session being answered -- its outbox/ is written to prompt_id:str, # the id of the prompt being answered (as returned by drop_prompt / seen on an inbox bundle) name:str=None, # reply filename within outbox/; defaults to '{prompt_id}.md'):
The responder’s half of a session exchange: stream a reply into session_id’s outbox/, answering prompt_id. The counterpart to drop_prompt() on the producer side. A reply is a plain appended-to file rather than an atomically-renamed bundle because* it has to be readable while it’s still being written – a bundle only becomes visible once complete, which is right for a prompt (written all at once) and useless for a reply someone is watching arrive. DONE_MARKER is appended when the with block exits normally, and deliberately NOT when it exits via an exception: a reply file that stops growing without its terminator is therefore distinguishable from a finished one, which is the whole reason completion is marked in-band instead of inferred from the file merely existing.*
def stream(dir:str, # directory to drop into (created if it doesn't exist) name:str, # filename to write -- same as a plain drop(name=...)):
Context manager for writing a dead drop incrementally. Inside the with block, call the yielded StreamWriter’s .write(chunk) for each piece of text as it’s produced; each call appends straight to dir/name and flushes, so a concurrent tail -f sees it live.
Handle yielded by stream() – call .write(chunk) for each piece of text as it’s produced.
CLI responder tools
The legacy prompts/-and-responses/ exchange (the one boopiter’s deaddrop/claude model speaks) needs a responder on the other side – typically an agent like a Claude Code session. These two wrap that role as CLI-friendly one-shot commands: watch_prompts blocks until a prompt arrives and prints it (run it as a background command; its exit is the wake-up), respond_prompt writes the paired reply. Being in TOOLS means an agent needs only the one slmn shell permission for the whole loop, instead of ad hoc python -c snippets that trip permission prompts on every rearm.
def respond_prompt( name:str, # the prompt's filename, as reported by watch_prompts text:str=None, # reply body; omit to read it from stdin (spares shell-quoting a long reply)dir:str=None, # dead-drop root (default ~/dead_drop); writes into its responses/ subdirectory model:str=None, # if given, appended as a '---MODEL: <model>---' line before the DONE marker (the self-identification line boopiter's _DEADDROP_MODEL_RE parses into the reply details))->str:
Write the reply to a legacy-route prompt: dir/responses/name, terminated with DONE_MARKER so a follow_file() reader knows it’s complete. Written via drop() – one atomic rename – so the reader sees the whole reply at once; that trades away live streaming (use stream() by hand if you want that), which for a one-shot CLI responder is the right default: no reader ever samples a half-written reply.
def watch_prompts(dir:str=None, # dead-drop root (default ~/dead_drop); watches its prompts/ subdirectory poll_interval:float=0.25, # seconds between pickup checks)->str: # one JSON line: {"name":..., "content":...}
*Block until a prompt file lands in dir/prompts, then consume it and return it as one JSON line – the legacy-route responder’s entry point, pairing with a producer that drops into prompts/ and follows responses/ (e.g. boopiter’s _deaddrop_stream_legacy). Blocking like watch(), so CLI/import-only, never MCP: run it as a background command and treat its exit as the wake-up signal. The prompt file is deleted on pickup – its name is all a responder needs to address the reply (see respond_prompt).*
Smoke test
import tempfile, threading, ostmp_dir = tempfile.mkdtemp()# drop / pickupp1 = drop(tmp_dir, "first message")p2 = drop(tmp_dir, "second message")print(p1, p2)d = pickup(tmp_dir)assert d['content'] =="first message"d = pickup(tmp_dir)assert d['content'] =="second message"assert pickup(tmp_dir) isNoneprint("drop/pickup OK")# watch (drop from a background thread after a short delay -- this exact scenario# used to hang forever under the old bash-mtime-poll implementation: see module docstring).# watch() now reads from a session-scoped subdirectory of tmp_dir, so the delayed# drop targets that same subdirectory explicitly rather than tmp_dir itself.received = []watch_session_dir = os.path.join(tmp_dir, "test-session")def _delayed_drop(): time.sleep(0.3) drop(watch_session_dir, "watched message")threading.Thread(target=_delayed_drop).start()n = watch(tmp_dir, received.append, poll_interval=0.1, max_iters=2, session_id="test-session")print(n, received)assert n ==1and received[0]['content'] =="watched message"print("watch OK")# list_sessionsassert list_sessions(tmp_dir) == ["test-session"]assert list_sessions(os.path.join(tmp_dir, "no-such-dir")) == []print("list_sessions OK")# MACHINE_INFO.mdinfo_path = os.path.join(tmp_dir, "MACHINE_INFO.md")assert os.path.exists(info_path)first_contents =open(info_path).read()assert"GPU"in first_contents and"VRAM"in first_contents and"RAM"in first_contents and"OS"in first_contentswatch(tmp_dir, lambda d: None, poll_interval=0.1, max_iters=0, session_id="another-session")assertopen(info_path).read() == first_contents # second watch() call must not overwrite itprint("MACHINE_INFO.md OK")# sessions: a prompt bundle into inbox/, a streamed reply back out of outbox/, nothing movedbundle_root = tempfile.mkdtemp()SESS ="bundle-session"# a session is invisible until something waits on it -- await_prompt creates inbox/ + outbox/assert list_sessions(bundle_root, require=('inbox','outbox')) == []pid = drop_prompt(bundle_root, SESS, {"prompt.txt": "Hello, can you check the CI status?", "note.txt": "attachment"})assert pid.split("-")[1] =="helloc"assert pending_prompts(bundle_root, SESS) == [pid]print("drop_prompt/pending_prompts OK", pid)got = await_prompt(bundle_root, SESS, poll_interval=0.1) # already waiting -- returns at onceassert got["id"] == pidassert got["files"] == {"prompt.txt": "Hello, can you check the CI status?", "note.txt": "attachment"}assert list_sessions(bundle_root, require=('inbox','outbox')) == [SESS] # now discoverableprint("await_prompt OK")# the reply streams in on a background thread while the producer follows it livedef _delayed_reply(): time.sleep(0.2)with reply_stream(bundle_root, SESS, pid) as w:for part in ("CI is ", "green"): w.write(part) time.sleep(0.2)threading.Thread(target=_delayed_reply).start()chunks =list(follow_reply(bundle_root, SESS, pid, poll_interval=0.05))assert"".join(chunks) =="CI is green", chunksassertlen(chunks) >1, f"reply was not read incrementally: {chunks}"print("reply_stream/follow_reply OK", chunks)# answered, and the prompt survives -- nothing was moved, renamed or deletedassert pending_prompts(bundle_root, SESS) == []assert (Path(bundle_root)/SESS/'inbox'/pid/'prompt.txt').exists()assert (Path(bundle_root)/SESS/'outbox'/f'{pid}.md').exists()print("nothing moved OK")# an unterminated reply must not read as a finished onehalf = Path(bundle_root)/SESS/'outbox'/'unfinished.md'half.write_text("half a thought")g = follow_file(str(half), poll_interval=0.05)assertnext(g) =="half a thought"ended = []threading.Thread(target=lambda: ended.append(next(g, "RETURNED")), daemon=True).start()time.sleep(0.4)assertnot ended, "follow_file ended without DONE_MARKER -- a truncated reply would look complete"print("DONE_MARKER discipline OK")# next_promptprompt_path = os.path.join(tmp_dir, "prompts.txt")open(prompt_path, "w").close()def _delayed_prompt(): time.sleep(0.3)withopen(prompt_path, "a") as f: f.write("line one\nline two\n---SEND---\n")threading.Thread(target=_delayed_prompt).start()prompt = next_prompt(prompt_path, poll_interval=0.1)print(repr(prompt))assert prompt =="line one\nline two"print("next_prompt OK")
/tmp/tmpdr5x2a9x/1784656078518161303-d2180ae7 /tmp/tmpdr5x2a9x/1784656078518296573-0721eab2
drop/pickup OK
1 [{'path': '/tmp/tmpdr5x2a9x/test-session/1784656078819363742-39db4f30', 'name': '1784656078819363742-39db4f30', 'content': 'watched message'}]
watch OK
list_sessions OK
MACHINE_INFO.md OK
drop_prompt OK 20260721T124758-helloc-747838
watch_session OK
reply_bundle/await_reply/pending_replies OK
complete_reply/unprocessed_replies OK
'line one\nline two'
next_prompt OK