path: root/test/monkey9p.py

#!/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)
    sizes = chunking(op.get('chunks', 'whole'), len(data), w.msize - 23)
    opened = ok
    sent = data
    if ok:
        offset = 0
        took = []
        for size in sizes:
            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 ok:
                took.append(size)
            if not ok and op.get('stop_on_error', True):
                break
        # Misses are allowed the Twrites that succeeded: a line written a
        # byte a Twrite is a look a byte, each its own write.
        allowances(res, path, data[:sum(took)], took, w.msize - 24)
    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 served_lines(data, sizes, room):
    """The lines a lines file runs for `data` written in pieces of `sizes`
    (fs.md, tree.zig write): each ended line, and the unended tail of a piece
    shorter than a Twrite holds (`room`, msize less 24), which is whole in
    itself, as acme takes each write; a piece that fills its Twrite may go
    on, and its tail waits for the next. A line written a byte a Twrite is
    so as many lines as bytes."""
    lines, pending, at = [], b'', 0
    for size in sizes:
        piece = data[at:at + size]
        at += size
        pending += piece
        *done, pending = pending.split(b'\n')
        lines += done
        if pending and len(piece) < room:
            lines.append(pending)
            pending = b''
    if pending:
        lines.append(pending)
    return lines


def allowances(res, path, data, sizes, room):
    """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 served_lines(data, sizes, room) 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 check_served_lines():
    assert served_lines(b'a\nb\n', [4], 8000) == [b'a', b'b']
    assert served_lines(b'ab\n', [1, 1, 1], 8000) == [b'a', b'b', b'']
    assert served_lines(b'abcd', [2, 2], 2) == [b'abcd']
    assert served_lines(b'ab\ncd', [3, 2], 8000) == [b'ab', b'cd']


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
    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:
        allowances(res, path, data, [len(data)], sess.wire.msize - 24)
    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()
    # The rule is a write's: a refused open (pane/new's and an OTRUNC one
    # too), create or remove logs no err, nor does a write to pane/new
    # (`permission denied`: it is only read).
    failed = [w for w in writes if w[2] is False and w[0] in ('write', 'clunk-write')
              and w[1].rstrip('/') != '/pane/new']
    errs = occurrences(res.window, 'err')
    msgs = occurrences(res.window, 'msg')
    # Allowed only the Twrites that succeeded (allowances): a failed one
    # logs its one err.
    allowed = sum(r[3] for r in res.requests if r[0] == 'allow')
    # 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']
    lo, hi = len(failed), len(failed) + allowed + len(closes)
    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()
    check_served_lines()
    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())