From 893e12345c2458e70b333faedaaaaf8fbffb6932 Mon Sep 17 00:00:00 2001
From: Gabriel Schneider <gbrls@0x4200.cafe>
Date: Tue, 29 Sep 2026 17:22:24 -0300
Subject: A 9P monkey drives random requests at a throwaway session and checks
 the documented rules

test/monkey9p.py starts pardes --detach with its own HOME, XDG dirs and
runtime dir, speaks raw 9P (test/ninep.py) with a timeout on every reply
and Tflush for held reads, and after each seeded operation checks that
pardes is alive, the one-failure rule over a marker-delimited log window
(with and without a follower), /index against /pane, and chunked against
whole reads. Every run writes a replayable ops log; a failure is saved
with its seed and step, shrunk by delta debugging over replays, and the
run goes on in a fresh session. Teardown makes the fuzzer a subreaper and
kills every process carrying the session's token.

zig build monkey-9p -- --seed N --steps M | --replay F | --shrink F;
a 150-step fixed-seed smoke run is part of fs-test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/monkey9p.py | 1958 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 1958 insertions(+)
 create mode 100644 test/monkey9p.py

(limited to 'test/monkey9p.py')

diff --git a/test/monkey9p.py b/test/monkey9p.py
new file mode 100644
index 00000000..4c1732e5
--- /dev/null
+++ b/test/monkey9p.py
@@ -0,0 +1,1958 @@
+#!/usr/bin/env python3
+"""pardes 9P monkey: random 9P operations against a throwaway detached session.
+
+  zig build monkey-9p -Dplatform=tty --prefix <scratch> -- [--seed N] [--steps M]
+  zig build monkey-9p -Dplatform=tty --prefix <scratch> -- --replay FILE
+  zig build monkey-9p -Dplatform=tty --prefix <scratch> -- --shrink FILE
+  python3 test/monkey9p.py <pardes> [same flags]
+
+A run starts `pardes --detach` with its own HOME, XDG dirs and a short
+runtime dir in /tmp, speaks raw 9P to it (test/ninep.py), and for each
+step asks a GENERATOR for one concrete operation, runs it, then runs every
+INVARIANT. Everything random comes from --seed; every operation it ran is
+written to <out>/runs/seed-N/ops.jsonl, concrete but for pane and column
+references ({p3}: the 4th serial of /index, mod its length, when the op
+runs, so a shrunk sequence still names live panes), and `--replay` runs
+the same requests again without the generator. A failure writes a bug record
+(seed, step, the replayable operations since that session began) to
+<out>/bugs/, shrinks it by delta debugging over replays unless
+--no-shrink, and the run carries on in a fresh session.
+
+Plug-in points: a generator is a function `(ctx, rng) -> op dict` under
+@generator(weight); an op kind is a function `(sess, op) -> Result` under
+@op_kind(name); an invariant is a function `(ctx, res) -> str | None`
+under @invariant, returning why it failed.
+
+Safety: the session never sees PARDES_*, NINE_MOUNT or NAMESPACE; every
+payload that could reach a shell is drawn from an alphabet with no path
+or shell metacharacters (shell_safe), Edit text from a grammar without
+sam's w/e/r/f/b/B/D/n/!/</>/| commands, and nothing is ever written,
+created or removed under /os or /src (the host filesystem). This process
+makes itself a child subreaper, so every shell the session starts is its
+descendant even after pardes dies, and teardown kills them all.
+"""
+import argparse
+import collections
+import ctypes
+import hashlib
+import json
+import os
+import random
+import re
+import select
+import shutil
+import signal
+import socket
+import struct
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from ninep import Client, string  # noqa: E402
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DEFAULT_OUT = os.path.join(os.path.dirname(REPO), '.scratch', 'monkey9p')
+
+# 9P message types.
+TWALK, TOPEN, TCREATE, TREAD, TWRITE, TCLUNK, TREMOVE, TSTAT, TFLUSH = 110, 112, 114, 116, 118, 120, 122, 124, 108
+RERROR = 107
+OREAD, OWRITE, ORDWR, OTRUNC = 0, 1, 2, 16
+DMDIR = 0x80000000
+
+TIMEOUT = 5.0         # every reply, unless the op is a documented slow one
+SLOW_TIMEOUT = 30.0   # Grep, Find, Dump, Restore, Tty, huge writes
+HELD_WAIT = 0.1       # how long a read of a held file is let wait before Tflush
+MSIZE = 65536 + 24
+
+
+class Hang(Exception):
+    """A 9P reply did not come within its timeout."""
+
+
+class Dropped(Exception):
+    """The server closed the connection (or reset it)."""
+
+
+class Ended(Exception):
+    """pardes quit cleanly: exit 0, no crash file. Closing the session's last
+    pane quits it (fs.md: 'Closing the session's last pane quits pardes'),
+    and a terminal whose shell exits closes its pane, so this can follow any
+    step; the run starts a fresh session and carries on."""
+
+
+# --------------------------------------------------------------------------
+# Wire: raw 9P with per-request timeouts and Tflush for held reads.
+
+class Wire(Client):
+    def __init__(self, address, msize=MSIZE):
+        super().__init__(address, timeout=TIMEOUT, msize=msize)
+        self.next_fid = 100
+
+    def newfid(self):
+        self.next_fid += 1
+        return self.next_fid
+
+    def _send(self, kind, payload):
+        self.tag = (self.tag + 1) % 65535
+        packet = struct.pack('<IBH', 7 + len(payload), kind, self.tag) + payload
+        if len(packet) > self.msize:
+            raise ValueError(f'harness bug: a {len(packet)}-byte request exceeds msize {self.msize}')
+        try:
+            self.socket.sendall(packet)
+        except socket.timeout:
+            raise Hang('send did not drain within the timeout')
+        except OSError as why:
+            raise Dropped(f'send: {why}')
+        return self.tag
+
+    def _recv(self, timeout):
+        self.socket.settimeout(timeout)
+        try:
+            size, kind, tag = struct.unpack('<IBH', self.receive(7))
+            body = self.receive(size - 7)
+        except socket.timeout:
+            raise Hang(f'no reply within {timeout:g}s')
+        except (EOFError, ConnectionResetError, BrokenPipeError) as why:
+            raise Dropped(str(why))
+        except OSError as why:
+            if isinstance(why, socket.timeout):
+                raise Hang(f'no reply within {timeout:g}s')
+            raise Dropped(str(why))
+        return kind, tag, body
+
+    def call(self, kind, payload, timeout=TIMEOUT):
+        """(ok, body_or_error_text)."""
+        tag = self._send(kind, payload)
+        while True:
+            rkind, rtag, body = self._recv(timeout)
+            if rtag != tag:
+                continue  # a late reply to a flushed request
+            if rkind == RERROR:
+                n = struct.unpack_from('<H', body)[0]
+                return False, body[2:2 + n].decode(errors='replace')
+            if rkind != kind + 1:
+                raise Dropped(f'expected reply {kind + 1}, received {rkind}')
+            return True, body
+
+    def call_held(self, kind, payload, wait=HELD_WAIT):
+        """A request the server may hold: after `wait`, flush it. Returns
+        (ok, body) for an answer, or (None, 'flushed')."""
+        tag = self._send(kind, payload)
+        self.socket.settimeout(None)
+        ready, _, _ = select.select([self.socket], [], [], wait)
+        if ready:
+            rkind, rtag, body = self._recv(TIMEOUT)
+            if rtag == tag:
+                if rkind == RERROR:
+                    n = struct.unpack_from('<H', body)[0]
+                    return False, body[2:2 + n].decode(errors='replace')
+                return True, body
+        ftag = self._send(TFLUSH, struct.pack('<H', tag))
+        answer = (None, 'flushed')
+        while True:
+            rkind, rtag, body = self._recv(TIMEOUT)
+            if rtag == tag:
+                if rkind == RERROR:
+                    n = struct.unpack_from('<H', body)[0]
+                    answer = (False, body[2:2 + n].decode(errors='replace'))
+                else:
+                    answer = (True, body)
+            elif rtag == ftag:
+                return answer
+
+    # Convenience over call().
+    def walk_names(self, names, timeout=TIMEOUT):
+        """(fid or None, error)."""
+        fid = self.newfid()
+        old = 1
+        if not names:
+            ok, body = self.call(TWALK, struct.pack('<IIH', old, fid, 0), timeout)
+            return (fid, None) if ok else (None, body)
+        for start in range(0, len(names), 16):
+            part = names[start:start + 16]
+            ok, body = self.call(TWALK, struct.pack('<IIH', old, fid, len(part)) +
+                                 b''.join(string(n) for n in part), timeout)
+            if not ok:
+                if old != 1:
+                    self.call(TCLUNK, struct.pack('<I', fid))
+                return None, body
+            if struct.unpack_from('<H', body)[0] != len(part):
+                if old != 1:
+                    self.call(TCLUNK, struct.pack('<I', fid))
+                return None, 'walk stopped short'
+            old = fid
+        return fid, None
+
+    def clunk(self, fid):
+        return self.call(TCLUNK, struct.pack('<I', fid))
+
+    def read_all(self, fid, count=None, limit=1 << 22):
+        count = min(count or (self.msize - 11), self.msize - 11)
+        data = bytearray()
+        while len(data) < limit:
+            ok, body = self.call(TREAD, struct.pack('<IQI', fid, len(data), count))
+            if not ok:
+                raise OSError(body)
+            chunk = body[4:]
+            if not chunk:
+                break
+            data.extend(chunk)
+        return bytes(data)
+
+    def read_path(self, path):
+        fid, err = self.walk_names(split(path))
+        if fid is None:
+            raise FileNotFoundError(f'{path}: {err}')
+        try:
+            ok, body = self.call(TOPEN, struct.pack('<IB', fid, OREAD))
+            if not ok:
+                raise OSError(f'{path}: {body}')
+            return self.read_all(fid)
+        finally:
+            self.clunk(fid)
+
+    def write_path(self, path, data):
+        """(ok, error) for one open, one write, one clunk."""
+        fid, err = self.walk_names(split(path))
+        if fid is None:
+            return False, err
+        try:
+            ok, body = self.call(TOPEN, struct.pack('<IB', fid, OWRITE))
+            if not ok:
+                return False, body
+            ok, body = self.call(TWRITE, struct.pack('<IQI', fid, 0, len(data)) + data)
+            return (True, None) if ok else (False, body)
+        finally:
+            self.clunk(fid)
+
+
+def split(path):
+    return [p.encode('latin-1') if isinstance(p, str) else p for p in path.split('/') if p]
+
+
+def dir_names(data):
+    names, offset = [], 0
+    while offset + 2 <= len(data):
+        size = struct.unpack_from('<H', data, offset)[0]
+        n = struct.unpack_from('<H', data, offset + 41)[0]
+        names.append(data[offset + 43:offset + 43 + n].decode('latin-1'))
+        offset += size + 2
+    return names
+
+
+# --------------------------------------------------------------------------
+# Session lifecycle.
+
+def _libc():
+    try:
+        return ctypes.CDLL(None, use_errno=True)
+    except OSError:
+        return None
+
+
+def become_subreaper():
+    """Orphaned shells re-parent to us, not init, so teardown finds them."""
+    libc = _libc()
+    if libc is not None:
+        libc.prctl(36, 1, 0, 0, 0)  # PR_SET_CHILD_SUBREAPER
+
+
+def descendants(root):
+    parent = {}
+    for entry in os.listdir('/proc'):
+        if not entry.isdigit():
+            continue
+        try:
+            with open(f'/proc/{entry}/stat', 'rb') as f:
+                stat = f.read()
+            ppid = int(stat[stat.rindex(b')') + 2:].split()[1])
+        except (OSError, ValueError):
+            continue
+        parent.setdefault(ppid, []).append(int(entry))
+    out, stack = [], [root]
+    while stack:
+        for child in parent.get(stack.pop(), []):
+            out.append(child)
+            stack.append(child)
+    return out
+
+
+def tokened(token):
+    """Our own processes whose environment carries this session's token."""
+    found, needle = [], f'M9P_TOKEN={token}'.encode()
+    for entry in os.listdir('/proc'):
+        if not entry.isdigit():
+            continue
+        try:
+            with open(f'/proc/{entry}/environ', 'rb') as f:
+                if needle in f.read().split(b'\0'):
+                    found.append(int(entry))
+        except OSError:
+            continue
+    return found
+
+
+def sweep(prefix):
+    """Kill every process whose M9P_TOKEN starts with `prefix`, until none."""
+    needle = f'M9P_TOKEN={prefix}'.encode()
+    for _ in range(100):
+        found = []
+        for entry in os.listdir('/proc'):
+            if not entry.isdigit() or int(entry) == os.getpid():
+                continue
+            try:
+                with open(f'/proc/{entry}/environ', 'rb') as f:
+                    if any(v.startswith(needle) for v in f.read().split(b'\0')):
+                        found.append(int(entry))
+            except OSError:
+                continue
+        if not found:
+            return
+        for pid in found:
+            try:
+                os.kill(pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+        reap_zombies()
+        time.sleep(0.02)
+
+
+def reap_zombies():
+    while True:
+        try:
+            pid, _ = os.waitpid(-1, os.WNOHANG)
+        except ChildProcessError:
+            return
+        if pid == 0:
+            return
+
+
+RUNTIME_DIRS = []  # removed at exit, whatever ended the run
+
+
+class Session:
+    """One throwaway `pardes --detach`, its dirs, its connection(s)."""
+
+    counter = 0
+
+    def __init__(self, binary, run_dir, follower):
+        Session.counter += 1
+        self.binary = binary
+        self.follower_mode = follower
+        self.token = f'{os.getpid()}-{Session.counter}-{random.SystemRandom().randrange(1 << 30)}'
+        self.dir = os.path.join(run_dir, f'sess{Session.counter}')
+        self.home = os.path.join(self.dir, 'home')
+        self.runtime = tempfile.mkdtemp(prefix='m9p.', dir='/tmp')
+        RUNTIME_DIRS.append(self.runtime)
+        self.proc = None
+        self.wire = None
+        self.follower = None
+        self.handles = {}
+        self.dump_written = False
+        self.marker = None
+        self.window_base = 0
+        self.start()
+
+    def env(self):
+        env = {k: v for k, v in os.environ.items()
+               if not k.startswith('PARDES_') and k not in ('NINE_MOUNT', 'NAMESPACE')}
+        env.update(HOME=self.home, XDG_RUNTIME_DIR=self.runtime,
+                   XDG_CONFIG_HOME=os.path.join(self.home, 'config'),
+                   XDG_STATE_HOME=os.path.join(self.home, 'state'),
+                   XDG_DATA_HOME=os.path.join(self.home, 'data'),
+                   XDG_CACHE_HOME=os.path.join(self.home, 'cache'),
+                   SHELL='/bin/sh', PARDES_NOTIME='1', M9P_TOKEN=self.token,
+                   PARDES_DUMP=os.path.join(self.home, 'm9p.dump.zon'))
+        for language in ['RS', 'C', 'GO', 'TS', 'PY', 'ZIG']:
+            env['PARDES_LSP_' + language] = ''
+        return env
+
+    def start(self):
+        os.makedirs(self.home, exist_ok=True)
+        for name, text in FIXTURES.items():
+            with open(os.path.join(self.home, name), 'wb') as f:
+                f.write(text)
+        self.stderr = open(os.path.join(self.dir, 'stderr'), 'wb')
+        self.proc = subprocess.Popen([self.binary, '--detach=m9p', 'alpha.txt'], cwd=self.home,
+                                     env=self.env(), stdin=subprocess.DEVNULL,
+                                     stdout=subprocess.DEVNULL, stderr=self.stderr,
+                                     start_new_session=True)
+        self.address = os.path.join(self.runtime, 'pardes-9p-m9p.sock')
+        self.connect(first=True)
+        # Two more panes, so one close does not end the session.
+        for name in (b'utf8.txt\n', b'crlf.txt\n'):
+            self.wire.write_path('/look', name)
+
+    def connect(self, first=False):
+        deadline = time.monotonic() + (15 if first else 10)
+        last = None
+        while time.monotonic() < deadline:
+            if self.proc.poll() is not None:
+                raise Dropped(f'pardes exited {self.proc.returncode} before its socket answered')
+            if os.path.exists(self.address):
+                try:
+                    self.wire = Wire(self.address)
+                    break
+                except OSError as why:
+                    last = why
+            time.sleep(0.01)
+        else:
+            raise Hang(f'9P socket did not answer: {last}')
+        self.handles = {}
+        self.marker = None
+        if self.follower_mode:
+            self.follower = Follower(self.address)
+
+    def reconnect(self):
+        self.close_wires()
+        # A Restore hangs up and serves again: let the old socket go.
+        time.sleep(0.05)
+        self.connect()
+
+    def close_wires(self):
+        if self.follower is not None:
+            self.follower.stop()
+            self.follower = None
+        if self.wire is not None:
+            try:
+                self.wire.socket.close()
+            except OSError:
+                pass
+            self.wire = None
+
+    def alive(self):
+        return self.proc is not None and self.proc.poll() is None
+
+    def crash_text(self):
+        """The crash file's text (src/crash.zig) and stderr's tail."""
+        out = []
+        for base, _, files in os.walk(self.home):
+            if 'crashes' in files:
+                with open(os.path.join(base, 'crashes'), 'rb') as f:
+                    out.append(f.read().decode(errors='replace'))
+        try:
+            self.stderr.flush()
+            with open(os.path.join(self.dir, 'stderr'), 'rb') as f:
+                text = f.read()
+                # The panic line leads; the frames under it can run long.
+                tail = (text if len(text) <= 6000 else text[:3000] + b'\n...\n' + text[-2000:]).decode(errors='replace')
+            if tail.strip():
+                out.append('stderr: ' + tail)
+        except OSError:
+            pass
+        return '\n'.join(out)
+
+    def has_crash_file(self):
+        for base, _, files in os.walk(os.path.join(self.home, 'config')):
+            if 'crashes' in files:
+                return True
+        return False
+
+    def close(self, keep=False):
+        """Kill pardes and everything it started; verify nothing survives."""
+        self.close_wires()
+        if self.proc is not None and self.proc.poll() is None:
+            self.proc.terminate()
+            try:
+                self.proc.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+                self.proc.wait()
+        survivors = []
+        for _ in range(100):
+            reap_zombies()
+            pids = set(tokened(self.token)) | set(descendants(os.getpid()))
+            pids.discard(os.getpid())
+            if not pids:
+                break
+            for pid in pids:
+                try:
+                    os.kill(pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+            survivors = pids
+            time.sleep(0.02)
+        else:
+            raise RuntimeError(f'processes survived teardown: {sorted(survivors)}')
+        reap_zombies()
+        self.stderr.close()
+        shutil.rmtree(self.runtime, ignore_errors=True)
+        if not keep:
+            shutil.rmtree(self.dir, ignore_errors=True)
+
+
+class Follower:
+    """A second connection holding `follow new` on /log, reading records."""
+
+    def __init__(self, address):
+        self.lines = []
+        self.cond = threading.Condition()
+        self.done = False
+        self.wire = Wire(address)
+        fid, err = self.wire.walk_names([b'log'])
+        ok, body = self.wire.call(TOPEN, struct.pack('<IB', fid, ORDWR))
+        ok, body = self.wire.call(TWRITE, struct.pack('<IQI', fid, 0, 11) + b'follow new\n')
+        if not ok:
+            raise OSError(f'follow new: {body}')
+        self.fid = fid
+        self.thread = threading.Thread(target=self.run, daemon=True)
+        self.thread.start()
+
+    def run(self):
+        offset = 0
+        try:
+            while not self.done:
+                tag = self.wire._send(TREAD, struct.pack('<IQI', self.fid, offset, 8192))
+                while True:
+                    self.wire.socket.settimeout(None)
+                    kind, rtag, body = self.wire._recv(None)
+                    if rtag == tag:
+                        break
+                if kind == RERROR:
+                    break
+                chunk = body[4:]
+                if not chunk:
+                    break
+                offset += len(chunk)
+                with self.cond:
+                    self.lines.extend(chunk.decode(errors='replace').splitlines())
+                    self.cond.notify_all()
+        except (Hang, Dropped, OSError, struct.error):
+            pass
+        with self.cond:
+            self.done = True
+            self.cond.notify_all()
+
+    def wait_for(self, needle, timeout=TIMEOUT):
+        deadline = time.monotonic() + timeout
+        with self.cond:
+            while True:
+                for i in range(len(self.lines) - 1, -1, -1):
+                    if needle in self.lines[i]:
+                        return i
+                left = deadline - time.monotonic()
+                if left <= 0 or self.done:
+                    return None
+                self.cond.wait(left)
+
+    def stop(self):
+        self.done = True
+        try:
+            self.wire.socket.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+        self.wire.socket.close()
+        self.thread.join(timeout=2)
+
+
+FIXTURES = {
+    'alpha.txt': b'alpha beta gamma\nsecond line\n\nthird after a blank\n',
+    'utf8.txt': 'naïve café 日本語 é \U0001f600\nline two\n'.encode(),
+    'crlf.txt': b'one\r\ntwo\r\nthree\r\n',
+    'bad.txt': b'hello \xff\xfe world\nlatin-1 caf\xe9\ntruncated \xe6\x97 here\n',
+    'empty.txt': b'',
+    'noeol.txt': b'no newline at the end',
+}
+
+
+# --------------------------------------------------------------------------
+# Results, registries.
+
+class Result:
+    def __init__(self):
+        self.requests = []     # (kind, path, ok, why)
+        self.hung_up = False   # the op expected the server to hang up (Restore)
+        self.note = ''
+
+    def add(self, kind, path, ok, why=None):
+        self.requests.append((kind, path, ok, why))
+
+    def writes(self):
+        return [r for r in self.requests if r[0] in ('write', 'clunk-write', 'open-new', 'trunc')]
+
+
+GENERATORS = []   # (weight, fn)
+OPS = {}          # name -> fn(sess, op) -> Result
+INVARIANTS = []   # fn(ctx, res) -> str | None
+
+
+def generator(weight):
+    def wrap(fn):
+        GENERATORS.append((weight, fn))
+        return fn
+    return wrap
+
+
+def op_kind(name):
+    def wrap(fn):
+        OPS[name] = fn
+        return fn
+    return wrap
+
+
+def invariant(fn):
+    INVARIANTS.append(fn)
+    return fn
+
+
+def host_path(path):
+    """Guard: nothing mutating ever touches the served host tree."""
+    parts = split(path)
+    return bool(parts) and parts[0] in (b'os', b'src')
+
+
+def is_held(path):
+    return (path.rstrip('/') == '/log' or path.endswith('/event') or
+            path.endswith('/pty/data') or path.endswith('/pty/run'))
+
+
+def slow(data):
+    return len(data) > 65536 or any(w in data for w in (b'Grep', b'Find', b'Dump', b'Restore', b'Tty', b'Edit'))
+
+
+# --------------------------------------------------------------------------
+# Op kinds: each takes a concrete op (what the ops log holds) and runs it.
+
+def dec(s):
+    return s.encode('latin-1')
+
+
+def enc(b):
+    return b.decode('latin-1')
+
+
+@op_kind('write')
+def op_write(sess, op):
+    """open(path, mode); write data in the given chunk sizes; clunk."""
+    res = Result()
+    path, data, mode = op['path'], dec(op['data']), op.get('mode', OWRITE)
+    if host_path(path):
+        res.note = 'refused by guard: host tree'
+        return res
+    w = sess.wire
+    fid, err = w.walk_names(split(path))
+    if fid is None:
+        res.add('walk', path, False, err)
+        return res
+    timeout = SLOW_TIMEOUT if slow(data) else TIMEOUT
+    ok, body = w.call(TOPEN, struct.pack('<IB', fid, mode), timeout)
+    res.add(open_kind(path, mode), path, ok, None if ok else body)
+    allowances(res, path, data)
+    opened = ok
+    sent = data
+    if ok:
+        offset = 0
+        for size in chunking(op.get('chunks', 'whole'), len(data), w.msize - 23):
+            piece = data[offset:offset + size]
+            ok, body = w.call(TWRITE, struct.pack('<IQI', fid, op.get('offset', offset), len(piece)) + piece, timeout)
+            # A write on a fid opened for reading is 9P misuse, not a write
+            # the tree refused: the rule does not name it.
+            res.add('write' if mode & 3 else 'write-badfid', path, ok, None if ok else body)
+            offset += size
+            sent = data[:offset]
+            if not ok and op.get('stop_on_error', True):
+                break
+    ok, body = w.call(TCLUNK, struct.pack('<I', fid), timeout)
+    res.add('clunk-write' if mode & 3 else 'clunk', path, ok, None if ok else body)
+    # What the open holds when it closes: the last line written, if unended.
+    if opened and mode & 3 and sent and not sent.endswith(b'\n'):
+        res.add('close-runs', path, ok)
+    return res
+
+
+def allowances(res, path, data):
+    """Errs a successful write may log (fs.md): a look that finds nothing logs
+    one, per line, whether written to look, written back as an event record
+    (acme's xfid.c:842) or run as the Look builtin; an Edit block whose text
+    never ended runs, and fails, at the close."""
+    base = path.rsplit('/', 1)[-1]
+    lines = [line for line in data.split(b'\n') if line.strip()]
+    if base in ('look', 'event'):
+        n = len(lines)
+    elif base in ('exec', 'ctl', 'tagexec'):
+        n = sum(1 for line in lines if line.split()[:1] == [b'Look'])
+        if b'Edit' in data and len(lines) > 1:
+            n += 1
+    else:
+        return
+    if n:
+        res.add('allow', path, True, n)
+
+
+def open_kind(path, mode):
+    """An open the rule names: pane/new's (it makes a pane) and a truncation
+    ('every write or truncation the tree refused'); others are 'open'."""
+    if path.rstrip('/') == '/pane/new' and not mode & 3:
+        return 'open-new'
+    if mode & OTRUNC and mode & 3:
+        return 'trunc'
+    return 'open'
+
+
+def chunking(spec, n, cap):
+    if n == 0:
+        return [0]
+    if spec == 'bytes':
+        return [1] * n
+    if isinstance(spec, list):
+        out, total = [], 0
+        for s in spec:
+            s = max(1, min(s, cap, n - total))
+            out.append(s)
+            total += s
+            if total >= n:
+                return out
+        while total < n:
+            s = min(cap, n - total)
+            out.append(s)
+            total += s
+        return out
+    out, total = [], 0
+    while total < n:
+        s = min(cap, n - total)
+        out.append(s)
+        total += s
+    return out
+
+
+@op_kind('read')
+def op_read(sess, op):
+    """open(path, OREAD); read from offset in chunks of count until EOF or cap."""
+    res = Result()
+    path, count, offset = op['path'], op.get('count', 8192), op.get('offset', 0)
+    w = sess.wire
+    fid, err = w.walk_names(split(path))
+    if fid is None:
+        res.add('walk', path, False, err)
+        return res
+    mode = op.get('mode', OREAD)
+    ok, body = w.call(TOPEN, struct.pack('<IB', fid, mode))
+    res.add(open_kind(path, mode), path, ok, None if ok else body)
+    if ok:
+        for _ in range(op.get('reads', 4)):
+            payload = struct.pack('<IQI', fid, offset, min(count, w.msize - 11))
+            if is_held(path):
+                ok, body = w.call_held(TREAD, payload)
+            else:
+                ok, body = w.call(TREAD, payload)
+            res.add('read', path, ok, body if ok is False else None)
+            if not ok or len(body) <= 4:
+                break
+            offset += len(body) - 4
+    ok, body = w.call(TCLUNK, struct.pack('<I', fid))
+    res.add('clunk', path, ok, None if ok else body)
+    return res
+
+
+@op_kind('stat')
+def op_stat(sess, op):
+    res = Result()
+    fid, err = sess.wire.walk_names(split(op['path']))
+    res.add('walk', op['path'], fid is not None, err)
+    if fid is not None:
+        ok, body = sess.wire.call(TSTAT, struct.pack('<I', fid))
+        res.add('stat', op['path'], ok, None if ok else body)
+        sess.wire.clunk(fid)
+    return res
+
+
+@op_kind('remove')
+def op_remove(sess, op):
+    res = Result()
+    if host_path(op['path']):
+        res.note = 'refused by guard: host tree'
+        return res
+    fid, err = sess.wire.walk_names(split(op['path']))
+    res.add('walk', op['path'], fid is not None, err)
+    if fid is not None:
+        ok, body = sess.wire.call(TREMOVE, struct.pack('<I', fid), SLOW_TIMEOUT)
+        res.add('remove', op['path'], ok, None if ok else body)
+    return res
+
+
+@op_kind('create')
+def op_create(sess, op):
+    res = Result()
+    if host_path(op['path']):
+        res.note = 'refused by guard: host tree'
+        return res
+    fid, err = sess.wire.walk_names(split(op['path']))
+    res.add('walk', op['path'], fid is not None, err)
+    if fid is not None:
+        ok, body = sess.wire.call(TCREATE, struct.pack('<I', fid) + string(dec(op['name'])) +
+                                  struct.pack('<IB', op.get('perm', 0o666), op.get('mode', OREAD)))
+        res.add('create', op['path'] + '/' + op['name'], ok, None if ok else body)
+        sess.wire.clunk(fid)
+    return res
+
+
+@op_kind('hold')
+def op_hold(sess, op):
+    """Open and keep a fid across later ops as handle op['h']."""
+    res = Result()
+    path, mode = op['path'], op.get('mode', OREAD)
+    if host_path(path) and mode & 3:
+        res.note = 'refused by guard: host tree'
+        return res
+    fid, err = sess.wire.walk_names(split(path))
+    if fid is None:
+        res.add('walk', path, False, err)
+        return res
+    ok, body = sess.wire.call(TOPEN, struct.pack('<IB', fid, mode))
+    res.add(open_kind(path, mode), path, ok, None if ok else body)
+    if ok:
+        sess.handles[op['h']] = [fid, path, mode, [0], b'']
+    else:
+        sess.wire.clunk(fid)
+    return res
+
+
+@op_kind('hwrite')
+def op_hwrite(sess, op):
+    res = Result()
+    h = sess.handles.get(op['h'])
+    if h is None:
+        res.note = 'no such handle'
+        return res
+    fid, path, mode, off, _ = h
+    data = dec(op['data'])[:sess.wire.msize - 23]  # one Twrite
+    h[4] = data
+    allowances(res, path, data)
+    ok, body = sess.wire.call(TWRITE, struct.pack('<IQI', fid, off[0], len(data)) + data,
+                              SLOW_TIMEOUT if slow(data) else TIMEOUT)
+    res.add('write' if mode & 3 else 'write-badfid', path, ok, None if ok else body)
+    if ok:
+        off[0] += len(data)
+    return res
+
+
+@op_kind('hread')
+def op_hread(sess, op):
+    res = Result()
+    h = sess.handles.get(op['h'])
+    if h is None:
+        res.note = 'no such handle'
+        return res
+    fid, path, _, off, _ = h
+    offset = op['offset'] if op.get('offset') is not None else off[0]
+    payload = struct.pack('<IQI', fid, offset, min(op.get('count', 8192), sess.wire.msize - 11))
+    ok, body = sess.wire.call_held(TREAD, payload) if is_held(path) else sess.wire.call(TREAD, payload)
+    res.add('read', path, ok, body if ok is False else None)
+    if ok:
+        off[0] = offset + len(body) - 4
+    return res
+
+
+@op_kind('clunk')
+def op_clunk(sess, op):
+    res = Result()
+    h = sess.handles.pop(op['h'], None)
+    if h is None:
+        res.note = 'no such handle'
+        return res
+    ok, body = sess.wire.call(TCLUNK, struct.pack('<I', h[0]))
+    res.add('clunk-write' if h[2] & 3 else 'clunk', h[1], ok, None if ok else body)
+    if h[2] & 3 and h[4] and not h[4].endswith(b'\n'):
+        res.add('close-runs', h[1], ok)
+    return res
+
+
+@op_kind('restore')
+def op_restore(sess, op):
+    """Write Restore [f] to /ctl: refused with an Rerror, or the server
+    hangs up and serves the dump's panes again (fs.md: a Restore hangs its
+    connection up)."""
+    res = Result()
+    data = dec(op['data'])
+    try:
+        ok, why = sess.wire.write_path('/ctl', data)
+        res.add('write', '/ctl', ok, why)
+        if ok:
+            # Taken: the hangup follows the reply.
+            try:
+                sess.wire.call(TSTAT, struct.pack('<I', 1), 2.0)
+            except (Dropped, Hang):
+                pass
+            res.hung_up = True
+    except Dropped:
+        res.add('write', '/ctl', True, None)
+        res.hung_up = True
+    if res.hung_up:
+        sess.reconnect()
+    return res
+
+
+@op_kind('host')
+def op_host(sess, op):
+    """A change on disk under the session's HOME: write or delete a file."""
+    res = Result()
+    name = op['name']
+    assert '/' not in name and name not in ('.', '..')
+    path = os.path.join(sess.home, name)
+    if op['action'] == 'rm':
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+    else:
+        with open(path, 'wb') as f:
+            f.write(dec(op['data']))
+    return res
+
+
+@op_kind('seq')
+def op_seq(sess, op):
+    """Several ops as one step (e.g. clearing the window)."""
+    res = Result()
+    for sub in op['ops']:
+        r = OPS[sub['op']](sess, sub)
+        res.requests.extend(r.requests)
+        res.hung_up |= r.hung_up
+    return res
+
+
+# --------------------------------------------------------------------------
+# Input: data pools and generators (deliberately simple; swap freely).
+
+SAFE = (b'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 '
+        b'-_=+,:#^@%()\n\r\t\0')
+UNSAFE = set(b'/~$`;|&<>.\\*?![]{}\'"')
+
+
+def shell_safe(data):
+    """No path or shell metacharacters, whatever else it holds."""
+    return bytes(b for b in data if b not in UNSAFE)
+
+
+RUNES = ['é', 'é', '日本', '\U0001f600', 'Δ', ' ', '​', '\u0080', '\u009f']
+ODD = [b'', b'\n', b'\r\n', b'\0', b'\xff', b'\xfe\xff', b'\xe6\x97', b'\xc0\x80', b'\x1b[2J', b'\x7f', b'\t',
+       b'\n\n\n', b' ', b'  \n']
+WORDS = [b'alpha', b'beta', b'foo', b'bar', b'x', b'hello', b'zz', b'0', b'1', b'-1', b'99999999999',
+         b'on', b'off', b'follow', b'follow new', b'lock', b'unlock', b'clean', b'dirty', b'show',
+         b'dot=addr', b'addr=dot', b'limit=addr', b'nomark', b'mark', b'get', b'put', b'del', b'delete',
+         b'name', b'cleartag', b'answer', b'answer -', b'size 80 24', b'size 20 6', b'size 5000 5000',
+         b'winsize 80 24', b'winsize 0 0', b'sig INT', b'sig TERM', b'sig HUP', b'sig KILL', b'sig BOGUS',
+         b'exec', b'Placement acme', b'Placement pardes', b'Wrap on', b'Wrap off']
+SHELL_LINES = [b'true', b'false', b'echo hi', b'echo m9p', b'sleep 1', b'ls', b'cat nonexist',
+               b'printf abc', b'exit 3', b'seq 1 2000', b'yes | head -n 3000', b'date', b'pwd', b'env | wc']
+DENY = {'Exit', 'Attach', 'Detach', 'Mount', 'Unmount', 'Tty9p', 'ClipYank', 'ClipYankMain', 'ClipPaste',
+        'ClipPasteBefore', 'ClipReplace', 'Shell', 'ThemeFile', 'EffectCode', 'Help', 'Tutor', 'Changelog',
+        'Rename', 'Hover', 'CodeAction', 'SelectRefs', 'Symbols', 'Diagnostics', 'WsDiagnostics', 'Callers',
+        'Callees', 'Supertypes', 'Subtypes', 'WsSymbols', 'Lspinfo', 'Lspwhy', 'Restore'}
+BUILTINS = ['Look', 'Exec', 'Kill', 'Dump', 'Msg', 'NextColor', 'Themes', 'DumpThemes', 'TreeContext', 'PdfFit',
+            'PdfTint', 'PdfSections', 'Petscii', 'Palette', 'Ascii', 'Save', 'New', 'Newcol', 'Del', 'Filter',
+            'Mode', 'Togglettymode', 'Edit', 'Undo', 'Redo', 'Collapse', 'Delcol', 'DelAbove', 'DelBelow', 'Tty',
+            'Repl', 'Joincol', 'Config', 'LocationsConfig', 'Messages', 'Mini', 'Find', 'Grep', 'Left', 'Down',
+            'Up', 'Right', 'Back', 'Forward', 'Last', 'Recent', 'Jumplist', 'Colors', 'Wrap', 'Tagbottom', 'Debug',
+            'FocusTint', 'Verbose', 'Motion', 'InactiveDim', 'DumpDir', 'Theme', 'Placement', 'BootShell',
+            'LookWord', 'Get', 'Put', 'Undo', 'Redo', 'Zerox', 'Snarf', 'Paste', 'Cut', 'Sort', 'Font', 'Local']
+PANE_FILES = ['name', 'body', 'tag', 'ctl', 'addr', 'dot', 'limit', 'data', 'xdata', 'sel', 'dirty', 'mark',
+              'scroll', 'errors', 'event', 'look', 'exec', 'tagexec', 'pty', 'pty/run', 'pty/data', 'pty/ctl',
+              'pty/status']
+ROOT_FILES = ['README', 'index', 'status', 'focus', 'ctl', 'commands', 'recent', 'look', 'exec', 'log', 'screen',
+              'listeners', 'layout', 'tag', 'tagexec', 'pane', 'pane/new', 'col', '']
+COL_FILES = ['tag', 'ctl', 'exec']
+WRITABLE = ['ctl', 'exec', 'look', 'tag', 'tagexec', 'focus', 'body', 'data', 'xdata', 'addr', 'dot', 'limit',
+            'name', 'sel', 'dirty', 'mark', 'scroll', 'errors', 'event', 'pty/run', 'pty/data', 'pty/ctl', 'log']
+
+
+class Ctx:
+    """What the generators and invariants see."""
+
+    def __init__(self, seed):
+        self.seed = seed
+        self.rng = random.Random(seed)
+        self.sess = None
+        self.step = 0
+        self.serials = []
+        self.cols = []
+        self.gone = []       # serials seen and since closed (stale)
+        self.next_h = 0
+        self.last_index = b''
+        self.verbose = False
+
+
+def pick_serial(ctx, rng):
+    """A live pane as a token, {pN}: the Nth serial of /index (mod its
+    length) when the op runs, so a shrunk replay still names a pane that
+    exists; or a stale or impossible serial, literally."""
+    r = rng.random()
+    if ctx.serials and r < 0.85:
+        return '{p%d}' % rng.randrange(64)
+    if ctx.gone and r < 0.95:
+        return rng.choice(ctx.gone)
+    return rng.choice([0, 1, 999, 4294967295, 4294967296, -1])
+
+
+def pick_col(ctx, rng):
+    if ctx.cols and rng.random() < 0.85:
+        return '{c%d}' % rng.randrange(16)
+    return rng.choice([0, 99, 4294967295])
+
+
+def odd_bytes(rng, n):
+    out = bytearray()
+    while len(out) < n:
+        r = rng.random()
+        if r < 0.5:
+            out += rng.choice(WORDS)
+        elif r < 0.7:
+            out += rng.choice(RUNES).encode()
+        elif r < 0.9:
+            out += rng.choice(ODD)
+        else:
+            out.append(rng.randrange(256))
+        if rng.random() < 0.5:
+            out += b' '
+    return bytes(out[:n])
+
+
+def payload(rng, maxlen=64):
+    """Plain or odd bytes, sometimes huge, with or without a newline."""
+    r = rng.random()
+    if r < 0.35:
+        data = rng.choice(WORDS)
+    elif r < 0.55:
+        data = odd_bytes(rng, rng.randrange(0, maxlen))
+    elif r < 0.62:
+        data = rng.choice(ODD)
+    elif r < 0.66:
+        data = (rng.choice(WORDS) + b' ') * rng.randrange(100, 20000)  # huge
+    elif r < 0.7:
+        data = b'x' * rng.choice([255, 256, 1023, 1024, 1025, 65535, 65536, 70000])
+    else:
+        data = rng.choice(WORDS) + b' ' + odd_bytes(rng, rng.randrange(0, 12))
+    if rng.random() < 0.6 and not data.endswith(b'\n'):
+        data += rng.choice([b'\n', b'\n', b'\r\n', b'\n\n'])
+    return shell_safe(data)
+
+
+def weird_name(rng):
+    r = rng.random()
+    if r < 0.3:
+        return 'x' * rng.choice([255, 256, 300, 1000])
+    if r < 0.5:
+        return rng.choice(['..', '.', '', ' ', 'new', 'NEW', '01', '1 ', '\n', 'a\0b'])
+    if r < 0.7:
+        return enc(rng.choice(RUNES).encode())
+    return enc(bytes(rng.randrange(1, 256) for _ in range(rng.randrange(1, 20))).replace(b'/', b'_'))
+
+
+def random_path(ctx, rng, writable=False):
+    r = rng.random()
+    if r < 0.6:
+        f = rng.choice(WRITABLE if writable else PANE_FILES)
+        return f'/pane/{pick_serial(ctx, rng)}/{f}'
+    if r < 0.75:
+        f = rng.choice(['ctl', 'exec', 'look', 'tag', 'tagexec', 'focus', 'log'] if writable else ROOT_FILES)
+        return '/' + f
+    if r < 0.85:
+        return f'/col/{pick_col(ctx, rng)}/{rng.choice(COL_FILES)}'
+    if r < 0.93:
+        return f'/pane/{pick_serial(ctx, rng)}/{weird_name(rng)}'
+    return '/' + '/'.join(weird_name(rng) for _ in range(rng.randrange(1, 4)))
+
+
+def data_for(ctx, rng, path):
+    """Payload a file is likely to take, often mutated."""
+    base = path.rsplit('/', 1)[-1]
+    r = rng.random()
+    if r < 0.25:
+        return payload(rng)
+    if base in ('addr', 'dot', 'limit', 'sel'):
+        return address(rng)
+    if base == 'ctl' and path.endswith('pty/ctl'):
+        return rng.choice([b'sig INT\n', b'sig TERM\n', b'sig HUP\n', b'sig QUIT\n', b'sig KILL\n',
+                           b'winsize 80 24\n', b'winsize 1 1\n', b'winsize 99999 3\n', b'exec\n', b'sig\n'])
+    if base in ('ctl', 'exec', 'tagexec'):
+        return builtin_line(ctx, rng, path)
+    if base == 'run':
+        return shell_safe(rng.choice(SHELL_LINES)) + b'\n'
+    if base == 'data' and '/pty/' in path:
+        return shell_safe(rng.choice(SHELL_LINES)) + rng.choice([b'\r', b'\n', b'', b'\x03', b'\x04'])
+    if base == 'event':
+        return event_record(rng)
+    if base == 'focus':
+        return str(pick_serial(ctx, rng)).encode() + rng.choice([b'\n', b''])
+    if base == 'look':
+        return shell_safe(rng.choice([b'alpha', b'alpha.txt', b'alpha.txt:2', b'utf8.txt:#3', b'nothere',
+                                      b'bad.txt', b'crlf.txt:1', b'gamma', b'beta'])) + b'\n'
+    if base in ('dirty', 'mark', 'scroll'):
+        return rng.choice([b'0', b'1', b'0\n', b'1\n', b'2\n', b'', b'yes\n'])
+    if base == 'log':
+        return rng.choice([b'follow\n', b'follow new\n', b'follow', b'nope\n'])
+    return payload(rng, 200)
+
+
+def mutate(rng, data, rounds=None):
+    data = bytearray(data)
+    for _ in range(rounds or rng.randrange(1, 4)):
+        if not data:
+            data += rng.choice(WORDS)
+            continue
+        i = rng.randrange(len(data))
+        r = rng.random()
+        if r < 0.3:
+            del data[i]
+        elif r < 0.55:
+            data[i:i] = shell_safe(rng.choice(ODD) or b'\0')
+        elif r < 0.75:
+            j = rng.randrange(i, len(data) + 1)
+            data[i:i] = data[i:j]
+        else:
+            data[i:i] = rng.choice(RUNES).encode()
+    return bytes(data)
+
+
+def builtin_line(ctx, rng, path):
+    word = rng.choice(BUILTINS).encode()
+    r = rng.random()
+    if r < 0.4:
+        line = word
+    elif r < 0.7:
+        line = word + b' ' + rng.choice(WORDS)
+    elif r < 0.8:
+        line = b'Edit ' + edit_command(rng)
+        return line + b'\n'  # Edit text keeps its / delimiters
+    else:
+        line = mutate(rng, word + b' ' + rng.choice(WORDS))
+    if rng.random() < 0.1:
+        line = shell_safe(rng.choice(SHELL_LINES))
+    if rng.random() < 0.8:
+        line += b'\n'
+    return shell_safe(line)
+
+
+# sam's commands without the ones that reach files or shells.
+EDIT_TEXT = b'abcdghijklmopqstuvxyz0123456789 ACEGHIJKLMNOPQRSTUVWXYZ_-'
+
+
+def edit_text(rng):
+    s = bytes(rng.choice(EDIT_TEXT) for _ in range(rng.randrange(0, 8)))
+    if rng.random() < 0.3:
+        s += rng.choice(RUNES).encode()
+    return s
+
+
+def regex(rng):
+    parts = [edit_text(rng) or b'a']
+    for _ in range(rng.randrange(0, 4)):
+        parts.append(rng.choice([b'(a|b)', b'^', b'$', b'[a-z]', b'[^ ]', b'.*', b'\\n',
+                                 '[é日]'.encode(), b'x+', b'(', b'[', b'\\', b'a**',
+                                 b'(((a)))', b'[z-a]', (b'a|' * rng.randrange(1, 300)) + b'b']))
+        parts.append(edit_text(rng))
+    return b''.join(parts)
+
+
+def address(rng):
+    """sam addresses: valid, near-valid and garbage."""
+    simple = [b'0', b'$', b'.', b',', b';', b'#0', b'#3', b'#99999', b'1', b'2', b'3:2', b'1:1', b'0:0',
+              b'#1,#2', b'1,$', b'2,1', b'-', b'+', b'-1', b'+2', b'#-1', b'99999', b'1:99', b'?a?']
+    r = rng.random()
+    if r < 0.4:
+        a = rng.choice(simple)
+    elif r < 0.7:
+        a = b'/' + regex(rng) + b'/'
+    elif r < 0.85:
+        a = rng.choice(simple) + rng.choice([b',', b';', b'+', b'-']) + rng.choice(simple + [b'/a/'])
+    else:
+        a = mutate(rng, rng.choice(simple) + b'/' + regex(rng) + b'/')
+    return a + rng.choice([b'', b'\n'])
+
+
+def edit_command(rng):
+    t = lambda: edit_text(rng)  # noqa: E731
+    cmds = [lambda: b',x/' + regex(rng) + b'/c/' + t() + b'/',
+            lambda: b's/' + regex(rng) + b'/' + t() + b'/g',
+            lambda: b'a/' + t() + b'/',
+            lambda: b'i/' + t() + b'/',
+            lambda: b'c/' + t() + b'/',
+            lambda: b'd',
+            lambda: b',d',
+            lambda: b',y/' + regex(rng) + b'/d',
+            lambda: b',x/' + regex(rng) + b'/g/' + regex(rng) + b'/d',
+            lambda: b',x/' + regex(rng) + b'/v/' + regex(rng) + b'/c/' + t() + b'/',
+            lambda: b'1,2m$',
+            lambda: b'1t0',
+            lambda: b'k',
+            lambda: b'p',
+            lambda: b'=',
+            lambda: b'u',
+            lambda: b'{\na/' + t() + b'/\ni/' + t() + b'/\n}',
+            lambda: b',x/\\n/a/' + t() + b'/',
+            lambda: b'0,$s/' + regex(rng) + b'/&&/g']
+    cmd = rng.choice(cmds)()
+    if rng.random() < 0.4:
+        cmd = address(rng).rstrip(b'\n') + cmd
+    if rng.random() < 0.2:
+        cmd = mutate(rng, cmd)
+        # Mutation never introduces a banned command letter or shell char.
+        cmd = bytes(b for b in cmd if b not in b'werfbBDn!<>|')
+    return cmd
+
+
+def event_record(rng):
+    origin = rng.choice(b'EFKMZ')
+    action = rng.choice(b'xXlLiIdDZ')
+    q0, q1 = rng.choice([0, 1, 5, 10]), rng.choice([0, 2, 7, 1 << 40, -1])
+    rec = bytes([origin, action]) + f'{q0} {q1} {rng.choice([0, 1, 2, 8])} '.encode()
+    text = shell_safe(edit_text(rng))
+    rec += str(len(text)).encode() + b' ' + text + b'\n'
+    if rng.random() < 0.3:
+        rec = mutate(rng, rec)
+    return shell_safe(rec)
+
+
+def new_handle(ctx):
+    ctx.next_h += 1
+    return f'h{ctx.next_h}'
+
+
+@generator(30)
+def gen_write(ctx, rng):
+    path = random_path(ctx, rng, writable=True)
+    data = data_for(ctx, rng, path)
+    mode = rng.choice([OWRITE, OWRITE, ORDWR, OWRITE | OTRUNC, ORDWR | OTRUNC, OREAD])
+    r = rng.random()
+    chunks = 'bytes' if r < 0.08 and len(data) <= 64 else (
+        [rng.randrange(1, 16) for _ in range(4)] if r < 0.15 else 'whole')
+    op = {'op': 'write', 'path': path, 'data': enc(data), 'mode': mode, 'chunks': chunks}
+    if rng.random() < 0.05:
+        op['offset'] = rng.choice([1, 100, 1 << 40])
+    return op
+
+
+@generator(14)
+def gen_read(ctx, rng):
+    path = random_path(ctx, rng)
+    return {'op': 'read', 'path': path, 'count': rng.choice([1, 2, 7, 64, 200, 8192, 1 << 20]),
+            'offset': rng.choice([0, 0, 0, 1, 3, 100, 1 << 33]), 'reads': rng.choice([1, 2, 8])}
+
+
+@generator(4)
+def gen_stat(ctx, rng):
+    path = random_path(ctx, rng)
+    if rng.random() < 0.2:
+        path = '/os/' + rng.choice(['etc', 'etc/hosts', 'nonexistent', 'x' * 300])
+    return {'op': 'stat', 'path': path}
+
+
+@generator(5)
+def gen_newpane(ctx, rng):
+    if rng.random() < 0.7:
+        return {'op': 'read', 'path': '/pane/new', 'count': 64, 'reads': 1}
+    return {'op': 'hold', 'h': new_handle(ctx), 'path': '/pane/new', 'mode': rng.choice([OREAD, ORDWR])}
+
+
+@generator(2)
+def gen_remove(ctx, rng):
+    r = rng.random()
+    if r < 0.6:
+        path = f'/pane/{pick_serial(ctx, rng)}'
+    elif r < 0.85:
+        path = f'/col/{pick_col(ctx, rng)}'
+    else:
+        path = random_path(ctx, rng)
+    return {'op': 'remove', 'path': path}
+
+
+@generator(2)
+def gen_create(ctx, rng):
+    where = rng.choice(['/col', '/pane', '/', f'/pane/{pick_serial(ctx, rng)}'])
+    return {'op': 'create', 'path': where, 'name': weird_name(rng) if rng.random() < 0.5 else 'new',
+            'perm': rng.choice([0o666, DMDIR | 0o777]), 'mode': rng.choice([OREAD, OWRITE])}
+
+
+@generator(2)
+def gen_clear_window(ctx, rng):
+    """Down to the empty window (every pane closed), sometimes rebuilt."""
+    keep = rng.choice(ctx.serials) if ctx.serials and rng.random() < 0.8 else None
+    ops = [{'op': 'remove', 'path': f'/pane/{s}'} for s in ctx.serials if s != keep]
+    if rng.random() < 0.5:
+        ops += [{'op': 'remove', 'path': f'/col/{c}'} for c in ctx.cols]
+    if rng.random() < 0.6:
+        ops.append({'op': 'read', 'path': '/pane/new', 'count': 64, 'reads': 1})
+    return {'op': 'seq', 'ops': ops}
+
+
+@generator(6)
+def gen_hold(ctx, rng):
+    held = [h for h in (ctx.sess.handles if ctx.sess else {})]
+    r = rng.random()
+    if not held or r < 0.35:
+        path = random_path(ctx, rng)
+        if rng.random() < 0.3:
+            path = rng.choice(['/log', f'/pane/{pick_serial(ctx, rng)}/event', '/screen', '/index',
+                               f'/pane/{pick_serial(ctx, rng)}/ctl'])
+        return {'op': 'hold', 'h': new_handle(ctx), 'path': path, 'mode': rng.choice([OREAD, ORDWR, OWRITE])}
+    h = rng.choice(held)
+    path = ctx.sess.handles[h][1]
+    if r < 0.6:
+        return {'op': 'hwrite', 'h': h, 'data': enc(data_for(ctx, rng, path))}
+    if r < 0.85:
+        return {'op': 'hread', 'h': h, 'count': rng.choice([1, 16, 8192]),
+                'offset': rng.choice([None, None, 0, 5])}
+    return {'op': 'clunk', 'h': h}
+
+
+@generator(6)
+def gen_shell(ctx, rng):
+    s = pick_serial(ctx, rng)
+    line = shell_safe(rng.choice(SHELL_LINES))
+    r = rng.random()
+    if r < 0.25:
+        return {'op': 'write', 'path': f'/pane/{s}/ctl', 'data': enc(b'Tty\n'), 'mode': OWRITE}
+    if r < 0.45:
+        return {'op': 'write', 'path': f'/pane/{s}/pty/run', 'data': enc(line + b'\n'), 'mode': ORDWR}
+    if r < 0.6:
+        return {'op': 'write', 'path': f'/pane/{s}/pty/data', 'data': enc(line + b'\r'), 'mode': OWRITE}
+    if r < 0.75:
+        return {'op': 'write', 'path': f'/pane/{s}/pty/ctl',
+                'data': enc(rng.choice([b'sig INT\n', b'sig TERM\n', b'sig KILL\n', b'sig HUP\n'])), 'mode': OWRITE}
+    return {'op': 'write', 'path': rng.choice(['/exec', f'/pane/{s}/exec']), 'data': enc(line + b'\n'),
+            'mode': OWRITE}
+
+
+@generator(6)
+def gen_edit(ctx, rng):
+    s = pick_serial(ctx, rng)
+    path = rng.choice([f'/pane/{s}/ctl', f'/pane/{s}/exec', '/exec'])
+    return {'op': 'write', 'path': path, 'data': enc(b'Edit ' + edit_command(rng) + b'\n'), 'mode': OWRITE}
+
+
+@generator(8)
+def gen_addr(ctx, rng):
+    s = pick_serial(ctx, rng)
+    ops = [{'op': 'write', 'path': f'/pane/{s}/addr', 'data': enc(address(rng)), 'mode': OWRITE}]
+    r = rng.random()
+    if r < 0.4:
+        ops.append({'op': 'read', 'path': f'/pane/{s}/{rng.choice(["xdata", "data", "addr", "dot"])}',
+                    'count': 8192, 'reads': 2})
+    elif r < 0.7:
+        ops.append({'op': 'write', 'path': f'/pane/{s}/{rng.choice(["data", "xdata"])}',
+                    'data': enc(payload(rng, 20)), 'mode': rng.choice([OWRITE, OWRITE | OTRUNC])})
+    return {'op': 'seq', 'ops': ops}
+
+
+@generator(3)
+def gen_tag(ctx, rng):
+    path = rng.choice([f'/pane/{pick_serial(ctx, rng)}/tag', '/tag', f'/col/{pick_col(ctx, rng)}/tag'])
+    return {'op': 'write', 'path': path, 'data': enc(payload(rng, 40)),
+            'mode': rng.choice([OWRITE, OWRITE | OTRUNC])}
+
+
+@generator(3)
+def gen_event(ctx, rng):
+    s = pick_serial(ctx, rng)
+    return {'op': 'hold', 'h': new_handle(ctx), 'path': f'/pane/{s}/event', 'mode': ORDWR}
+
+
+@generator(2)
+def gen_dump_restore(ctx, rng):
+    r = rng.random()
+    if r < 0.5:
+        return {'op': 'write', 'path': '/ctl', 'data': enc(rng.choice([b'Dump\n', b'Dump', b'Dump m9pdump\n'])),
+                'mode': OWRITE}
+    return {'op': 'restore', 'data': enc(rng.choice([b'Restore\n', b'Restore\n', b'Restore m9pdump\n',
+                                                     b'Restore nothere\n']))}
+
+
+@generator(3)
+def gen_host(ctx, rng):
+    name = rng.choice(list(FIXTURES) + ['fresh.txt'])
+    if rng.random() < 0.5:
+        return {'op': 'host', 'action': 'rm', 'name': name}
+    return {'op': 'host', 'action': 'write', 'name': name, 'data': enc(payload(rng, 100))}
+
+
+@generator(3)
+def gen_look_fixture(ctx, rng):
+    name = rng.choice(list(FIXTURES) + ['fresh.txt'])
+    suffix = rng.choice([b'', b':2', b':#4', b':1:3', b':/beta/'])
+    return {'op': 'write', 'path': rng.choice(['/look', f'/pane/{pick_serial(ctx, rng)}/look']),
+            'data': enc(name.encode() + suffix + b'\n'), 'mode': OWRITE}
+
+
+# --------------------------------------------------------------------------
+# Log window: the records one step caused, between two unique markers.
+
+MARK = re.compile(r'unknown control message "(m9p-mark-\d+)"')
+
+
+def write_marker(sess, step):
+    """An unknown ctl word logs `err - ctl: unknown control message "<word>"`."""
+    word = f'm9p-mark-{step}'
+    ok, why = sess.wire.write_path('/ctl', word.encode() + b'\n')
+    if ok:
+        return None, 'marker write was taken'
+    prev, sess.marker = sess.marker, word
+    if sess.follower is not None:
+        end = sess.follower.wait_for(f'"{word}"')
+        if end is None:
+            return None, 'marker record never reached the follower'
+        lines = sess.follower.lines[:end]
+        start = 0
+        if prev is not None:
+            for i in range(len(lines) - 1, -1, -1):
+                if f'"{prev}"' in lines[i]:
+                    start = i + 1
+                    break
+            else:
+                return None, None
+        window = lines[start:]
+        del sess.follower.lines[:end]
+        return window, None
+    lines = sess.wire.read_path('/log').decode(errors='replace').splitlines()
+    end = next((i for i in range(len(lines) - 1, -1, -1) if f'"{word}"' in lines[i]), None)
+    if end is None:
+        return None, 'marker record missing from /log'
+    if prev is None:
+        return None, None
+    start = next((i for i in range(end - 1, -1, -1) if f'"{prev}"' in lines[i]), None)
+    if start is None:
+        return None, None  # the ring moved past it
+    return lines[start + 1:end], None
+
+
+COUNT = re.compile(r'^(.*) \(x(\d+)\)$')
+
+
+def occurrences(lines, kind):
+    """Records of `kind` (err|msg), counting `(xN)` repeats (fs.md: a record said
+    again is counted; a follower sees each count as a line of its own)."""
+    out, last_text, last_n = [], None, 0
+    for line in lines:
+        if not line.startswith(kind + ' '):
+            last_text = None
+            continue
+        m = COUNT.match(line)
+        text, n = (m.group(1), int(m.group(2))) if m else (line, 1)
+        if text == last_text:
+            new = n - last_n
+        else:
+            new = n
+        out.extend([text] * max(new, 0))
+        last_text, last_n = text, n
+    return out
+
+
+# --------------------------------------------------------------------------
+# Invariants: small, cheap, each with its doc citation.
+
+@invariant
+def alive(ctx, res):
+    """Never crashes or panics (src/crash.zig: a panic leaves its crash file)."""
+    if not ctx.sess.alive():
+        return f'pardes died (exit {ctx.sess.proc.returncode})'
+    if ctx.sess.has_crash_file():
+        return 'crash file written: ' + ctx.sess.crash_text()[-2000:]
+    return None
+
+
+@invariant
+def one_failure_rule(ctx, res):
+    """fs.md 'One rule for what fails': a write fails whenever what it asked
+    for fails ... and logs its reason exactly once, as `err <serial|-> <file>:
+    <why>`, with no `msg` for it. What is not a failure: a look that finds
+    nothing answers nothing and logs one `err`, the write succeeding."""
+    if res.window is None or res.hung_up:
+        return None
+    writes = res.writes()
+    # A refused open with OTRUNC may be refused for the open, not the
+    # truncation (event's `file in use`): allowed an err or none, below.
+    failed = [w for w in writes if w[2] is False and w[0] != 'trunc']
+    errs = occurrences(res.window, 'err')
+    msgs = occurrences(res.window, 'msg')
+    # Only for a write that succeeded: a failed one logs its one err.
+    failed_paths = {w[1] for w in failed}
+    allowed = sum(r[3] for r in res.requests if r[0] == 'allow' and r[1] not in failed_paths)
+    # fs.md 'A write of command lines ... what is left when it closes runs at
+    # the close ... and its failure is in the log alone'.
+    closes = [r for r in res.requests if r[0] == 'close-runs']
+    # Refused opens (other than pane/new), removes and creates are not named
+    # by the rule: allow an err or none.
+    other = [r for r in res.requests if r[2] is False and (
+        r[0] in ('remove', 'create', 'trunc') or r[0] == 'open' and r[1].rstrip('/') != '/pane/new')]
+    lo, hi = len(failed), len(failed) + allowed + len(closes) + len(other)
+    if not lo <= len(errs) <= hi:
+        return (f'{len(failed)} failed write(s) logged {len(errs)} err record(s) '
+                f'(allowed {lo}..{hi}); failed={[(w[0], w[1], w[3]) for w in failed]} window={res.window}')
+    if failed and msgs:
+        return f'a failed write also logged msg: {msgs}; failed={[(w[1], w[3]) for w in failed]}'
+    return None
+
+
+LINE_FILES = ('look', 'exec', 'tagexec', 'ctl')
+
+
+@invariant
+def close_runs_only_line_files(ctx, res):
+    """fs.md: 'A write of command lines -- to look, exec, tagexec, a ctl of the
+    root, a pane or a column, or a column's exec -- runs each line once it is
+    whole ... what is left when it closes runs at the close'. Another file
+    that fails only at its close breaks 'a write fails whenever what it asked
+    for fails' (or the list is short)."""
+    if res.window is None:
+        return None
+    closes = [r[1] for r in res.requests if r[0] == 'close-runs']
+    failed_files = {w[1].rsplit('/', 1)[-1] for w in res.writes() if w[2] is False}
+    for line in occurrences(res.window, 'err'):
+        m = re.match(r'err \S+ ([^:]+):', line)
+        if not m:
+            continue
+        name = m.group(1).rsplit('/', 1)[-1]
+        if name in LINE_FILES or name in failed_files:
+            continue
+        for path in closes:
+            if path.rsplit('/', 1)[-1] == name:
+                return f'a write to {name} with no newline was answered, then failed at the close: {line}'
+    return None
+
+
+@invariant
+def index_matches_pane_dirs(ctx, res):
+    """fs.md /index: one line per pane; /pane/<n>/ each pane's directory."""
+    w = ctx.sess.wire
+    first = w.read_path('/index')
+    fid, err = w.walk_names([b'pane'])
+    if fid is None:
+        return f'walking /pane failed: {err}'
+    ok, _ = w.call(TOPEN, struct.pack('<IB', fid, OREAD))
+    names = dir_names(w.read_all(fid))
+    w.clunk(fid)
+    second = w.read_path('/index')
+    ctx.last_index = second
+    serials = []
+    for line in second.decode(errors='replace').splitlines():
+        try:
+            serials.append(int(line.split()[0]))
+        except (IndexError, ValueError):
+            return f'/index line unparsable: {line!r}'
+    old = set(ctx.serials)
+    ctx.serials = serials
+    ctx.gone = sorted(set(ctx.gone) | (old - set(serials)))[-20:]
+    if first != second:
+        return None  # it changed between the reads; no claim
+    dirs = sorted(int(n) for n in names if n.isdigit())
+    extra = [n for n in names if not n.isdigit() and n != 'new']
+    if extra:
+        return f'/pane lists non-serial entries {extra}'
+    if sorted(serials) != dirs:
+        return f'/index serials {sorted(serials)} != /pane dirs {dirs}'
+    return None
+
+
+LISTINGS = ['/index', '/layout', '/', '/pane', '/col', '/pane/{s}', '/status', '/commands', '/pane/{s}/tag',
+            '/recent', '/README']
+
+
+@invariant
+def chunked_equals_whole(ctx, res):
+    """9P read(5): reads at successive offsets see the one file; a listing read
+    in chunks equals one read whole within the same open."""
+    w = ctx.sess.wire
+    path = LISTINGS[ctx.step % len(LISTINGS)]
+    if '{s}' in path:
+        if not ctx.serials:
+            return None
+        path = path.format(s=ctx.serials[ctx.step % len(ctx.serials)])
+    fid, err = w.walk_names(split(path))
+    if fid is None:
+        return None  # gone since
+    try:
+        ok, body = w.call(TOPEN, struct.pack('<IB', fid, OREAD))
+        if not ok:
+            return None
+        whole = w.read_all(fid)
+        isdir = path in ('/', '/pane', '/col') or path.count('/') == 2 and path.startswith('/pane/')
+        size = 128 + (ctx.step * 37) % 900 if isdir else 1 + (ctx.step * 7) % 50
+        chunked = w.read_all(fid, count=size)
+        if isdir:
+            whole_n, chunk_n = dir_names(whole), dir_names(chunked)
+            if whole_n != chunk_n:
+                return f'{path}: whole {whole_n} != chunked({size}) {chunk_n}'
+        elif whole != chunked:
+            return f'{path}: whole {whole[:200]!r} != chunked({size}) {chunked[:200]!r}'
+    finally:
+        w.clunk(fid)
+    return None
+
+
+# --------------------------------------------------------------------------
+# Runner.
+
+class Failure(Exception):
+    def __init__(self, kind, why, step):
+        super().__init__(f'{kind}: {why}')
+        self.kind, self.why, self.step = kind, why, step
+
+    def signature(self):
+        """What makes two failures the same bug: the kind, and the message with
+        numbers, quoted words and pane paths blurred."""
+        why = re.sub(r'"[^"]*"', '""', self.why)
+        why = re.sub(r"'[^']*'", "''", why) if 'window=' not in why else why
+        why = re.sub(r'/home/\S*', 'PATH', why)
+        return self.kind + ':' + re.sub(r'\d+', 'N', why)[:300]
+
+
+def check_death(sess, step, what):
+    """After an error: did pardes quit cleanly (Ended), or die (Failure)?"""
+    deadline = time.monotonic() + 1.0
+    while sess.alive() and time.monotonic() < deadline:
+        time.sleep(0.02)
+    if sess.alive():
+        return
+    if sess.proc.returncode == 0 and not sess.has_crash_file():
+        raise Ended(what)
+    raise Failure('crash', f'{what}: pardes exit {sess.proc.returncode} ' + sess.crash_text()[:3000], step)
+
+
+TOKEN = re.compile(r'\{([pc])(\d+)\}')
+
+
+def resolve(ctx, op):
+    """Replace {pN}/{cN} in an op's paths with live serials."""
+    def sub(m):
+        live = ctx.serials if m.group(1) == 'p' else ctx.cols
+        return str(live[int(m.group(2)) % len(live)]) if live else '0'
+    out = dict(op)
+    if 'path' in out:
+        out['path'] = TOKEN.sub(sub, out['path'])
+    if 'ops' in out:
+        out['ops'] = [resolve(ctx, o) for o in out['ops']]
+    if 'data' in out and '/focus' in out.get('path', ''):
+        out['data'] = TOKEN.sub(sub, out['data'])
+    return out
+
+
+def run_step(ctx, op):
+    """Run one op and every invariant. Raises Failure or Ended."""
+    sess = ctx.sess
+    op = resolve(ctx, op)
+    ctx.last_op = op
+    what = f'{op["op"]} {op.get("path", "")}'.strip()
+    try:
+        res = OPS[op['op']](sess, op)
+    except Hang as why:
+        check_death(sess, ctx.step, what)
+        raise Failure('hang', f'{what}: {why}', ctx.step)
+    except Dropped as why:
+        check_death(sess, ctx.step, what)
+        raise Failure('dropped', f'{what}: {why}', ctx.step)
+    res.window = None
+    try:
+        if not sess.alive():
+            check_death(sess, ctx.step, what)
+        # Liveness: a trivial request answers within the timeout.
+        ok, _ = sess.wire.call(TSTAT, struct.pack('<I', 1))
+        window, why = write_marker(sess, ctx.step)
+        if why:
+            raise Failure('log', why, ctx.step)
+        res.window = None if res.hung_up else window
+        if ctx.verbose:
+            print(f'  op {ctx.step}: {json.dumps(op)[:300]}')
+            for r in res.requests:
+                print(f'    {r[0]} {r[1]!r} -> {"ok" if r[2] else ("held" if r[2] is None else "Rerror")}'
+                      f'{": " + str(r[3]) if r[3] else ""}')
+            for line in window or []:
+                print(f'    log: {line}')
+        for check in INVARIANTS:
+            problem = check(ctx, res)
+            if problem:
+                raise Failure(check.__name__, problem, ctx.step)
+        # The column list, for the generators.
+        fid, err = sess.wire.walk_names([b'col'])
+        if fid is not None:
+            sess.wire.call(TOPEN, struct.pack('<IB', fid, OREAD))
+            ctx.cols = [int(n) for n in dir_names(sess.wire.read_all(fid)) if n.isdigit()]
+            sess.wire.clunk(fid)
+    except Hang as why:
+        check_death(sess, ctx.step, what)
+        raise Failure('hang', f'after {what}: {why}', ctx.step)
+    except Dropped as why:
+        check_death(sess, ctx.step, what)
+        raise Failure('dropped', f'after {what}: {why}', ctx.step)
+    except Failure as f:
+        # An invariant that failed because pardes was quitting is not one.
+        check_death(sess, ctx.step, what)
+        raise f
+    except OSError as why:
+        check_death(sess, ctx.step, what)
+        raise Failure('checker', f'after {what}: {why}', ctx.step)
+    return res
+
+
+class Runner:
+    def __init__(self, args):
+        self.args = args
+        self.out = args.out
+        os.makedirs(os.path.join(self.out, 'bugs'), exist_ok=True)
+        os.makedirs(os.path.join(self.out, 'runs'), exist_ok=True)
+
+    def new_session(self, ctx, follower, run_dir):
+        if ctx.sess is not None:
+            ctx.sess.close()
+        ctx.sess = Session(self.args.binary, run_dir, follower)
+        ctx.serials, ctx.cols, ctx.gone = [], [], []
+        idx = ctx.sess.wire.read_path('/index')
+        ctx.serials = [int(line.split()[0]) for line in idx.decode(errors='replace').splitlines() if line]
+
+    def execute(self, ops, run_dir, ctx=None, log=None, stop_on_failure=True, verbose=False):
+        """Run concrete ops (a replay). Returns the first Failure or None."""
+        ctx = ctx or Ctx(0)
+        ctx.verbose = verbose
+        failure = None
+        try:
+            for i, op in enumerate(ops):
+                ctx.step = i
+                if op['op'] == 'fresh':
+                    self.new_session(ctx, op.get('follower', False), run_dir)
+                    continue
+                if ctx.sess is None:
+                    self.new_session(ctx, False, run_dir)
+                try:
+                    run_step(ctx, op)
+                except Ended:
+                    ctx.sess.close()
+                    ctx.sess = None
+                except Failure as f:
+                    failure = f
+                    if verbose:
+                        print(f'step {i}: {f}')
+                    if stop_on_failure:
+                        break
+        finally:
+            if ctx.sess is not None:
+                ctx.sess.close(keep=False)
+                ctx.sess = None
+        return failure
+
+    def fuzz(self, seed, steps):
+        name = f'seed-{seed}'
+        run_dir = os.path.join(self.out, 'runs', name)
+        shutil.rmtree(run_dir, ignore_errors=True)
+        os.makedirs(run_dir)
+        ctx = Ctx(seed)
+        rng = ctx.rng
+        log_path = os.path.join(run_dir, 'ops.jsonl')
+        segment = []
+        found = collections.OrderedDict()
+        t0 = time.monotonic()
+        weights = [w for w, _ in GENERATORS]
+        ended = 0
+        with open(log_path, 'w') as log:
+            log.write(json.dumps({'seed': seed, 'steps': steps, 'binary': self.args.binary}) + '\n')
+
+            def fresh():
+                op = {'op': 'fresh', 'follower': rng.random() < 0.5}
+                log.write(json.dumps(op) + '\n')
+                segment.clear()
+                segment.append(op)
+                self.new_session(ctx, op['follower'], run_dir)
+
+            fresh()
+            for step in range(steps):
+                ctx.step = step
+                gen = rng.choices(GENERATORS, weights)[0][1]
+                op = gen(ctx, rng)
+                log.write(json.dumps(op) + '\n')
+                log.flush()
+                segment.append(op)
+                f = None
+                try:
+                    run_step(ctx, op)
+                except Ended:
+                    ended += 1
+                    fresh()
+                    continue
+                except Failure as failure:
+                    f = failure
+                except Exception as why:  # a harness bug: say so, carry on
+                    import traceback
+                    f = Failure('harness', f'{type(why).__name__}: {why} ' +
+                                traceback.format_exc()[-1500:], step)
+                if f is not None:
+                    sig = f.signature()
+                    first = sig not in found
+                    found.setdefault(sig, []).append(step)
+                    print(f'[seed {seed} step {step}] FAIL {f.kind}: {f.why[:400]}', flush=True)
+                    if first:
+                        crash = ctx.sess.crash_text()
+                        ctx.sess.close(keep=True)
+                        ctx.sess = None
+                        self.record(seed, step, f, list(segment), crash)
+                    fresh()
+                if self.args.progress and step and step % self.args.progress == 0:
+                    rate = step / (time.monotonic() - t0)
+                    print(f'[seed {seed}] step {step}/{steps} {rate:.0f} steps/s, {len(found)} distinct failure(s)',
+                          flush=True)
+        if ctx.sess is not None:
+            ctx.sess.close()
+        dt = time.monotonic() - t0
+        print(f'[seed {seed}] {steps} steps in {dt:.0f}s ({steps / dt:.0f}/s), {ended} clean quit(s); '
+              f'ops log {log_path}')
+        for sig, at in found.items():
+            print(f'  {len(at)}x {sig[:200]}  (first at step {at[0]})')
+        return found
+
+    def record(self, seed, step, f, ops, crash):
+        digest = hashlib.sha1(f.signature().encode()).hexdigest()[:8]
+        path = os.path.join(self.out, 'bugs', f'{f.kind}-{digest}-seed{seed}-step{step}.jsonl')
+        header = {'seed': seed, 'step': step, 'kind': f.kind, 'why': f.why, 'signature': f.signature(),
+                  'crash': crash[-4000:], 'binary': self.args.binary}
+        write_ops(path, header, ops)
+        print(f'  bug record {path}', flush=True)
+        if not self.args.no_shrink:
+            self.shrink(path)
+
+    def shrink(self, path):
+        header, ops = read_ops(path)
+        # The first replay pins the signature (this code's, which may have
+        # changed since the record was written); later ones must match it.
+        sig = None
+        kind = header.get('kind')
+        run_dir = os.path.join(self.out, 'runs', f'shrink-{os.getpid()}')
+        os.makedirs(run_dir, exist_ok=True)
+        follower = ops[0].get('follower', False) if ops and ops[0]['op'] == 'fresh' else False
+        body = [op for op in ops if op['op'] != 'fresh']
+        budget = [self.args.shrink_budget]
+        deadline = time.monotonic() + self.args.shrink_seconds
+
+        def fails(candidate):
+            if budget[0] <= 0 or time.monotonic() > deadline:
+                return False
+            budget[0] -= 1
+            nonlocal sig
+            f = self.execute([{'op': 'fresh', 'follower': follower}] + candidate, run_dir)
+            if f is None:
+                return False
+            if sig is None:
+                if kind is not None and f.kind != kind:
+                    return False
+                sig = f.signature()
+                return True
+            return f.signature() == sig
+
+        if not fails(body):
+            # One more try with the other log mode, then give up.
+            follower = not follower
+            if not fails(body):
+                header['shrink'] = 'did not reproduce on replay'
+                write_ops(path, header, ops)
+                print(f'  not reproduced on replay: {path}', flush=True)
+                return
+        small = ddmin(body, fails)
+        # Then shrink each op's payload.
+        small = shrink_payloads(small, fails)
+        header['shrink'] = f'{len(body)} -> {len(small)} ops'
+        out = path.replace('.jsonl', '.min.jsonl')
+        write_ops(out, header, [{'op': 'fresh', 'follower': follower}] + small)
+        print(f'  shrunk {len(body)} -> {len(small)} ops: {out}', flush=True)
+
+
+def ddmin(items, fails):
+    """Zeller's delta debugging: a 1-minimal failing subsequence."""
+    n = 2
+    # Trailing ops after the failure never matter; the failing op is last.
+    while len(items) >= 2:
+        size = max(1, len(items) // n)
+        chunks = [items[i:i + size] for i in range(0, len(items), size)]
+        reduced = False
+        for i in range(len(chunks)):
+            complement = [op for j, c in enumerate(chunks) if j != i for op in c]
+            if complement and fails(complement):
+                items, n, reduced = complement, max(n - 1, 2), True
+                break
+        if not reduced:
+            if n >= len(items):
+                break
+            n = min(len(items), n * 2)
+    return items
+
+
+def shrink_payloads(ops, fails):
+    """Halve long data payloads while the failure stays."""
+    ops = [dict(op) for op in ops]
+    for i, op in enumerate(ops):
+        while 'data' in op and len(op['data']) > 8:
+            half = dict(op, data=op['data'][:len(op['data']) // 2])
+            trial = ops[:i] + [half] + ops[i + 1:]
+            if fails(trial):
+                ops[i] = op = half
+            else:
+                break
+        if op.get('chunks') not in (None, 'whole'):
+            whole = dict(op, chunks='whole')
+            if fails(ops[:i] + [whole] + ops[i + 1:]):
+                ops[i] = whole
+    return ops
+
+
+def write_ops(path, header, ops):
+    with open(path, 'w') as f:
+        f.write(json.dumps(header) + '\n')
+        for op in ops:
+            f.write(json.dumps(op) + '\n')
+
+
+def read_ops(path):
+    with open(path) as f:
+        lines = [json.loads(line) for line in f if line.strip()]
+    return lines[0], lines[1:]
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
+    ap.add_argument('binary')
+    ap.add_argument('--seed', type=int, action='append', help='seed (repeatable); default 1')
+    ap.add_argument('--steps', type=int, default=1000)
+    ap.add_argument('--replay', help='run an ops log (or bug record) exactly')
+    ap.add_argument('--shrink', help='shrink a bug record by delta debugging over replays')
+    ap.add_argument('--no-shrink', action='store_true', help='record failures without shrinking them')
+    ap.add_argument('--shrink-budget', type=int, default=300, help='replays a shrink may spend')
+    ap.add_argument('--shrink-seconds', type=float, default=600)
+    ap.add_argument('--smoke', action='store_true', help='fixed seed, few steps, fail on any failure')
+    ap.add_argument('--progress', type=int, default=500)
+    ap.add_argument('--verbose', '-v', action='store_true', help='replay: print each request and log record')
+    ap.add_argument('--out', default=DEFAULT_OUT)
+    args = ap.parse_args()
+    args.binary = os.path.abspath(args.binary)
+    for key in [k for k in os.environ if k.startswith('PARDES_') or k in ('NINE_MOUNT', 'NAMESPACE')]:
+        del os.environ[key]
+    become_subreaper()
+    signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))
+    runner = Runner(args)
+    try:
+        return run_main(runner, args)
+    finally:
+        sweep(f'{os.getpid()}-')
+        for d in RUNTIME_DIRS:
+            shutil.rmtree(d, ignore_errors=True)
+
+
+def run_main(runner, args):
+    if args.replay:
+        header, ops = read_ops(args.replay)
+        run_dir = os.path.join(args.out, 'runs', f'replay-{os.getpid()}')
+        os.makedirs(run_dir, exist_ok=True)
+        f = runner.execute(ops, run_dir, verbose=args.verbose)
+        print(f'replay {args.replay}: ' + (f'FAIL at op {f.step}: {f.kind}: {f.why}' if f else 'passed'))
+        return 1 if f else 0
+    if args.shrink:
+        runner.shrink(args.shrink)
+        return 0
+    if args.smoke:
+        args.no_shrink = True
+        found = runner.fuzz(20260929, args.steps if args.steps != 1000 else 800)
+        return 1 if found else 0
+    total = 0
+    for seed in args.seed or [1]:
+        total += len(runner.fuzz(seed, args.steps))
+    return 1 if total else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())
-- 
cgit v1.3