dead_drop

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).

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.

---

source

drop

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.

---

source

pickup

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.

---

source

watch

def watch(
    dir:str, # directory to watch for drops
    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)
)->int:

*Blocking loop: repeatedly pickup() drops from dir and pass each to callback; when dir is empty, blocks on a directory-mtime change (see _await_change) instead of busy-polling pickup() itself. 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.*

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.

---

source

next_prompt

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.

---

source

stream

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.

---

source

StreamWriter

def StreamWriter(
    fh
):

Handle yielded by stream() – call .write(chunk) for each piece of text as it’s produced.

Smoke test

import tempfile, threading, os

tmp_dir = tempfile.mkdtemp()

# drop / pickup
p1 = 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) is None
print("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)
received = []
def _delayed_drop():
    time.sleep(0.3)
    drop(tmp_dir, "watched message")
threading.Thread(target=_delayed_drop).start()
n = watch(tmp_dir, received.append, poll_interval=0.1, max_iters=2)
print(n, received)
assert n == 1 and received[0]['content'] == "watched message"
print("watch OK")

# next_prompt
prompt_path = os.path.join(tmp_dir, "prompts.txt")
open(prompt_path, "w").close()

def _delayed_prompt():
    time.sleep(0.3)
    with open(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")