#!/usr/bin/env python3
import contextlib
import fcntl
import json
import os
import shutil
import select
import signal
import socket
from pathlib import Path
import subprocess
import struct
import sys
import tempfile
import termios
import threading
import time

from ninep import Client


@contextlib.contextmanager
def session(binary, root, name, *options, socket_name=None, file=None, tty=False, network=None,
            launch=None, inherited=None, terminal=None):
    env = {key: value for key, value in os.environ.items()
           if not key.startswith('PARDES_')}
    env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root),
               XDG_CONFIG_HOME=str(root / 'config'), PARDES_NOTIME='1',
               PARDES_DUMP=str(root / (name + '.dump.zon')))
    env.update(inherited or {})
    for language in ['RS', 'C', 'GO', 'TS', 'PY']:
        env['PARDES_LSP_' + language] = ''
    fixture = root / (name + '.txt')
    fixture.write_bytes(b'initial\n')
    address = root / ('pardes-9p-' + (socket_name or name) + '.sock')
    child = None
    master = slave = None
    reader = None
    try:
        args = [binary, '--detach=' + name, *options, str(file or fixture)]
        preexec = None
        if tty:
            master, slave = os.openpty()
            fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 24, 80, 0, 0))
            env['TERM'] = 'xterm-256color'
            args = [binary, '--tty', '--9p=' + (socket_name or name), *options, str(file or fixture)]
            preexec = lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0)
        if launch is not None:
            args = [binary, *launch]
        child = subprocess.Popen(args, cwd=root, env=env, stdin=slave,
                                 stdout=slave if tty else subprocess.DEVNULL,
                                 stderr=subprocess.PIPE, start_new_session=True,
                                 preexec_fn=preexec)
        if terminal is not None:
            # Borrowed handles for input-driving probes, valid inside this context.
            terminal.update(pid=child.pid, input_fd=master)
        if launch is not None and socket_name is None:
            address = root / f'pardes-9p-{child.pid}.sock'
        if slave is not None:
            os.close(slave)
            slave = None
            def drain():
                try:
                    while os.read(master, 65536):
                        pass
                except OSError:
                    pass
            reader = threading.Thread(target=drain, daemon=True)
            reader.start()
        deadline = time.monotonic() + 10
        while not address.exists():
            if child.poll() is not None:
                raise AssertionError((args, child.returncode, child.stderr.read().decode()))
            if time.monotonic() > deadline:
                raise TimeoutError('default 9P socket did not appear')
            time.sleep(.005)
        with Client(address) as client:
            if network is not None:
                for dial in client.read('/listeners').decode().splitlines():
                    parts = dial.split('!')
                    if parts[0] in ['tcp', 'quic']:
                        assert len(parts) == 3 and 0 < int(parts[2]) < 65536, dial
                        network[parts[0].upper()] = int(parts[2])
                assert set(network) == {'TCP', 'QUIC'}, network
            yield client, address
    finally:
        if terminal is not None:
            terminal.clear()
        if child is not None:
            if child.poll() is None:
                child.terminate()
            try:
                child.wait(timeout=5)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
            child.stderr.close()
        if slave is not None:
            os.close(slave)
        if master is not None:
            os.close(master)
        if reader is not None:
            reader.join(timeout=1)
            assert not reader.is_alive(), 'terminal reader did not stop'


def newest(client):
    """The newest pane's serial: /index is ordered by serial."""
    return int(client.read('/index').splitlines()[-1].split()[0])


def new_pane(client, contents):
    """An open of /pane/new makes a pane; the read names it."""
    serial = int(client.read('/pane/new'))
    if contents:
        client.write(f'/pane/{serial}/body', contents)
    return serial


def look(client, path, source=None):
    """A right click on `path` from a pane (a fresh scratch by default)."""
    serial = source if source is not None else new_pane(client, b'')
    client.write(f'/pane/{serial}/look', f'{path}\n'.encode())
    results = [int(row) for row in client.read('/look').split()]
    assert len(results) == 1 and results[0] != serial, (path, results)
    # The scratch it looked from goes: a column holds only so many panes,
    # each its tag and two rows.
    if source is None:
        client.remove(f'/pane/{serial}')
    return results[0]


def execute(client, serial, command):
    """A middle click on `command` from a pane."""
    client.write(f'/pane/{serial}/exec', f'{command}\n'.encode())


def execute_refused(client, serial, command, words):
    """A middle click on a builtin that fails: the write fails, saying why."""
    try:
        execute(client, serial, command)
    except OSError as why:
        assert words in str(why), (command, why)
        return
    raise AssertionError(f'{command!r} was taken')


def walk_tree(client, path='/', skip=('/log', '/os', '/screen')):
    """Stat every entry below `path`, as `find` does, without opening anything."""
    seen = []
    for name in client.list(path):
        child = (path.rstrip('/') + '/' + name)
        info = client.stat(child)
        seen.append((child, info))
        if info['dir'] and child not in skip:
            seen.extend(walk_tree(client, child, skip))
    return seen


def discovery(binary, embedded=False):
    with tempfile.TemporaryDirectory(prefix='pardes-discovery-') as directory:
        root = Path(directory)
        with session(binary, root, 'discovery') as (client, address):
            top = client.list('/')
            # A set: which files the root holds, not the order a listing gives.
            assert set(top) - {'src', 'shaders'} == {'README', 'index', 'status', 'focus', 'ctl', 'commands', 'recent',
                                                     'look', 'exec', 'pager', 'log', 'screen', 'listeners', 'layout', 'tag', 'tagexec',
                                                     'pane', 'col', 'os'}, top
            assert 'self' not in top and 'cons' not in top and 'new' not in top, top
            assert ('src' in top) == embedded, (top, embedded)
            before = client.read('/index')
            guide = client.read('/README')
            assert guide.count(b'\n') <= 45 and b'Msg hello' in guide and b'pane/new' in guide, guide
            assert client.stat('/README')['length'] == len(guide)
            # ls/stat/find over the whole tree, without opening, creates nothing.
            seen = walk_tree(client)
            names = {path for path, _ in seen}
            assert {'/index', '/status', '/look', '/exec', '/log', '/pane/1', '/pane/1/body',
                    '/pane/1/name', '/pane/1/sel', '/pane/1/ctl', '/pane/1/tag', '/pane/1/event',
                    '/pane/1/dot', '/pane/1/limit', '/pane/1/dirty', '/pane/1/mark',
                    '/pane/1/scroll', '/pane/1/look', '/pane/1/exec'} <= names, sorted(names)
            assert '/pane/1/pty' not in names, 'the fixture file pane grew a pty/'
            assert not any(name.startswith('/self') or name.endswith('/rdsel') or name.endswith('/wrsel')
                           for name in names), sorted(names)
            for path, info in seen:
                assert info['mode'] in (0o755, 0o644, 0o666, 0o444, 0o222, 0o555), (path, info)
                assert info['mtime'] > 0, (path, info)
            assert client.read('/index') == before, 'browsing created a pane'
            assert client.stat('/index')['length'] == len(before) > 0
            index = client.stat('/index')
            assert index['mode'] == 0o444 and client.stat('/status')['mode'] == 0o444
            assert client.stat('/exec')['mode'] == 0o666
            assert client.stat('/pane/1/body')['mode'] == 0o666 and client.stat('/pane')['mode'] == 0o755
            # A range file reads the pair it takes; a flag file reads 0 or 1.
            assert client.stat('/pane/1/addr')['length'] == 24
            assert client.stat('/pane/1/limit')['length'] == 0
            assert client.read('/pane/1/dirty') == b'0\n'
            assert client.read('/pane/1/scroll') == b'1\n'
            fixture = int(before.split()[0])
            assert before.split()[1:3] == [b'text', b'0'] and before.rsplit(b' ', 1)[0].endswith(b'/discovery.txt'), before
            # /status reports the editor; /exec has nothing to report yet.
            status = dict(line.split(maxsplit=1) for line in client.read('/status').decode().splitlines())
            assert int(status['pid']) > 0 and status['version'] and int(status['panes']) >= 1, status
            assert client.read('/exec') == b''
            # An open of /pane/new makes a pane and the read names it; each
            # open makes another. A stat makes none, which is why new can be
            # listed at all: ls -l stats every name a listing gave it.
            # /log answers the recent past and ends; `follow` written to the
            # same open makes its reads past that wait for new records.
            log = client.open('/log', 2)
            frozen = len(client.read_fid(log))
            client.rpc(118, struct.pack('<IQI', log, 0, 7) + b'follow\n')
            first = int(client.read('/pane/new'))
            assert first == newest(client)
            second = int(client.read('/pane/new'))
            assert first != second and first != fixture, (first, second)
            assert 'new' in client.list('/pane'), 'new should be visible to ls'
            client.stat('/pane/new')
            assert newest(client) == second, 'a stat of new made a pane'
            # What the editor said is in the stream too.
            def pane_event():
                while (record := client.read_fid(log, frozen)).startswith((b'msg ', b'err ')):
                    pass
                return record
            assert pane_event() == f'new {first} {root}/+New\n'.encode()
            assert pane_event() == f'new {second} {root}/+New\n'.encode()
            # A follower that is waiting wakes when the editor logs, not when
            # some unrelated request next happens to wait.
            with Client(address) as follower:
                waiting = follower.open('/log', 2)
                start = len(follower.read_fid(waiting))
                follower.rpc(118, struct.pack('<IQI', waiting, 0, 7) + b'follow\n')
                woke = []
                reader = threading.Thread(target=lambda: woke.append(follower.read_fid(waiting, start)), daemon=True)
                reader.start()
                time.sleep(.2)
                client.write('/exec', b'Msg woken\n')
                reader.join(5)
                assert woke and woke[0].endswith(b' woken\n'), woke
            # A pane's ctl takes acme's lock: a second open's lock fails at
            # once with "file in use" until the holder unlocks, and nothing
            # else waits on it meanwhile.
            with Client(address) as other:
                mine = client.open(f'/pane/{first}/ctl', 2)
                client.rpc(118, struct.pack('<IQI', mine, 0, 5) + b'lock\n')
                theirs = other.open(f'/pane/{first}/ctl', 1)
                try:
                    other.rpc(118, struct.pack('<IQI', theirs, 0, 5) + b'lock\n')
                    raise AssertionError('a second lock was taken')
                except OSError as refused:
                    assert 'file in use' in str(refused), refused
                client.write(f'/pane/{first}/addr', b'#0')
                client.rpc(118, struct.pack('<IQI', mine, 0, 7) + b'unlock\n')
                other.rpc(118, struct.pack('<IQI', theirs, 0, 5) + b'lock\n')
                client.close(mine)
                other.close(theirs)
            # /focus names the pane with the keyboard and moves it, as rio's
            # wctl `current` does; a pane's ctl line ends saying which it is.
            client.write('/focus', f'{first}\n'.encode())
            assert client.read('/focus') == f'{first}\n'.encode()
            assert client.read(f'/pane/{first}/ctl').endswith(b' current ')
            assert client.read(f'/pane/{second}/ctl').endswith(b' notcurrent ')
            try:
                client.write('/focus', b'99999\n')
                raise AssertionError('focus took a serial no pane has')
            except OSError as refused:
                assert 'no such pane' in str(refused), refused
            # /ctl reads the settings as a write takes them and takes the
            # session's builtins; a pane's ctl takes the pane's.
            settings = client.read('/ctl')
            assert b'Verbose ' in settings and b'Platform' not in settings, settings
            client.write('/ctl', settings)
            assert client.read('/ctl') == settings
            client.write('/ctl', b'Verbose off\n')
            assert b'Verbose off\n' in client.read('/ctl')
            for path, line, why in (('/ctl', b'Del\n', 'not a session control message "Del"'),
                                    ('/ctl', b'Nope 3\n', 'unknown control message "Nope 3"'),
                                    (f'/pane/{first}/ctl', b'Newcol\n', 'not a window control message "Newcol"')):
                try:
                    client.write(path, line)
                    raise AssertionError(f'{path} took {line!r}')
                except OSError as refused:
                    assert why in str(refused), refused
            commands = client.read('/commands').decode().splitlines()
            def row(start):
                return any(r == start or r.startswith(start + ' ') for r in commands)
            assert row('Newcol root') and row('Save arg pane') and row('Edit arg both'), commands
            # word, `arg` if it takes one, its ctl, a setting's values, then `-- ` what it does
            assert all(r.split()[2 if r.split()[1] == 'arg' else 1] in ('root', 'pane', 'both') for r in commands), commands
            assert row('Placement arg root acme,pardes'), commands
            assert client.stat('/commands')['length'] == len('\n'.join(commands).encode()) + 1  # bytes: a description may end in …
            client.write(f'/pane/{first}/ctl', b'Msg said through ctl\n')
            assert 'said through ctl' in ''.join(cell[0] for cell in client.screen()['cells'])
            assert set(client.list('/pane')) == {'new', str(fixture), str(first), str(second)}
            client.write(f'/pane/{first}/body', b'first pane', truncate=True)
            assert client.read(f'/pane/{first}/body') == b'first pane'
            assert client.read(f'/pane/{second}/body') == b''
            assert client.stat(f'/pane/{first}/body')['length'] == len(b'first pane')
            rows = {int(row.split()[0]): row.split() for row in client.read('/index').splitlines()}
            assert rows[first][1:3] == [b'text', b'1'] and rows[second][1:3] == [b'text', b'0'], rows
            # name: read the file name, write to rename; sel: the selection.
            assert client.read(f'/pane/{first}/name') == f'{root}/+New\n'.encode()
            client.write(f'/pane/{first}/name', b'notes.txt\n')
            assert client.read(f'/pane/{first}/name') == f'{root}/notes.txt\n'.encode()
            assert pane_event() == f'rename {first} {root}/notes.txt\n'.encode()
            assert client.stat(f'/pane/{first}/name')['length'] == len(f'{root}/notes.txt\n')
            # Copying addr onto dot selects that range; sel reads and replaces it.
            client.write(f'/pane/{first}/addr', b'#0,#5')
            client.write(f'/pane/{first}/dot', client.read(f'/pane/{first}/addr'))
            assert client.read(f'/pane/{first}/dot') == client.read(f'/pane/{first}/addr')
            assert client.read(f'/pane/{first}/sel') == b'first'
            assert client.stat(f'/pane/{first}/sel')['length'] == 5
            client.write(f'/pane/{first}/sel', b'FIRST')
            assert client.read(f'/pane/{first}/body') == b'FIRST pane'
            assert client.read(f'/pane/{first}/sel') == b''
            # exec Msg shows text; exec Save writes the named file; rmdir closes.
            client.write(f'/pane/{first}/exec', b'Msg hello from 9P\n')
            assert 'hello from 9P' in ''.join(cell[0] for cell in client.screen()['cells'])
            assert client.read(f'/pane/{first}/dirty') == b'1\n'
            client.write(f'/pane/{first}/exec', b'Save\n')
            assert (root / 'notes.txt').read_bytes() == b'FIRST pane'
            assert pane_event() == f'save {first} {root}/notes.txt\n'.encode()
            assert client.read('/exec') == f'{first}\n'.encode()
            assert client.read(f'/pane/{first}/dirty') == b'0\n'
            client.remove(f'/pane/{second}')
            assert pane_event() == f'del {second} {root}/+New\n'.encode()
            assert str(second) not in client.list('/pane')
            # look opens a file; the serial reads back.
            (root / 'opened.txt').write_bytes(b'one\ntwo\nthree\n')
            client.write('/look', f'{root}/opened.txt:2\n'.encode())
            opened = int(client.read('/look'))
            assert opened not in (fixture, first, second), opened
            assert client.read(f'/pane/{opened}/body') == b'one\ntwo\nthree\n'
            assert client.read(f'/pane/{opened}/name') == f'{root}/opened.txt\n'.encode()
            assert pane_event() == f'new {opened} {root}/opened.txt\n'.encode()
            client.write(f'/pane/{first}/look', f'{root}/opened.txt:3\n'.encode())
            assert client.read('/look') == f'{opened}\n'.encode(), 'a repeated look targets the open pane'
            assert {str(fixture), str(first), str(opened)} <= set(client.list('/pane'))
            # Writing an event record back still performs the action it names.
            client.write(f'/pane/{first}/body', b'Msg via event', truncate=True)
            client.write(f'/pane/{first}/event', b'FX0 13\n')
            assert 'via event' in ''.join(cell[0] for cell in client.screen()['cells'])
            # A command line may not carry control characters, and nothing else
            # in the tree may be created or removed.
            refused = [lambda: client.write('/exec', b'Msg a\x00b'),
                       lambda: client.write('/status', b'anything\n'),
                       lambda: client.mkdir('/', 'x'),
                       lambda: client.mkdir('/pane', 'one'),
                       lambda: client.create('/pane', 'plain-file'),
                       lambda: client.remove('/pane/1/body'),
                       lambda: client.remove('/index')]
            for index, attempt in enumerate(refused):
                try:
                    attempt()
                except OSError:
                    pass
                else:
                    raise AssertionError(f'refusal {index} was accepted')
            client.close(log)
            if embedded:
                assert client.read('/src/pardes.zig').startswith(b'const std')
                assert client.stat('/src/pardes.zig')['mode'] == 0o444
                assert b'pub const Pardes' in client.read(f'/pane/{look(client, "/virtual/src/pardes.zig")}/body')
            # An open of data with OTRUNC (a shell's >) deletes the addressed
            # range only, so the write after it replaces that range.
            scratch = new_pane(client, b'hello world\nsecond line\n')
            client.write(f'/pane/{scratch}/addr', b'#0,#5')
            client.write(f'/pane/{scratch}/data', b'HOWDY', truncate=True)
            assert client.read(f'/pane/{scratch}/body') == b'HOWDY world\nsecond line\n'
            # sam's regexps: ^ at any line; a miss says so and leaves no
            # address, so the data write after it refuses.
            client.write(f'/pane/{scratch}/addr', b'/^second/')
            assert client.read(f'/pane/{scratch}/addr') == b'         12          18 '
            for path, data, why in ((f'/pane/{scratch}/addr', b'/nowhere/', 'no match for regexp'),
                                    (f'/pane/{scratch}/data', b'LOST', 'no address')):
                try:
                    client.write(path, data)
                    raise AssertionError(f'{path} took {data!r}')
                except OSError as refused:
                    assert why in str(refused), refused
            assert client.read(f'/pane/{scratch}/body') == b'HOWDY world\nsecond line\n'
            # Through a mount both read as EINVAL; the log says which and why.
            assert f'err {scratch} addr: no match for regexp\n'.encode() in client.read('/log')
            client.remove(f'/pane/{scratch}')
            # Real `diff -u` in a command pane: the terminal turns the tab
            # before each timestamp into blanks, and a look at a changed line
            # still opens the new file at it.
            (root / 'diff-old.txt').write_text('one\ntwo\nthree\n')
            (root / 'diff-new.txt').write_text('one\nTWO\nthree\n')
            seen = client.read('/log').decode().count('\nexit ')
            client.write('/exec', b'diff -u diff-old.txt diff-new.txt\n')
            deadline = time.monotonic() + 10
            while client.read('/log').decode().count('\nexit ') <= seen:
                assert time.monotonic() < deadline, 'diff -u never ended'
                time.sleep(.05)
            # A command's pane is kind cmd, not term: no shell to type into.
            ran = [row.split()[0] for row in client.read('/index').decode().splitlines() if row.split()[1] == 'cmd']
            shown = client.read(f'/pane/{ran[-1]}/body').decode()
            assert '+++ diff-new.txt' in shown and '\t' not in shown, shown
            client.write(f'/pane/{ran[-1]}/look', b'+TWO\n')
            opened = [row for row in client.read('/index').decode().splitlines() if row.split()[3] == f'{root}/diff-new.txt']
            assert len(opened) == 1, client.read('/index')
            client.remove(f'/pane/{opened[0].split()[0]}')
            client.remove(f'/pane/{ran[-1]}')
            # A Save makes the directories its file goes in, whatever names
            # it: the pane's own name or Save's path.
            # `~` is home (this session's HOME, the test's root) wherever a
            # path is typed: Save, name, a look, with an address or not.
            tilde = new_pane(client, b'one\ntwo\nthree\n')
            client.write(f'/pane/{tilde}/ctl', b'Save ~/tilde-made/sub/x.txt\n')
            assert (root / 'tilde-made/sub/x.txt').read_bytes() == b'one\ntwo\nthree\n'
            client.write(f'/pane/{tilde}/name', b'~/tilde-named.txt\n')
            assert client.read(f'/pane/{tilde}/name') == f'{root}/tilde-named.txt\n'.encode()
            client.remove(f'/pane/{tilde}')
            client.write('/look', b'~/tilde-made/sub/x.txt:3\n')
            looked = [row for row in client.read('/index').decode().splitlines() if row.split()[3] == f'{root}/tilde-made/sub/x.txt']
            assert len(looked) == 1, client.read('/index')
            assert client.read(f'/pane/{looked[0].split()[0]}/addr').startswith(b'8 '), client.read(f'/pane/{looked[0].split()[0]}/addr')
            client.remove(f'/pane/{looked[0].split()[0]}')
            made_dirs = new_pane(client, b'deep\n')
            client.write(f'/pane/{made_dirs}/ctl', f'Save {root}/save-made/a/b/x.txt\n'.encode())
            assert (root / 'save-made/a/b/x.txt').read_bytes() == b'deep\n'
            client.write(f'/pane/{made_dirs}/name', f'{root}/save-named/c/y.txt\n'.encode())
            client.write(f'/pane/{made_dirs}/ctl', b'Save\n')
            assert (root / 'save-named/c/y.txt').read_bytes() == b'deep\n'
            client.remove(f'/pane/{made_dirs}')
            # A save the host cannot do (a directory it may not make) fails
            # the write that asked for it, and the log's err record names
            # the path and why.
            unsaved = new_pane(client, b'unsaved\n')
            try:
                client.write(f'/pane/{unsaved}/ctl', b'Save /nonexistent-pardes-dir/x.txt\n')
                raise AssertionError('a Save into a missing directory succeeded')
            except OSError as refused:
                assert 'Save /nonexistent-pardes-dir/x.txt: no such directory' in str(refused), refused
            assert f'err {unsaved} ctl: Save /nonexistent-pardes-dir/x.txt: '.encode() in client.read('/log'), client.read('/log')
            client.remove(f'/pane/{unsaved}')

            # Sixteen connections fill the slots; the next is told why.
            extra = [Client(address) for _ in range(15)]
            try:
                try:
                    Client(address).__exit__()
                    raise AssertionError('a seventeenth connection was served')
                except OSError as refused:
                    assert 'too many connections' in str(refused), refused
            finally:
                for other in extra:
                    other.__exit__()
            deadline = time.monotonic() + 5
            while b'err - 9p: too many connections' not in client.read('/log'):
                assert time.monotonic() < deadline, client.read('/log')
                time.sleep(.01)

    print('9P discovery: listing/stat/find are inert; new, remove, look, exec, name, sel, log, ctl lock, focus, the ctl split and commands behave')


def run_file(binary):
    """pty/run: a line in, how it ended out, on the same open."""
    def run(c, serial, line):
        fid = c.open(f'/pane/{serial}/pty/run', 2)
        try:
            c.rpc(118, struct.pack('<IQI', fid, 0, len(line)) + line)
            return c.read_fid(fid)
        finally:
            c.close(fid)
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)
        with session(binary, root, 'run') as (client, address):
            client.write('/exec', b'Shell bash\n')
            client.write('/exec', b'Tty\n')
            term = int(max(r.split()[0] for r in client.read('/index').splitlines() if r.split()[1] == b'term'))
            deadline = time.monotonic() + 10
            while run(client, term, b'true\n') == b'busy\n':  # until its first prompt
                assert time.monotonic() < deadline, 'the shell never reached a prompt'
                time.sleep(.05)
            assert run(client, term, b'false\n') == b'exit 1\n'
            assert run(client, term, b'printf "one\\ntwo\\n"\n') == b'exit 0\none\ntwo\n'  # then what it printed
            assert run(client, term, b'sh -c "exit 7"\n') == b'exit 7\n'
            assert run(client, term, b'fi\n') == b'error not run\n'  # a syntax error runs nothing
            assert run(client, term, b"echo 'open\n") == b'error not run\n'  # nor does an open quote
            assert run(client, term, b'true\n') == b'exit 0\n'  # ...and the shell is clear after
            # With no frontend attached the pty's size is in pixels too, at a
            # nominal 8x16 cell: a program sizes its kitty graphics by it.
            said = run(client, term, b'python3 -c \'import fcntl,termios,struct;print(*struct.unpack("4H",fcntl.ioctl(0,termios.TIOCGWINSZ,bytes(8))))\'\n')
            rows, cols, xpixel, ypixel = map(int, said.split(b'\n')[1].split())
            assert (xpixel, ypixel) == (cols * 8, rows * 16) and xpixel > 0, said
            # A paging command run through pty/run does not wait in less: the
            # terminal's PAGER is `pardes -`, and its text is a +Pager pane.
            if shutil.which('git'):
                repo = root / 'pagedrepo'
                repo.mkdir()
                git = ['git', '-C', str(repo), '-c', 'user.name=p', '-c', 'user.email=p@x']
                subprocess.run(git + ['init', '-q'], check=True, env=dict(os.environ, HOME=str(root)))
                (repo / 'f').write_text('x\n')
                subprocess.run(git + ['add', 'f'], check=True, env=dict(os.environ, HOME=str(root)))
                subprocess.run(git + ['commit', '-qm', 'paged-commit-subject'], check=True, env=dict(os.environ, HOME=str(root)))
                assert run(client, term, b'git -C pagedrepo log\n').startswith(b'exit 0\n')
                pager = [r.split()[0].decode() for r in client.read('/index').splitlines() if r.split()[3:4] == [f'{repo}/+Pager'.encode()]]  # git -C runs it in the repo
                assert len(pager) == 1, client.read('/index')
                assert b'paged-commit-subject' in client.read(f'/pane/{pager[0]}/body')
                assert b'\x1b' not in client.read(f'/pane/{pager[0]}/body')
                client.remove(f'/pane/{pager[0]}')
            # systemd's tools page through their own variable: set too, its
            # secure mode off, the path unquoted where it needs none.
            systemd_pager = run(client, term, b'printenv SYSTEMD_PAGER SYSTEMD_PAGERSECURE\n')
            assert systemd_pager.startswith(b'exit 0\n') and systemd_pager.rstrip().endswith(b' -\n0'), systemd_pager
            if shutil.which('journalctl'):
                assert run(client, term, b'journalctl -n 3 --no-hostname\n').startswith(b'exit ')
                for row in client.read('/index').splitlines():
                    if row.split()[3:4] == [f'{root}/+Pager'.encode()]:
                        client.remove(f'/pane/{row.split()[0].decode()}')
            # A pty is not given one row (it loses the prompt's input mark,
            # and the shell would read busy for ever): refused, and said.
            try:
                client.write(f'/pane/{term}/pty/ctl', b'winsize 20 1\n')
                raise AssertionError('winsize 20 1 was taken')
            except OSError as refused:
                assert 'at least 2 rows' in str(refused), refused
            client.write(f'/pane/{term}/pty/ctl', b'winsize 20 2\n')
            time.sleep(.5)
            assert run(client, term, b'true\n') == b'exit 0\n'
            assert client.read(f'/pane/{term}/pty/ctl').split()[2] == b'2', client.read(f'/pane/{term}/pty/ctl')
            client.write(f'/pane/{term}/pty/ctl', b'winsize 80 24\n')
            # A shell started again that cannot start (its directory is
            # there but cannot be entered) fails the write and leaves the
            # running shell be: runs still work.
            locked = root / 'locked'
            locked.mkdir()
            assert run(client, term, f'cd {locked}\n'.encode()) == b'exit 0\n'
            deadline = time.monotonic() + 5
            while str(locked).encode() not in client.read('/index'):
                assert time.monotonic() < deadline, client.read('/index')
                time.sleep(.05)
            locked.chmod(0o600)
            try:
                client.write(f'/pane/{term}/pty/ctl', b'exec\n')
                raise AssertionError('pty/ctl exec of a shell that cannot start was taken')
            except OSError as refused:
                assert 'shell' in str(refused), refused
            finally:
                locked.chmod(0o755)
            kept = run(client, term, b'true\n')
            assert kept == b'exit 0\n', (kept, client.read('/log')[-600:])
            assert run(client, term, f'cd {root}\n'.encode()) == b'exit 0\n'
            with Client(address) as other:
                slow = []
                waiter = threading.Thread(target=lambda: slow.append(run(other, term, b'sleep 1\n')), daemon=True)
                waiter.start()
                time.sleep(.3)
                # The job the other client ran holds the tty, and says so.
                assert run(client, term, b'true\n') == b'busy: sleep is running\n'
                waiter.join(5)
                assert slow == [b'exit 0\n'], slow
                closing = []
                waiter = threading.Thread(target=lambda: closing.append(run(other, term, b'sleep 30\n')), daemon=True)
                waiter.start()
                time.sleep(.3)
                client.write(f'/pane/{term}/exec', b'Del\n')
                waiter.join(5)
                assert closing == [b'error shell gone\n'], closing


def exit_answers(binary):
    """Exit written to /ctl is answered before the editor quits."""
    with tempfile.TemporaryDirectory(prefix='pardes-exit-') as directory:
        seen = {}
        with session(binary, Path(directory), 'exit', terminal=seen) as (client, _):
            client.write('/ctl', b'Exit\n')
            deadline = time.monotonic() + 5
            while time.monotonic() < deadline:
                try:
                    # a zombie has quit; the session reaps it
                    if Path(f'/proc/{seen["pid"]}/stat').read_text().split(') ')[1][0] == 'Z':
                        break
                except FileNotFoundError:
                    break
                time.sleep(.01)
            else:
                raise AssertionError("Exit did not quit the editor")


def unattended_messages_expire(binary):
    """With nothing attached, a message leaves after MessageLinger on its own
    clock: no key, no click, no other request comes to move it along."""
    with tempfile.TemporaryDirectory(prefix='pardes-msg-') as directory:
        with session(binary, Path(directory), 'msg') as (client, _):
            client.write('/pane/1/ctl', b'Msg lingering-words\n')
            def shown():
                screen = json.loads(client.read('/screen'))
                return 'lingering-words' in ''.join(cell[0] for cell in screen['cells'])
            assert shown()
            # Reading /screen is itself a request, so wait without asking.
            time.sleep(2.5)
            assert not shown(), 'an unattended message outlived MessageLinger'


def new_terminals_named_once(binary):
    """A terminal started outside the session's directory is named by that
    directory from its `new` on: no rename to the session's and back."""
    with tempfile.TemporaryDirectory(prefix='pardes-names-') as directory:
        root = Path(directory)
        (root / 'sub').mkdir()
        with session(binary, root, 'names') as (client, _):
            made = [look(client, str(root / 'sub'))]
            before = {int(row.split()[0]) for row in client.read('/index').splitlines()}
            client.write('/pane/1/ctl', b'Tty\n')
            if shutil.which('fish'):
                client.write('/pane/1/ctl', b'Tty fish\n')
            made += [s for s in (int(row.split()[0]) for row in client.read('/index').splitlines()) if s not in before]
            time.sleep(1.5)  # their shells start and say where they are
            # A shell started again by pty/ctl exec: named as it was.
            client.write(f'/pane/{made[0]}/pty/ctl', b'exec\n')
            time.sleep(1.5)
            log = client.read('/log').decode()
            for serial in made:
                assert f'\nrename {serial} ' not in '\n' + log, (serial, log)
                assert f'\nnew {serial} {root}' in '\n' + log, (serial, log)
            # A shell that cannot start (its interpreter is not there): Tty
            # fails the write with why and leaves no pane; pty/ctl exec
            # fails with why and keeps its terminal.
            bad = root / 'badsh'
            bad.write_bytes(b'#!/nonexistent/interp\n')
            bad.chmod(0o755)
            panes = client.read('/index')
            for where, line in (('/pane/1/ctl', f'Tty {bad}\n'), ('/ctl', f'Shell {bad}\n')):
                try:
                    client.write(where, line.encode())
                    raise AssertionError(f'{line} of a shell that cannot start was taken')
                except OSError as refused:
                    assert 'interpreter /nonexistent/interp not found' in str(refused), refused
            time.sleep(.3)
            assert client.read('/index').count(b'\n') == panes.count(b'\n'), client.read('/index')
            assert b'shell: shell not found' not in client.read('/log'), client.read('/log')
            assert str(made[0]).encode() in b' '.join(r.split()[0] for r in client.read('/index').splitlines())


def recent_survives_restart(binary):
    """A file opened and closed is in /recent closed, and still is in the
    next session: the list is kept under $XDG_STATE_HOME (here HOME's
    .local/state) across restarts."""
    with tempfile.TemporaryDirectory(prefix='pardes-recent-') as directory:
        root = Path(directory)
        kept = root / 'kept.txt'
        kept.write_bytes(b'one\ntwo\n')
        with session(binary, root, 'recent1') as (client, _):
            serial = look(client, f'{kept}:2')
            assert f'open {kept}\n' in client.read('/recent').decode()
            client.write(f'/pane/{serial}/ctl', b'Del\n')
            assert f'closed {kept}\n' in client.read('/recent').decode(), client.read('/recent')
        assert (root / '.local' / 'state' / 'pardes' / 'recent').exists()
        with session(binary, root, 'recent2') as (client, _):
            listed = client.read('/recent').decode()
            assert f'closed {kept}\n' in listed, listed
            # Recent shows it, and a look at its row opens it again.
            client.write('/pane/1/exec', b'Recent\n')
            rows = int(client.read('/pane/1/exec').split()[-1])
            row = next(r for r in client.read(f'/pane/{rows}/body').decode().splitlines() if r.startswith(f'{kept}:'))
            client.write(f'/pane/{rows}/look', (row.split(' ')[0] + '\n').encode())
            assert f'open {kept}\n' in client.read('/recent').decode()


def workflow_canary(binary):
    """One session driven as the dogfood agents drive it, asserting the log
    and the read-backs at every step: the canary for the main paths (the
    round-18 Grep that never found anything went through every narrower
    test). Look files, Grep and Find hits and misses, a plain word looked
    at, Edit in two files, Save, get, pty/run failing then passing, a
    command pane and Kill, a REPL send, the layout and a column's files,
    ask/answer, and Dump/Restore last."""
    started = time.monotonic()
    with tempfile.TemporaryDirectory(prefix='pardes-flow-') as directory:
        root = Path(directory)
        (root / 'a.txt').write_bytes(b'alpha one\nshared word\n')
        (root / 'b.txt').write_bytes(b'beta two\nshared word\n')
        (root / 's.sh').write_bytes(b'echo from-file\n')

        def wait(ready, what, seconds=10):
            deadline = time.monotonic() + seconds
            while not ready():
                assert time.monotonic() < deadline, what
                time.sleep(.05)

        with session(binary, root, 'flow') as (client, address):
            def log():
                return client.read('/log').decode('utf-8', 'replace')

            def made_by(serial, line):
                """The pane an exec at `serial` made, as exec reads it back."""
                client.write(f'/pane/{serial}/exec', line)
                return int(client.read(f'/pane/{serial}/exec').split()[-1])

            # Look opens files, and the log says new.
            a = look(client, str(root / 'a.txt'))
            b = look(client, str(root / 'b.txt'))
            assert client.read(f'/pane/{a}/body') == b'alpha one\nshared word\n'
            assert f'new {a} {root}/a.txt\n' in log() and f'new {b} {root}/b.txt\n' in log()

            # Grep and Find: a hit opens its +Search, rows and all; a miss
            # fails ENOENT, opening nothing.
            grep = made_by(a, b'Grep shared\n')
            rows = client.read(f'/pane/{grep}/body').decode()
            assert 'a.txt:2' in rows and 'b.txt:2' in rows, rows
            client.remove(f'/pane/{grep}')
            execute_refused(client, a, 'Grep zzqqnomatch', 'Grep: pattern not found')
            found = made_by(a, b'Find b.txt\n')
            assert 'b.txt' in client.read(f'/pane/{found}/body').decode()
            client.remove(f'/pane/{found}')
            execute_refused(client, a, 'Find zzqqnomatch', 'Find: name not found')

            # A plain word looked at is found in the pane, after its dot.
            client.write(f'/pane/{a}/dot', b'          0           0 ')
            client.write(f'/pane/{a}/look', b'shared\n')
            assert client.read(f'/pane/{a}/dot').split() == [b'10', b'16'], client.read(f'/pane/{a}/dot')

            # Edit in two files, then Save each: on disk, and logged.
            for pane in (a, b):
                client.write(f'/pane/{pane}/ctl', b'Edit ,s/shared/SHARED/g\n')
            assert client.read(f'/pane/{b}/body') == b'beta two\nSHARED word\n'
            for pane in (a, b):
                client.write(f'/pane/{pane}/exec', b'Save\n')
            assert (root / 'a.txt').read_bytes() == b'alpha one\nSHARED word\n'
            assert f'save {a} {root}/a.txt\n' in log() and f'save {b} {root}/b.txt\n' in log()

            # get reloads what changed on disk.
            (root / 'b.txt').write_bytes(b'from disk\n')
            client.write(f'/pane/{b}/ctl', b'get\n')
            assert client.read(f'/pane/{b}/body') == b'from disk\n'

            # A terminal: pty/run failing then passing.
            term = made_by(1, b'Tty\n')

            def run(line):
                fid = client.open(f'/pane/{term}/pty/run', 2)
                try:
                    client.rpc(118, struct.pack('<IQI', fid, 0, len(line)) + line)
                    return client.read_fid(fid)
                finally:
                    client.close(fid)
            deadline = time.monotonic() + 10
            while (answer := run(b'true\n')) != b'exit 0\n':
                assert time.monotonic() < deadline, answer
                time.sleep(.05)
            assert run(b'false\n') == b'exit 1\n'
            assert run(b'echo passed\n') == b'exit 0\npassed\n'
            # (A run's answer is its record: the log has no err for these.)
            assert f'err {term} pty/run' not in log(), log()

            # A command pane, and Kill: its exit is logged.
            sleeper = made_by(a, b'sleep 30\n')
            assert f'run {sleeper} sleep 30\n' in log()
            client.write('/ctl', b'Kill sleep\n')
            wait(lambda: f'\nexit {sleeper} ' in log(), 'Kill: no exit logged')

            # A REPL send: a program in the terminal (cat, which says back
            # what it is sent; a nested shell reads as the shell itself)
            # bound as bash's REPL, an exec on a .sh file's body (its event
            # written back) goes there.
            s = look(client, str(root / 's.sh'))
            client.write(f'/pane/{term}/pty/data', b'cat\r')
            wait(lambda: client.read(f'/pane/{term}/pty/status').split()[-1] == b'1', ('the REPL program did not start', client.read(f'/pane/{term}/body')[-300:]))
            client.write(f'/pane/{term}/ctl', b'Repl bash\n')
            fid = client.open(f'/pane/{s}/event', 2)
            try:
                record = b'MX0 0 0 14 echo from-file\n'
                client.rpc(118, struct.pack('<IQI', fid, 0, len(record)) + record)
            finally:
                client.close(fid)
            wait(lambda: f'\nsend {s} {term} ' in log(), ('no REPL send logged', log()[-800:]))
            wait(lambda: client.read(f'/pane/{term}/body').count(b'echo from-file') >= 2, 'the REPL did not get it')
            client.write(f'/pane/{term}/ctl', b'Repl -\n')
            client.write(f'/pane/{term}/pty/data', b'\x04')
            wait(lambda: run(b'true\n') == b'exit 0\n', 'the terminal did not come back to its shell')

            # The layout and a column's files: Newcol, its tag, rmdir.
            assert len(client.read('/layout').splitlines()) >= 2  # a column, then `active`
            client.write('/ctl', b'Newcol\n')
            columns = [row for row in client.read('/layout').splitlines() if not row.startswith(b'active')]
            assert len(columns) == 2, columns
            empty = columns[-1].split()[0].decode()
            assert b'New' in client.read(f'/col/{empty}/tag')
            client.remove(f'/col/{empty}')
            assert len([row for row in client.read('/layout').splitlines() if not row.startswith(b'active')]) == 1

            # ask/answer: Save on a terminal asks for a path over 9P.
            client.write(f'/pane/{term}/exec', b'Save\n')
            wait(lambda: f'ask {term} save path\n' in log(), 'no ask logged')
            client.write(f'/pane/{term}/ctl', f'answer {root}/scroll.txt\n'.encode())
            wait(lambda: (root / 'scroll.txt').exists(), 'the answered Save wrote nothing')
            assert f'answer {term} {root}/scroll.txt\n' in log()

            # Dump, then Restore it: the new session names its panes.
            client.write('/ctl', b'Dump\n')
            saved = root / 'flow.dump.zon'
            wait(saved.exists, 'Dump wrote nothing')
            assert f'dump {saved}\n' in log()
            try:
                client.write('/ctl', f'Restore {saved}\n'.encode())
            except (EOFError, OSError):
                pass
            time.sleep(.5)
            with Client(address) as restored:
                wait(lambda: f'restore {saved}' in restored.read('/log').decode(), 'Restore not logged')
                names = restored.read('/index').decode()
                assert f'{root}/a.txt' in names and f'{root}/b.txt' in names, names
                # Recent goes on across the Restore, and is kept after it.
                assert f'{root}/a.txt\n' in restored.read('/recent').decode(), restored.read('/recent')
                (root / 'after.txt').write_bytes(b'after\n')
                look(restored, str(root / 'after.txt'))
                assert f'open {root}/after.txt\n' in restored.read('/recent').decode()
                state = root / '.local' / 'state' / 'pardes' / 'recent'
                wait(lambda: f'{root}/after.txt'.encode() in state.read_bytes(), 'Recent not kept after Restore')
    assert time.monotonic() - started < 30, 'the canary took too long'


def python_repl_runs_one_line(binary):
    """A one-line exec sent to Python's own REPL (3.13's pyrepl, which asks
    for bracketed paste) runs at once, with no second Enter: the Enter
    reaches it after the paste, not inside it. A real pty, a real python3."""
    python = shutil.which('python3')
    if python is None:
        return
    version = subprocess.run([python, '-c', 'import sys; print(sys.version_info >= (3, 13))'],
                             capture_output=True, text=True).stdout.strip()
    if version != 'True':
        return
    with tempfile.TemporaryDirectory(prefix='pardes-repl-') as directory:
        with session(binary, Path(directory), 'repl') as (client, _):
            def panes(client):
                return {int(row.split()[0]) for row in client.read('/index').splitlines()}
            before = panes(client)
            client.write('/pane/1/ctl', b'Tty\n')
            deadline = time.monotonic() + 10
            while time.monotonic() < deadline and panes(client) == before:
                time.sleep(.05)
            tty = max(panes(client) - before)
            def body():
                return client.read(f'/pane/{tty}/body').decode('utf-8', 'replace')
            def wait_for(text, seconds=15):
                deadline = time.monotonic() + seconds
                while time.monotonic() < deadline:
                    if text in body():
                        return True
                    time.sleep(.1)
                return False
            assert wait_for('$') or wait_for('>'), body()
            client.write(f'/pane/{tty}/pty/data', python.encode() + b' -q\r')
            assert wait_for('>>>'), body()
            client.write(f'/pane/{tty}/ctl', b'Repl python\n')
            # An exec on the REPL's own body goes to the REPL, as its event
            # record written back does: an empty range acts on its text.
            fid = client.open(f'/pane/{tty}/event', 2)
            try:
                record = b'MX0 0 0 10 print(6*7)\n'
                client.rpc(118, struct.pack('<IQI', fid, 0, len(record)) + record)
                assert wait_for('\n42'), body()
                # After the session has sat idle, the Enter still waits for
                # the paste: core time is as old as the last frame then.
                for n in range(5):
                    time.sleep(2)
                    line = f'print({1000 + n})'.encode()
                    record = b'MX0 0 0 %d %s\n' % (len(line), line)
                    client.rpc(118, struct.pack('<IQI', fid, 0, len(record)) + record)
                    assert wait_for(f'\n{1000 + n}'), (n, body())
                # More than one line, with and without a newline at the end,
                # and a block with a statement after it: each runs whole,
                # and the REPL is back at its prompt, not at `...`.
                for code, outputs in [(b'print(11)\nprint(22)', ['11', '22']),
                                      (b'print(33)\nprint(44)\n', ['33', '44']),
                                      (b'for i in range(2):\n    print(55 + i)\nprint(77)\n', ['55', '56', '77'])]:
                    record = b'MX0 0 0 %d %s\n' % (len(code), code)
                    client.rpc(118, struct.pack('<IQI', fid, 0, len(record)) + record)
                    for out in outputs:
                        assert wait_for(f'\n{out}\n'), (code, out, body())
                    time.sleep(.5)
                    assert body().rstrip().endswith('>>>'), (code, body()[-200:])
            finally:
                client.close(fid)


def signal_cleans_up(binary):
    """A detached session told to stop by a signal unlinks its sockets."""
    for sig in [signal.SIGTERM, signal.SIGINT, signal.SIGHUP]:
        with tempfile.TemporaryDirectory(prefix='pardes-sig-') as directory:
            root = Path(directory)
            seen = {}
            with session(binary, root, 'sig', terminal=seen) as (client, address):
                detached = root / 'pardes-detached-sig.sock'
                assert address.exists() and detached.exists()
                os.kill(seen['pid'], sig)
                deadline = time.monotonic() + 5
                while (address.exists() or detached.exists()) and time.monotonic() < deadline:
                    time.sleep(.01)
                assert not address.exists() and not detached.exists(), (sig, list(root.iterdir()))


def odd_paths(binary):
    """Long and odd paths written where a path goes -- name, look, DumpDir,
    Dump, Save -- are refused or said, never a crash: a part over 255 bytes
    (ENAMETOOLONG), a loop of links (ELOOP), a file taken for a directory
    (ENOTDIR), one not ours (EACCES)."""
    with tempfile.TemporaryDirectory(prefix='pardes-odd-') as directory:
        root = Path(directory)
        (root / 'loop').symlink_to(root / 'loop')
        (root / 'plain').write_bytes(b'x\n')
        odd = ['/tmp/' + 'x' * 300 + '/f.txt', str(root / ('y' * 300)), str(root / 'loop' / 'f'),
               str(root / 'plain' / 'f'), '/root/nope/f', str(root / 'a' / '..' / 'b.txt')]
        with session(binary, root, 'odd') as (client, _):
            serial = new_pane(client, b'text\n')
            for path in odd:
                for where, line in ((f'/pane/{serial}/name', path), ('/look', path),
                                    ('/ctl', f'DumpDir {path}'), ('/ctl', 'Dump'), (f'/pane/{serial}/ctl', 'Save')):
                    try:
                        client.write(where, (line + '\n').encode())
                    except OSError:
                        pass
                    assert client.read('/index'), (where, line)
            client.write('/ctl', b'DumpDir\n')
            # A theme file that is not there, or no theme, fails the write.
            (root / 'bad.zon').write_bytes(b'.{ .nonsense = 1 }')
            for line, why in ((f'ThemeFile {root}/missing.zon', 'theme file'), (f'ThemeFile {root}/bad.zon', 'theme file'),
                              ('ThemeFile notes.txt', 'a theme file is a .zon')):
                try:
                    client.write('/ctl', (line + '\n').encode())
                    raise AssertionError(f'{line} was taken')
                except OSError as refused:
                    assert why in str(refused), (line, refused)


def delcol_last_keeps_window(binary):
    """Delcol on the last column leaves the window, empty but for the
    workspace tag, as acme does; Newcol and New start again from there."""
    with tempfile.TemporaryDirectory(prefix='pardes-delcol-') as directory:
        with session(binary, Path(directory), 'delcol') as (client, _):
            def columns():
                return [l for l in client.read('/layout').decode().splitlines() if l and not l.startswith('active')]
            while columns():
                client.write(f'/col/{columns()[0].split()[0]}/ctl', b'Delcol\n')
            # Still here, empty.
            assert client.read('/index') == b'', client.read('/index')
            client.write('/ctl', b'Newcol\n')
            assert len(columns()) == 1, columns()
            serial = new_pane(client, b'back\n')
            assert str(serial).encode() in client.read('/index'), client.read('/index')


def cut_writes(binary):
    """A write bigger than a 9P message comes as several, cut anywhere: each
    command line runs once, whole, an Edit block once it ends, and a last
    line with no newline when the open closes. Never a fragment."""
    with tempfile.TemporaryDirectory(prefix='pardes-cut-') as directory:
        with session(binary, Path(directory), 'cut') as (client, _):
            burst = b''.join(b'Msg burst-%04d\n' % i for i in range(1000))
            assert len(burst) > client.msize
            client.write('/exec', burst)
            log = client.read('/log').decode()
            said = [line.split(' ', 2)[2] for line in log.splitlines() if line.startswith('msg ') and 'burst-' in line]
            assert said == ['burst-%04d' % i for i in range(1000)], said[:3] + said[-3:]
            assert '\nrun ' not in log and 'exit 127' not in log, log[-400:]
            client.write('/exec', b'Msg no-newline-at-end')
            assert 'msg ' in client.read('/log').decode().split('no-newline-at-end')[0][-40:]
            serial = new_pane(client, b'')
            block = b'Edit ,a\n' + b''.join(b'line %05d of the block\n' % i for i in range(2500)) + b'.\n'
            assert len(block) > 50000
            client.write(f'/pane/{serial}/ctl', block)
            body = client.read(f'/pane/{serial}/body').decode().splitlines()
            assert body == ['line %05d of the block' % i for i in range(2500)], (len(body), body[:2])
            # A detached session with no frontend is 160x50 until told.
            screen = client.screen()
            assert (screen['cols'], screen['rows']) == (160, 50), (screen['cols'], screen['rows'])
            client.write('/ctl', b'size 132 40\n')
            screen = client.screen()
            assert (screen['cols'], screen['rows']) == (132, 40), (screen['cols'], screen['rows'])


def test(binary, quic=False):
    started = time.monotonic()
    cut_writes(binary)
    odd_paths(binary)
    delcol_last_keeps_window(binary)
    run_file(binary)
    exit_answers(binary)
    unattended_messages_expire(binary)
    new_terminals_named_once(binary)
    workflow_canary(binary)
    recent_survives_restart(binary)
    python_repl_runs_one_line(binary)
    signal_cleans_up(binary)
    for options, message in [
        (['--fs'], b'no such option'),
        (['--fs9'], b'no such option'),
        (['--', 'a', 'b'], b'is the second'),
        (['--9p'], b'--9p needs a socket name'),
        (['--9p='], b'invalid 9P socket name'),
        (['--9p=a/b'], b'invalid 9P socket name'),
        (['--attach=work', '--9p=work'], b'--attach has none'),
        (['--9p-tcp'], b'--9p-tcp needs'),
        (['--9p-tcp='], b'--9p-tcp needs'),
        (['--9p-tcp=tcp!localhost!5640'], b'--9p-tcp needs'),
        (['--9p-tcp=quic!127.0.0.1!5640'], b'--9p-tcp needs'),
        (['--9p-tcp=tcp!127.0.0.1!5640', '--9p-tcp=tcp!127.0.0.1!5641'], b'specified twice'),
        (['--attach=work', '--9p-tcp=tcp!127.0.0.1!5640'], b'--attach has none'),
        (['--9p-quic'], b'--9p-quic needs'),
        (['--9p-quic='], b'--9p-quic needs'),
        (['--9p-quic=tcp!127.0.0.1!5640'], b'--9p-quic needs'),
        (['--mount=os=work'], b'invalid mount name'),
        (['--mount=peer='], b'invalid mount name or dial'),
        (['--mount=peer=one', '--mount=peer=two'], b'duplicate mount'),
        (['--mount=peer=tcp!127.0.0.1!0'], b'invalid mount dial'),
        (['--attach=work', '--mount=peer=one'], b'--attach has none'),
    ]:
        result = subprocess.run([binary, *options], capture_output=True, timeout=5)
        assert result.returncode == 1 and message in result.stderr, (options, result)
    with tempfile.TemporaryDirectory(prefix='pardes-fs-') as directory:
        root = Path(directory)
        with session(binary, root, 'renamed', '--9p=explicit', socket_name='explicit') as (named, named_address):
            assert named_address.name == 'pardes-9p-explicit.sock'
            assert not (root / 'pardes-9p-renamed.sock').exists()
            assert named.read('/pane/1/body') == b'initial\n'
        with session(binary, root, 'first') as (client, address):
            assert {'pane', 'index', 'status', 'look', 'exec', 'log', 'os'} <= set(client.list('/'))
            assert 'self' not in client.list('/')
            assert client.read('/pane/1/body') == b'initial\n'
            assert client.read('/os' + str(root / 'first.txt')) == b'initial\n'
            assert 'first.txt' in client.list('/os' + str(root))
            assert client.read('/pane/1/../1/body') == b'initial\n'
            assert client.read('/os' + str(root) + '/../' + root.name + '/first.txt') == b'initial\n'

            listing = root / 'listing'
            listing.mkdir()
            (listing / 'broken').symlink_to(listing / 'missing')
            for number in range(24):
                (listing / f'entry-{number:02}').touch()
            with Client(address, msize=256) as small:
                assert set(small.list('/os' + str(listing))) == {
                    f'entry-{number:02}' for number in range(24)}
            # 64 KiB frames: a ctl line with no newline, whole in its one
            # Twrite, runs with it however long -- the cutoff for a write
            # that may go on is the frame the client asked for, less 24.
            with Client(address, msize=65536) as big:
                assert big.msize == 65536
                ctl = big.open('/pane/1/ctl', 1)
                line = b'bogus' + b'x' * 20000
                try:
                    big.rpc(118, struct.pack('<IQI', ctl, 0, len(line)) + line)
                except OSError:
                    pass  # refused with its write (a line over 1024 bytes), not held
                else:
                    raise AssertionError('a whole 20000-byte line waited for a newline')
                big.close(ctl)

            frozen = client.open('/screen')
            before = bytearray(client.read_fid(frozen, count=31))
            client.write('/pane/1/body', b'screen changed\n')
            while chunk := client.read_fid(frozen, len(before)):
                before.extend(chunk)
            client.close(frozen)
            old_screen = json.loads(before)
            old_text = ''.join(cell[0] for cell in old_screen['cells'])
            assert 'initial' in old_text and 'screen changed' not in old_text
            assert 'screen changed' in ''.join(cell[0] for cell in client.screen()['cells'])

            payload = ('const value = "λ";\n' * 3000).encode()
            serial = new_pane(client, payload)
            body = f'/pane/{serial}/body'
            assert client.read(body) == payload
            client.write(body, b'replaced\n', truncate=True)
            assert client.read(body) == b'replaced\n'

            event = client.open(f'/pane/{serial}/event')
            client.write(body, b'event\n')
            record = client.read_fid(event)
            assert record.startswith(b'EI') and b'event\n' in record, record
            client.close(event)

            os_path = '/os' + str(root / 'first.txt')
            client.write(os_path, b'written through 9P\n', truncate=True)
            assert (root / 'first.txt').read_bytes() == b'written through 9P\n'
            assert client.read(os_path) == b'written through 9P\n'

            for index in range(12):
                with Client(address) as other:
                    assert other.read(body) == b'replaced\nevent\n', index

            missing = False
            try:
                client.read('/pane/4294967295/body')
            except OSError:
                missing = True
            assert missing
            assert client.read(body) == b'replaced\nevent\n'

            if 'src' in client.list('/'):
                source = look(client, '/virtual/src/pardes.zig')
                assert b'pub const Pardes' in client.read(f'/pane/{source}/body')

            (root / 'index').write_bytes(b'OS takes precedence\n')
            opened = look(client, 'index')
            assert client.read(f'/pane/{opened}/body') == b'OS takes precedence\n'
            opened = look(client, '/n/self/index')
            index = client.read(f'/pane/{opened}/body')
            assert index.split()[0] == b'1' and b'first.txt' in index, index
            (root / 'explicit.txt').write_bytes(b'explicit OS mount\n')
            opened = look(client, '/n/os' + str(root / 'explicit.txt'))
            assert client.read(f'/pane/{opened}/body') == b'explicit OS mount\n'

            own = root / 'pardes-9p-second.sock'
            with session(binary, root, 'second', '--mount=peer=' + str(address),
                         '--mount=own=' + str(own)) as (remote, _):
                opened = look(remote, '/n/own/pane/1/body')
                assert remote.read(f'/pane/{opened}/body') == b'initial\n'
                opened = look(remote, '/n/peer/pane/' + str(serial) + '/body')
                assert remote.read(f'/pane/{opened}/body') == b'replaced\nevent\n'
                remote.write(f'/pane/{opened}/body', b'saved to peer\n', truncate=True)
                remote.write(f'/pane/{opened}/exec', b'Save\n')
                assert client.read(body) == b'saved to peer\n'

                large = b'mounted file contents\n' * 60000
                large_path = root / 'large.txt'
                large_path.write_bytes(large)
                opened = look(remote, '/n/peer/os' + str(large_path))
                assert remote.read(f'/pane/{opened}/body') == large
                edited = large[:-3] + b'edited\n'
                remote.write(f'/pane/{opened}/body', edited, truncate=True)
                remote.write(f'/pane/{opened}/exec', b'Save\n')
                assert large_path.read_bytes() == edited
                assert client.read('/os' + str(large_path)) == edited

            with session(binary, root, 'runtime') as (remote, own_address):
                control = new_pane(remote, b'')
                execute(remote, control, 'Mount peer ' + str(address))
                opened = look(remote, '/n/peer/pane/1/body', source=control)
                assert remote.read(f'/pane/{opened}/body') == client.read('/pane/1/body')
                # A builtin that fails through exec fails the write (one rule).
                execute_refused(remote, control, 'Mount peer ' + str(own_address), 'already mounted')
                execute_refused(remote, control, 'Unmount peer', 'mount in use')  # a pane is open on it
                remote.write(f'/pane/{opened}/body', b'runtime mounted Save\n', truncate=True)
                remote.write(f'/pane/{opened}/exec', b'Save\n')
                assert client.read('/pane/1/body') == b'runtime mounted Save\n'
                assert remote.read('/pane/1/body') == b'initial\n'
                roots = look(remote, '/n', source=control)
                assert b'/n/peer/\n' in remote.read(f'/pane/{roots}/body')
                remote.write(f'/pane/{opened}/exec', b'Del\n')
                execute(remote, control, 'Unmount peer')
                remote.write(f'/pane/{roots}/ctl', b'get\n')
                assert remote.read(f'/pane/{roots}/body') == b'/n/os/\n/n/self/\n'
                execute(remote, control, 'Mount peer ' + str(own_address))
                opened = look(remote, '/n/peer/pane/1/body', source=control)
                assert remote.read(f'/pane/{opened}/body') == b'initial\n'

            with session(binary, root, 'mounted-startup', '--mount=peer=' + str(address),
                         file='/n/peer/pane/1/body:1') as (remote, _):
                assert remote.read('/pane/1/body') == b'runtime mounted Save\n'

            client.write(f'/pane/{serial}/exec', b'Del\n')
            assert str(serial) not in client.list('/pane')

        for tty in [False, True]:
            name = 'restore tty space' if tty else 'restore detached space'
            with session(binary, root, name, tty=tty) as (old, address):
                old.write('/pane/1/body', b'dumped state\n', truncate=True)
                control = new_pane(old, b'')
                execute_refused(old, control, 'Restore ' + str(root / 'missing dump.zon'), 'no such file')
                assert old.read('/pane/1/body') == b'dumped state\n'
                execute(old, control, 'Mount own ' + str(address))
                execute(old, control, 'Dump')
                saved = root / (name + '.dump.zon')
                deadline = time.monotonic() + 5
                while not saved.exists():
                    assert time.monotonic() < deadline, 'Dump did not write its state'
                    time.sleep(.005)
                held_body = old.open('/pane/1/body')
                held_screen = old.open('/screen')
                old_ids = {int(row.split()[0]) for row in old.read('/index').splitlines()}
                assert old.read_fid(held_body) == b'dumped state\n'
                assert old.read_fid(held_screen, count=1) == b'{'
                old.write('/pane/1/body', b'changed after dump\n', truncate=True)
                # Unsaved text is asked about once, as Exit asks; the second
                # Restore goes ahead, and is answered before the hang-up.
                try:
                    execute(old, control, 'Restore ' + str(saved))
                except OSError as refused:
                    assert 'Modified' in str(refused), refused
                assert old.read('/pane/1/body') == b'changed after dump\n'
                # Answered before the hang-up: the write's reply comes; the
                # clunk after it may meet the connection already cut.
                restore = old.open(f'/pane/{control}/exec', 1)
                line = ('Restore ' + str(saved) + '\n').encode()
                old.rpc(118, struct.pack('<IQI', restore, 0, len(line)) + line)
                try:
                    old.close(restore)
                except (EOFError, OSError):
                    pass
                # Refused while the connections are cut, or cut already.
                try:
                    old.read_fid(held_body)
                except (EOFError, OSError):
                    pass
                else:
                    raise AssertionError('Restore left an old body fid usable')
                with Client(address) as restored:
                    # The new log names its panes, then where they came from.
                    log = restored.read('/log')
                    assert b'\nrestore ' + str(saved).encode() + b'\n' in log, log
                    # then which serial each pane had: restored <old> <new>
                    assert (b'\n'.join(b'restored %d %d' % (old, new) for old, new in zip(sorted(old_ids), sorted(int(r.split()[0]) for r in restored.read('/index').splitlines()))) + b'\nrestoredcol ') in log, log
                    restored_ids = [int(row.split()[0]) for row in restored.read('/index').splitlines()]
                    assert len(restored_ids) == 2 and old_ids.isdisjoint(restored_ids), restored_ids
                    file_id, control_id = restored_ids
                    contents = restored.read(f'/pane/{file_id}/body')
                    assert contents == b'dumped state\n', (contents, saved.read_text(), restored.read('/index'))
                    assert restored.read(f'/pane/{control_id}/body') == b'', 'commands ran through exec, not the body'
                    assert restored.screen()['cols'] == (80 if tty else 160)  # a detached one is 160x50
                    opened = look(restored, '/n/own/os' + str(root / (name + '.txt')), source=control_id)
                    assert restored.read(f'/pane/{opened}/body') == b'initial\n'

        with session(binary, root, 'forwarding') as (client, address):
            child_dir = root / 'child-working-directory'
            child_dir.mkdir()
            spaced = child_dir / 'space name.txt'
            spaced.write_bytes(b'first line\nsecond line\n')
            env = os.environ.copy()
            # PARDES_PID is what says "you are inside a pardes"; the socket
            # and the pane only say how to reach it. Without the pid the child
            # starts its own session, so this test used to pass only when the
            # runner itself happened to be running inside one.
            status = dict(line.split(maxsplit=1) for line in client.read('/status').decode().splitlines())
            env.update(PARDES_9P=str(address), PARDES_PANE='1', PARDES_PID=status['pid'])
            env.pop('PARDES_FORWARD_LOOK', None)
            for word, expected, selected, reuse in [('space name.txt:2:4', spaced.read_bytes(), [14, 14], False),
                                                    ('/n/self/pane/1/body', b'initial\n', None, False),
                                                    ('/virtual/pane/1/body:1:2-4', b'initial\n', [1, 4], True)]:
                before = {int(row.split()[0]) for row in client.read('/index').splitlines()}
                result = subprocess.run([binary, '--tty', word], cwd=child_dir, env=env,
                                        capture_output=True, timeout=3)
                assert result.returncode == 0, (word, result.stderr)
                after = {int(row.split()[0]) for row in client.read('/index').splitlines()}
                if reuse:
                    assert after == before
                else:
                    opened, = after - before
                assert client.read(f'/pane/{opened}/body') == expected
                assert client.read('/pane/1/body') == b'initial\n'
                if selected is not None:
                    # The selection is dot; reading it takes no command at all.
                    actual = list(map(int, client.read(f'/pane/{opened}/dot').split()))
                    assert actual == selected, (word, actual, selected)
            addr = client.open('/pane/1/addr', 2)
            limit = client.open('/pane/1/limit', 2)
            for fid, text in [(addr, b'#0,#3'), (limit, b'#0,#3'), (addr, b'#0')]:
                client.rpc(118, struct.pack('<IQI', fid, 0, len(text)) + text)
            try:
                for _ in range(8):
                    result = subprocess.run([binary], cwd=child_dir, env=env,
                                            capture_output=True, timeout=3)
                    assert result.returncode == 1 and b'already inside pardes' in result.stderr
                    text = b'/tial/'
                    try:
                        client.rpc(118, struct.pack('<IQI', addr, 0, len(text)) + text)
                    except OSError:
                        pass
                    else:
                        raise AssertionError('no-argument child cleared the parent address limit')
            finally:
                client.close(addr)
                client.close(limit)
            client.read('/pane/1/ctl')

            control = new_pane(client, b'')
            execute(client, control, 'Dump')
            saved = root / 'forwarding.dump.zon'
            deadline = time.monotonic() + 3
            while not saved.exists():
                assert time.monotonic() < deadline, 'forwarding fixture Dump did not finish'
                time.sleep(.005)
            before = client.read('/index')
            inherited = {key: env[key] for key in ['PARDES_9P', 'PARDES_PANE', 'PARDES_PID']}
            cases = [
                ('mount', ['--mount=peer=' + str(address), '/n/peer/pane/1/body'], None),
                ('name', ['--9p=forwarded-name'], 'forwarded-name'),
                ('tcp', ['--9p-tcp=tcp!127.0.0.1!0'], None),
                ('load', ['-l', str(saved)], None),
                ('shells', ['-n', '3'], None),
                ('toggle', ['--tty-toggle=q'], None),
                ('toggle-arg', ['--tty-toggle', 'q'], None),
            ]
            if quic:
                cases.append(('quic', ['--9p-quic=quic!127.0.0.1!0'], None))
            for name, options, socket_name in cases:
                with session(binary, root, 'forward-' + name, tty=True, socket_name=socket_name,
                             launch=[*([] if name == 'shells' else ['--tty']), *options],
                             inherited=inherited) as (local, local_address):
                    assert local_address != address
                    index = local.read('/index').splitlines()
                    if name == 'mount':
                        assert len(index) == 1 and b'/n/peer/' in index[0], index
                        assert local.read('/pane/1/body') == b'initial\n'
                    elif name == 'load':
                        assert local.read('/index') == before
                        assert local.read('/pane/1/body') == b'initial\n'
                    else:
                        # A bare tty launch is a shell plus the empty text pane
                        # the boot puts under it; --shells=3 is the classic
                        # three-shell layout and has no scratch.
                        assert len(index) == (3 if name == 'shells' else 2), index
                    if name in ['tcp', 'quic']:
                        assert (name + '!127.0.0.1!').encode() in local.read('/listeners')
                    assert client.read('/index') == before
                    assert client.read('/pane/1/body') == b'initial\n'

            # A session that answers takes the file or says why, as acme's B
            # does: a pane it has not is its refusal and exit 1, never a
            # nested editor drawn inside the pane that asked.
            gone = subprocess.run([binary, '--tty', str(spaced)], cwd=root,
                                  env=dict(env, PARDES_PANE='4294967295'), stdin=subprocess.DEVNULL,
                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert gone.returncode == 1 and gone.stderr.startswith(b'pardes: '), gone
            assert client.read('/index') == before
            for name, identity, word in [
                ('stale-file', dict(inherited, PARDES_9P=str(root / 'absent.sock')), str(spaced)),
                ('stale-missing', dict(inherited, PARDES_9P=str(root / 'absent.sock')), 'missing-child-file.txt'),
                ('stale-noarg', dict(inherited, PARDES_9P=str(root / 'absent.sock')), None),
                # No pid means "not inside a pardes at all", which is what
                # --nested withholds and what a plain shell has.
                ('not-nested', {k: v for k, v in inherited.items() if k != 'PARDES_PID'}, str(spaced)),
            ]:
                with session(binary, root, name, tty=True, inherited=identity,
                             launch=['--tty', *([word] if word else [])]) as (local, local_address):
                    assert local_address != address
                    if word == str(spaced):
                        assert local.read('/pane/1/body') == spaced.read_bytes()
                    elif word:
                        assert word.encode() in local.read('/index') or word in ''.join(
                            cell[0] for cell in local.screen()['cells'])
                    assert client.read('/index') == before

        # --wait (acme's E): a launch inside a pane lasts until the pane it
        # opened is deleted, as an $EDITOR must; one on a file already open
        # waits on that pane; a session that goes away ends it with 1.
        with session(binary, root, 'wait') as (client, address):
            # A command run from a pane's tag gets acme's $winid, that pane;
            # one from the workspace's tag, none.
            def command_body(node, line):
                client.write(node, line)
                serial = int(client.read(node).split()[-1])
                deadline = time.monotonic() + 10
                while True:
                    log = client.read('/log').decode()
                    ran = log.rfind(f'\nrun {serial} ')
                    if ran >= 0 and f'\nexit {serial} ' in log[ran:]:
                        body = client.read(f'/pane/{serial}/body').decode().split('% echo winid')[-1]
                        client.remove(f'/pane/{serial}')
                        return body
                    assert time.monotonic() < deadline, log[-400:]
                    time.sleep(.05)
            first_pane = min(int(row.split()[0]) for row in client.read('/index').splitlines())
            assert f'winid={first_pane}\n' in command_body(f'/pane/{first_pane}/tagexec', b'echo winid=$winid\n')
            assert 'winid=\n' in command_body('/tagexec', b'echo winid=$winid\n')
            # A second --detach of a name a live session holds: said, exit 1,
            # no Zig error trace.
            second_env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_')}
            second_env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root))
            second = subprocess.run([binary, '--detach=wait'], cwd=root, env=second_env,
                                    stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=20)
            assert second.returncode == 1, second
            assert second.stderr == b'pardes: a session named wait is already running\n', second.stderr
            status = dict(line.split(maxsplit=1) for line in client.read('/status').decode().splitlines())
            env = {k: v for k, v in os.environ.items() if k != 'PARDES_FORWARD_LOOK'}
            env.update(PARDES_9P=str(address), PARDES_PANE='1', PARDES_PID=status['pid'])
            edited = root / 'wait edit.txt'
            edited.write_bytes(b'edit me\n')

            def serials():
                return {int(row.split()[0]) for row in client.read('/index').splitlines()}

            def launch(flag):
                before = serials()
                child = subprocess.Popen([binary, flag, str(edited)], cwd=root, env=env,
                                         stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
                deadline = time.monotonic() + 3
                while child.poll() is None and not serials() - before and time.monotonic() < deadline:
                    time.sleep(.01)
                return child

            first = launch('--wait')
            opened, = serials() - {1}
            second = launch('-w')
            time.sleep(.5)
            assert first.poll() is None and second.poll() is None, (first.stderr.read(), second.stderr.read())
            assert serials() == {1, opened}
            client.write(f'/pane/{opened}/body', b'edited\n', truncate=True)
            execute(client, opened, 'Save')
            client.remove(f'/pane/{opened}')
            deleted = time.monotonic()
            assert first.wait(timeout=5) == 0
            # Woken by the `del` record, not a poll.
            assert time.monotonic() - deleted < .05, time.monotonic() - deleted
            assert second.wait(timeout=5) == 0
            assert edited.read_bytes() == b'edited\n'
            # `--` ends the options: a name starting with a dash is a name.
            dash = root / '-dash.txt'
            dash.write_bytes(b'dash\n')
            before = serials()
            assert subprocess.run([binary, '--', dash.name], cwd=root, env=env, capture_output=True, timeout=10).returncode == 0
            dashed, = serials() - before
            assert client.read(f'/pane/{dashed}/name').rstrip(b'\n') == str(dash).encode()
            client.remove(f'/pane/{dashed}')
            # A name under a directory that may not be searched is refused,
            # permission denied, exit 1, no pane: its Save could never be.
            (root / 'noperm').mkdir()
            (root / 'noperm').chmod(0)
            before = serials()
            try:
                denied = subprocess.run([binary, 'noperm/a/b/c'], cwd=root, env=env, capture_output=True, timeout=10)
            finally:
                (root / 'noperm').chmod(0o755)
            assert denied.returncode == 1 and b'permission denied' in denied.stderr, denied
            assert serials() == before
            # --wait waits on the pane its look answered, on the look's own
            # open: a name /index shows escaped (a backslash is `\\` there)
            # is still the pane waited on, not one already gone.
            odd = root / 'wait\\odd.txt'
            odd.write_bytes(b'odd\n')
            before = serials()
            waiting = subprocess.Popen([binary, '--wait', str(odd)], cwd=root, env=env,
                                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
            deadline = time.monotonic() + 3
            while not serials() - before and time.monotonic() < deadline:
                time.sleep(.01)
            odd_pane, = serials() - before
            time.sleep(.5)
            assert waiting.poll() is None, waiting.stderr.read()
            client.remove(f'/pane/{odd_pane}')
            assert waiting.wait(timeout=5) == 0
            # A name not there yet opens a pane named it, as acme's B does,
            # and Save creates it: git's and fish's temporary files are.
            fresh = root / 'wait new.txt'
            before = serials()
            made = subprocess.Popen([binary, '--wait', fresh.name], cwd=root, env=env,
                                    stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
            # The launch makes the pane (pane/new), then names it: wait for
            # the name, not the pane.
            deadline = time.monotonic() + 3
            while str(fresh).encode() not in client.read('/index'):
                assert made.poll() is None, made.stderr.read()
                assert time.monotonic() < deadline, 'a new name opened no pane named it'
                time.sleep(.01)
            named, = serials() - before
            client.write(f'/pane/{named}/body', b'new\n')
            execute(client, named, 'Save')
            client.remove(f'/pane/{named}')
            code = made.wait(timeout=5)
            assert code == 0, (code, made.stderr.read())
            assert fresh.read_bytes() == b'new\n'
            # ...in a directory not there either: forwarded all the same, and
            # Save makes the directories.
            deep = root / 'wait-missing' / 'sub' / 'new.txt'
            before = serials()
            made = subprocess.Popen([binary, 'wait-missing/sub/new.txt'], cwd=root, env=env,
                                    stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
            assert made.wait(timeout=5) == 0, made.stderr.read()
            deadline = time.monotonic() + 3
            while str(deep).encode() not in client.read('/index'):
                assert time.monotonic() < deadline, 'a name in a missing directory opened no pane named it'
                time.sleep(.01)
            named, = serials() - before
            client.write(f'/pane/{named}/body', b'deep\n')
            execute(client, named, 'Save')
            client.remove(f'/pane/{named}')
            assert deep.read_bytes() == b'deep\n'
            # `pardes '~/x'`, quoted so the shell left the `~`: home all the
            # same: the launching shell's HOME.
            before = serials()
            quoted = subprocess.run([binary, '~/fwd-tilde.txt'], cwd=root, env=dict(env, HOME=str(root)),
                                    stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=10)
            assert quoted.returncode == 0, quoted.stderr
            assert f'{root}/fwd-tilde.txt'.encode() in client.read('/index'), client.read('/index')
            for serial in serials() - before:
                client.remove(f'/pane/{serial}')
            # A name the session refuses: its reason and exit 1, as B's,
            # never a nested editor, and no pane left for it.
            before = serials()
            refused = subprocess.run([binary, 'x' * 300 + '.txt'], cwd=root, env=env,
                                     stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert refused.returncode == 1, refused
            assert b'a component over 255 bytes' in refused.stderr, refused.stderr
            assert serials() == before, (serials(), before)
            # A pane address the session has not is refused in words, exit
            # 1, no pane made for a file of that name; one it has is looked
            # at. A name with a newline is said in words, not an error name.
            # `pardes -` (a terminal's PAGER): stdin, its escapes stripped, in a
            # +Pager pane of the session, clean, and back at once.
            fed = subprocess.run([binary, '-'], cwd=root, env=env, input=b'\x1b[1mbold\x1b[0m text\n',
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert fed.returncode == 0, fed.stderr
            made_now, = serials() - before
            assert client.read(f'/pane/{made_now}/name') == f'{root}/+Pager\n'.encode()
            assert client.read(f'/pane/{made_now}/body') == b'bold text\n'
            row = [r for r in client.read('/index').splitlines() if r.split()[0] == str(made_now).encode()][0]
            assert row.split()[2] == b'0', row  # clean: nothing to ask at Exit
            # Its `new` record names it as it is: no +New and a rename.
            log_now = client.read('/log').decode()
            assert f'\nnew {made_now} {root}/+Pager\n' in '\n' + log_now, log_now[-400:]
            assert f'\nrename {made_now} ' not in '\n' + log_now
            # The directory's one +Pager: the next paged text refills it.
            fed = subprocess.run([binary, '-'], cwd=root, env=env, input=b'again\r\nprogress 10%\rprogress 100%\n',
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert fed.returncode == 0, fed.stderr
            assert serials() - before == {made_now}, (serials(), before)
            assert client.read(f'/pane/{made_now}/body') == b'again\nprogress 100%\n'
            client.remove(f'/pane/{made_now}')
            # More than one 8 MiB write's worth arrives whole, byte for byte,
            # in the one +Pager (each write once replaced the last).
            big = b''.join(b'line %08d of the paged text\n' % i for i in range(400000))  # ~11 MiB
            assert len(big) > 8 * 1024 * 1024
            fed = subprocess.run([binary, '-'], cwd=root, env=env, input=big,
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=120)
            assert fed.returncode == 0, fed.stderr
            big_pane, = serials() - before
            assert client.read(f'/pane/{big_pane}/body') == big, 'the paged body is not the text'
            # Past the 256 MiB cap: the first 256 MiB, then the cut note.
            cap = 256 * 1024 * 1024
            huge = subprocess.Popen(['sh', '-c', f'yes 0123456789abcdef | head -c {cap + 4096}'], stdout=subprocess.PIPE)
            fed = subprocess.run([binary, '-'], cwd=root, env=env, stdin=huge.stdout,
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=600)
            huge.wait()
            assert fed.returncode == 0, fed.stderr
            assert serials() - before == {big_pane}
            got = client.read(f'/pane/{big_pane}/body')
            note = b'\n[pardes: the rest of stdin is cut: a pane holds at most a file\'s limit]\n'
            assert got.endswith(note) and len(got) == cap + len(note), (len(got), got[-120:])
            assert got[:cap] == (b'0123456789abcdef\n' * (cap // 17 + 1))[:cap]
            del got, big
            client.remove(f'/pane/{big_pane}')
            # From a directory whose name could not be a ctl line: the
            # session's directory's +Pager, the text whole, no stray +New.
            odd = root / 'two\nlines'
            odd.mkdir()
            fed = subprocess.run([binary, '-'], cwd=odd, env=env, input=b'odd dir text\n',
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert fed.returncode == 0, fed.stderr
            odd_pane, = serials() - before
            assert client.read(f'/pane/{odd_pane}/name') == f'{root}/+Pager\n'.encode()
            assert client.read(f'/pane/{odd_pane}/body') == b'odd dir text\n'
            client.remove(f'/pane/{odd_pane}')
            # Nothing on stdin: no pane, and done.
            fed = subprocess.run([binary, '-'], cwd=root, env=env, input=b'',
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert fed.returncode == 0 and serials() == before, (fed, serials())
            # A new name in a directory there but not writable: refused now,
            # not a pane that only fails at Save.
            shut = root / 'shut-dir'
            shut.mkdir()
            shut.chmod(0o500)
            try:
                refused = subprocess.run([binary, 'shut-dir/new.txt'], cwd=root, env=env,
                                         stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
                assert refused.returncode == 1, refused
                assert b'may not be read or written' in refused.stderr, refused.stderr
                assert serials() == before, (serials(), before)
            finally:
                shut.chmod(0o755)
            for word, said in (('@p999:1', b'pardes: @p999:1: this session has no such pane'),
                               ('two\nlines.txt', b'pardes: two\\nlines.txt: a file name is one line')):
                refused = subprocess.run([binary, word], cwd=root, env=env,
                                         stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
                assert refused.returncode == 1, refused
                assert said in refused.stderr, refused.stderr
                assert b'NotOneLine' not in refused.stderr and b'NoSuchPane' not in refused.stderr
                assert serials() == before, (serials(), before)
            looked = subprocess.run([binary, '@p1:1'], cwd=root, env=env,
                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
            assert looked.returncode == 0, looked.stderr
            assert serials() == before, (serials(), before)
            orphan = launch('--wait')
            assert orphan.poll() is None
        assert orphan.wait(timeout=5) == 1

        # `pardes -` outside a session: an editor whose first pane is a
        # +Pager of stdin, as `vim -`; here a detached one, to read it.
        outside_env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_')}
        outside_env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root), PARDES_NOTIME='1')
        boot = subprocess.Popen([binary, '--detach=stdinboot', '-'], cwd=root, env=outside_env,
                                stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
                                start_new_session=True)
        try:
            boot.stdin.write(b'\x1b[32mfrom stdin\x1b[m\nsecond\n')
            boot.stdin.close()
            sock = root / 'pardes-9p-stdinboot.sock'
            deadline = time.monotonic() + 10
            while not sock.exists():
                assert boot.poll() is None, boot.stderr.read()
                assert time.monotonic() < deadline, 'the stdin boot never listened'
                time.sleep(.05)
            with Client(str(sock)) as booted:
                first = booted.read('/index').split(b'\n')[0].split()
                assert first[3] == f'{root}/+Pager'.encode(), first
                assert booted.read(f'/pane/{first[0].decode()}/body') == b'from stdin\nsecond\n'
        finally:
            boot.terminate()
            boot.wait(timeout=10)

        # Killed with SIGTERM or SIGHUP, a tty editor puts its terminal
        # back (main screen, cooked mode) before it dies: a nested one
        # killed in a pane leaves that pane usable.
        # ...and its 9P socket goes with it. One a dead editor left behind
        # (no process has its pid) is swept at the next start; a live
        # pid's, here this test's own, never is.
        def unix_socket_file(path):
            made = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            made.bind(str(path))
            made.close()
        for sig in (signal.SIGTERM, signal.SIGHUP):
            dead_socket = root / 'pardes-9p-2147483647.sock'
            live_socket = root / f'pardes-9p-{os.getpid()}.sock'
            for path in (dead_socket, live_socket):
                if not path.exists():
                    unix_socket_file(path)
            master, slave = os.openpty()
            fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 24, 80, 0, 0))
            cooked = termios.tcgetattr(slave)
            env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_')}
            env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root), TERM='xterm-256color', PARDES_NOTIME='1')
            killed = subprocess.Popen([binary, '--tty', 'x.txt'], cwd=root, env=env,
                                      stdin=slave, stdout=slave, stderr=subprocess.DEVNULL,
                                      start_new_session=True,
                                      preexec_fn=lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0))
            seen = b''
            try:
                deadline = time.monotonic() + 10
                while b'\x1b[?1049h' not in seen:
                    assert time.monotonic() < deadline, seen[-200:]
                    if select.select([master], [], [], .1)[0]:
                        seen += os.read(master, 65536)
                time.sleep(.3)
                own_socket = root / f'pardes-9p-{killed.pid}.sock'
                assert own_socket.exists()
                assert not dead_socket.exists()
                assert live_socket.exists()
                killed.send_signal(sig)
                assert killed.wait(timeout=5) == -sig
                assert not own_socket.exists(), own_socket
                while select.select([master], [], [], .2)[0]:
                    try:
                        chunk = os.read(master, 65536)
                    except OSError:
                        break
                    if not chunk:
                        break
                    seen += chunk
                tail = seen[seen.rindex(b'\x1b[?1049h'):]
                assert b'\x1b[?1049l' in tail, tail[-200:]
                assert termios.tcgetattr(slave)[3] & termios.ICANON == cooked[3] & termios.ICANON
            finally:
                if killed.poll() is None:
                    killed.kill()
                    killed.wait()
                os.close(master)
                os.close(slave)

        # A launch with no terminal to draw on (a pty that is no one's
        # controlling terminal) says so and exits 1, no error trace.
        master, slave = os.openpty()
        try:
            env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_')}
            env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root), TERM='xterm-256color')
            result = subprocess.run([binary, '--tty', 'x.txt'], cwd=root, env=env, stdin=slave, stdout=slave,
                                    stderr=subprocess.PIPE, start_new_session=True, timeout=10)
        finally:
            os.close(master)
            os.close(slave)
        assert result.returncode == 1, result
        assert b'pardes: no terminal to draw on' in result.stderr and b'.zig:' not in result.stderr, result.stderr

        for tty in [False, True]:
            name = 'lsp-tty' if tty else 'lsp-detached'
            with session(binary, root, name, tty=tty) as (client, _):
                control = new_pane(client, b'')
                execute(client, control, 'Lspinfo')
                deadline = time.monotonic() + 3
                while True:
                    matches = [int(row.split()[0]) for row in client.read('/index').splitlines()
                               if b'+Lsp' in row]
                    if matches:
                        break
                    assert time.monotonic() < deadline, (name, 'Lspinfo did not complete')
                    time.sleep(.005)
                assert len(matches) == 1, matches
                result = client.read(f'/pane/{matches[0]}/body')
                assert result.startswith(b'backend: lsp-client\n'), result
                assert b'protocol servers (lsp-client):' in result, result
                assert client.read('/pane/1/body') == b'initial\n'
                assert '+Lsp' in ''.join(cell[0] for cell in client.screen()['cells'])

        with session(binary, root, 'highlight') as (client, _):
            client.write('/pane/1/name', b'keywords.zig\n')
            source = b'// padding\n' * 9999 + b'fn check() void {\n    if (true) return;\n}\n'
            client.write('/pane/1/body', source, truncate=True)
            screen = client.screen()
            assert len(screen['cells']) == screen['cols'] * screen['rows']
            found = set()
            for row in range(screen['rows']):
                cells = screen['cells'][row * screen['cols']:(row + 1) * screen['cols']]
                text = ''.join(cell[0] for cell in cells)
                for word in ['fn', 'if']:
                    start = text.find(word + ' ')
                    if start < 0:
                        continue
                    styles = [screen['styles'][cell[1]] for cell in cells[start:start + len(word)]]
                    # Weight is opt-in (SyntaxBold, off by default); colour is
                    # not, so the keyword must differ from the text beside it.
                    plain = screen['styles'][cells[start + len(word)][1]]
                    assert all(not style['bold'] and style['fg'] == styles[0]['fg'] for style in styles), (word, styles)
                    assert styles[0]['fg'] != plain['fg'], (word, styles[0], plain)
                    found.add(word)
            assert found == {'fn', 'if'}, screen

        with session(binary, root, 'long-names') as (client, _):
            directory = root / ('directory-' + 'd' * 54)
            directory.mkdir()
            for length in [29, 63, 127, 255]:
                name = 'f' * (length - 4) + '.txt'
                path = directory / name
                path.write_bytes(b'long filename\n')
                wire_path = '/os' + str(path)
                assert name in client.list('/os' + str(directory))
                assert client.read(wire_path) == b'long filename\n'
                client.write(wire_path, b'written through long filename\n', truncate=True)
                assert path.read_bytes() == b'written through long filename\n'
                opened = look(client, '/n/os' + str(path))
                assert client.read(f'/pane/{opened}/body') == path.read_bytes()
                client.remove(f'/pane/{opened}')  # a long name's tag takes rows
            opened = look(client, '/n/os' + str(directory))
            listing = client.read(f'/pane/{opened}/body')
            for path in directory.iterdir():
                assert ('/n/os' + str(path)).encode() in listing
    if quic:
        quic_test(binary)
    print(f'9P integration: passed ({time.monotonic() - started:.3f}s)')


def quic_test(binary):
    for tty in [False, True]:
        for host in ['127.0.0.1', '::1']:
            with tempfile.TemporaryDirectory(prefix='pardes-quic-fs-') as directory:
                root = Path(directory)
                ports = {}
                with session(binary, root, 'server', '--9p-tcp=tcp!' + host + '!0',
                             '--9p-quic=quic!' + host + '!0', tty=tty, network=ports) as (unix, address):
                    assert set(unix.read('/listeners').decode().splitlines()) == {
                        'unix!' + str(address), 'tcp!' + host + '!' + str(ports['TCP']),
                        'quic!' + host + '!' + str(ports['QUIC'])}
                    with Client((host, ports['TCP'])) as tcp:
                        unix_body = unix.open('/pane/1/body')
                        tcp_body = tcp.open('/pane/1/body')
                        with session(binary, root, 'client') as (remote, _):
                            control = new_pane(remote, b'')
                            dial = 'quic!' + host + '!' + str(ports['QUIC'])
                            execute(remote, control, 'Mount peer ' + dial)
                            opened = look(remote, '/n/peer/pane/1/body', source=control)
                            body = f'/pane/{opened}/body'
                            assert remote.read(body) == b'initial\n'
                            payload = 'saved café λ\n'.encode() + b'x' * (3 * 8192 + 27) + b'\n'
                            remote.write(body, payload, truncate=True)
                            remote.write(f'/pane/{opened}/exec', b'Save\n')
                            assert unix.read_fid(unix_body) == payload[:unix.msize - 11]
                            assert tcp.read('/pane/1/body') == payload
                            tcp.write('/pane/1/body', b'TCP to QUIC Get\n', truncate=True)
                            remote.write(f'/pane/{opened}/ctl', b'get\n')
                            assert remote.read(body) == b'TCP to QUIC Get\n'
                            assert unix.read_fid(unix_body) == b'TCP to QUIC Get\n'
                            assert tcp.read_fid(tcp_body) == b'TCP to QUIC Get\n'
                            screen_pane = look(remote, '/n/peer/screen', source=control)
                            screen = json.loads(remote.read(f'/pane/{screen_pane}/body'))
                            assert len(screen['cells']) == screen['cols'] * screen['rows']
                            # The peer's own screen, whole, over QUIC. Its pane's
                            # tag, not its text: where the view sits after the
                            # body shrank from a long wrapped line depends on
                            # whether a frame was drawn meanwhile (sometimes
                            # on the empty last line, `     2`, text off-screen).
                            assert 'server.txt Save' in ''.join(cell[0] for cell in screen['cells'])
                            large = ('normal λ line\n' * 20000).encode()
                            fixture = root / 'large.txt'
                            fixture.write_bytes(large)
                            large_pane = look(remote, '/n/peer/os' + str(fixture), source=control)
                            assert remote.read(f'/pane/{large_pane}/body') == large
                            edited = b'QUIC file saved\n'
                            remote.write(f'/pane/{large_pane}/body', edited, truncate=True)
                            remote.write(f'/pane/{large_pane}/exec', b'Save\n')
                            assert fixture.read_bytes() == edited
                            for pane in [opened, screen_pane, large_pane]:
                                remote.write(f'/pane/{pane}/exec', b'Del\n')
                            execute(remote, control, 'Unmount peer')
                            roots = look(remote, '/n', source=control)
                            assert remote.read(f'/pane/{roots}/body') == b'/n/os/\n/n/self/\n'
                            assert unix.read_fid(unix_body) == b'TCP to QUIC Get\n'
                            assert tcp.read_fid(tcp_body) == b'TCP to QUIC Get\n'
                        # Not checked: a session Looking at its OWN tree through
                        # a QUIC mount ('Mount own quic!...'). QUIC is still
                        # answered on the editor's thread (cloud9's QUIC adapter
                        # is not std.Io based), so that Look waits on the loop
                        # it is running in and opens nothing; features.txt lists
                        # it under "Left as they are". Unix and TCP self-mounts
                        # are covered by test/selfmount.py.
                        unix.close(unix_body)
                        tcp.close(tcp_body)


if __name__ == '__main__':
    if len(sys.argv) in [3, 4] and sys.argv[2] == '--discovery':
        discovery(str(Path(sys.argv[1]).resolve()), embedded=sys.argv[3:] == ['--embedded'])
        raise SystemExit(0)
    if len(sys.argv) not in [2, 3] or (len(sys.argv) == 3 and sys.argv[2] != '--quic'):
        raise SystemExit('usage: fs.py <pardes> [--quic | --discovery]')
    test(str(Path(sys.argv[1]).resolve()), quic=len(sys.argv) == 3)