# dead_drop


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## drop / pickup

One dead drop per message, in a shared directory.
[`drop()`](https://drscotthawley.github.io/slmn/dead_drop.html#drop)
uses Maildir’s write-temp-then-atomic-rename trick, so a concurrent
[`pickup()`](https://drscotthawley.github.io/slmn/dead_drop.html#pickup)/[`watch()`](https://drscotthawley.github.io/slmn/dead_drop.html#watch)
never observes a half-written file.

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L16"
target="_blank" style="float:right; font-size:smaller">source</a>

### drop

``` python
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()`](https://drscotthawley.github.io/slmn/dead_drop.html#pickup)
or
[`watch()`](https://drscotthawley.github.io/slmn/dead_drop.html#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.*

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L31"
target="_blank" style="float:right; font-size:smaller">source</a>

### pickup

``` python
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()`](https://drscotthawley.github.io/slmn/dead_drop.html#pickup).
Idles by polling `os.stat().st_mtime` (sub-second precision) instead of
re-running
[`pickup()`](https://drscotthawley.github.io/slmn/dead_drop.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L58"
target="_blank" style="float:right; font-size:smaller">source</a>

### watch

``` python
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()`](https://drscotthawley.github.io/slmn/dead_drop.html#next_prompt)
is the same idea generalized (any path, any marker regex, no hardcoded
filenames) and reimplemented in Python, reusing
[`_await_change`](https://drscotthawley.github.io/slmn/dead_drop.html#_await_change)
for the blocking wait.

The response side doesn’t need a new primitive – once a response is
composed,
[`drop()`](https://drscotthawley.github.io/slmn/dead_drop.html#drop) it
to whatever file/directory the other end is watching.

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L91"
target="_blank" style="float:right; font-size:smaller">source</a>

### next_prompt

``` python
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.

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L122"
target="_blank" style="float:right; font-size:smaller">source</a>

### stream

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/drscotthawley/slmn/blob/main/slmn/dead_drop.py#L113"
target="_blank" style="float:right; font-size:smaller">source</a>

### StreamWriter

``` python
def StreamWriter(
    fh
):
```

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

## Smoke test

``` python
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")
```