#!/usr/bin/env python3
"""Instant-budget census: key press or click to the first changed pixel, for every row of the budget.

  flea-field-instant budget                                      the journey table as JSON, the data a gate reads
  flea-field-instant fixtures ROOT [--usb DIR] [--nas DIR]       build every fixture in marked sandbox roots
  flea-field-instant run PART --root ROOT --flea TREE --out DIR [--samples N] [--dry] [--only REGEX]   one census part
  flea-field-instant measure PLAN.json                           the same from a plan file: {"part", "root", "flea", "out", "samples", "only"}
  flea-field-instant report JSONL... [--out FILE] [--calibrate FILE]   the census table in Markdown
  flea-field-instant gate JSONL... [--out FILE] [--part PART]... [--only REGEX]   pass or fail against the budget ceilings
  flea-field-instant calibrate [--runs N]                        negative control, 0, 50 and 200 ms, the key repeat count, the fling floor
  flea-field-instant selftest                                    detector, budget, held-sweep and report logic, no display
  flea-field-instant clean ROOT                                  remove the fixture roots ROOT/fixtures.json names

FLEA_INSTANT_PARENT (default /home/flea-sandbox) is where ROOT must live; FLEA_INSTANT_SMALL=1 shrinks the two big
files to a few MiB, for a builder smoke away from minipc.

Parts: smoke, p1-cursor, p2-held, p3-navigate, p4-open, p5-heavy, p6-ops, p7-load, p8-rename (PARTS below).

The method is tools/flea-field-nav's, which is tools/flea-field-search's: tools/flea-framecap records which presented
frames change inside a rectangle, stamped with the compositor's presentation time in CLOCK_MONOTONIC, and every input
(key, button, wheel notch, touchpad stroke) goes to a uinput device stamped in CLOCK_MONOTONIC just before the write.
A sample is the input to the first frame in which the watched region changes with ink (first_ms, the number the
budget judges), beside the first change of any kind (first_any_ms) and the last change before a quiet gap (settled_ms).
A key that changes nothing before the next input is a miss and counts as over every limit, never as a dropped sample.

Ops (row 6). The backend runs one write at a time and refuses a second as an error line, and the status bar draws only the
oldest unacknowledged error, so a paste that starts under a running job or a queued error draws nothing of its own. Every op
sample therefore starts on an idle status bar: statusActivityState (ui/Ipc.qml) is read, Esc dismisses each queued error and
cancels a running transfer, and a bar that is not clear inside 20 s ends the sample with an error row, never a timing. A
paste that does not finish inside its wait (120 s, 900 s for op.copy.long) is cancelled with Esc and given 60 s to leave
the activity list before the next sample. Each op row carries errors_before and errors_after (queued errors), activity_after
(the activities still up), notice_before and notice_after (the status notice before and after the paste), esc_before (Escs
the bar needed before the paste) and refused, true when the paste left more queued errors than it found; the report counts
refused samples beside the misses.
A sample also starts in the folder sample 0 started in: the pane's path, listInFlight, total, cursor and fileClipboard
(ui/Ipc.qml) are read before the take and before the paste, the source and the destination are reached by row name (a symlink
to a folder is listed among the files, after b-src), and a sample that cannot reach them, or whose take holds the wrong
paths, is a harness fault row ("harness fault" in the report), never a refusal or a miss. op.size.recursive runs with the
Properties row shown (menu.hidden ships it hidden).

Fling floor. calibrate plays the census stroke (series_fling's, 40 mm in 120 ms) to tests/search-calib.qml, an idle Qt
client, and reads the stroke's first motion report to the client's first pixel-carrying wheel event (its log, on the wall clock
bridged to CLOCK_MONOTONIC; the client clock floors to a millisecond, so that figure reads up to 1 ms early) and to the
client's first changed frame (the recorder). That is the share of a fling's first_ms that libinput, Hyprland and Qt take
before any Flea code runs. calibrate prints it under "fling" and report prints it beside the fling rows (it reads the
calibrate.json beside its first JSONL, or --calibrate FILE, whose failure to give a floor is said in the note). judge never subtracts it.

Held key (row 2). j is held on a 10k folder for HOLD_REPEATS repeat periods plus half of one, so the release falls in
the middle of a repeat slot. repeats drawn: slots of one repeat period, anchored on the first change after the first
repeat is due, each needing a changed frame. overshoot: Flea's cursor steps between two IPC reads of the cursor index,
one before the press and one after 2 s of quiet, minus the most key events an idle Qt client (tests/search-calib.qml)
received for the same physical hold on the same host (repeat_probe). Qt's repeat timer never fires more often when the
client is busy, so any step past that bound was handled after the release: a backlog cannot pass.

Sample input, one recorder line: "F 660389861257350 660389874933379 32 95 126 300 2400 61" (see tools/flea-framecap.c).
Sample row: {"journey": "nav.list.down", "row": 1, "view": "list", "fixture": "local", "first_ms": 6.1, "settled_ms": 30.2}.
A journey on the USB stick or the NAS share is row 5 whatever its local row; under load it keeps its local row.
Cursor chords are the default preset's (keys.toml [[code]] Down Up Home End PageUp PageDown, [[ctrl]] D U, [[text]]
j k g G); the grid alone steps sideways on the bare arrows and on h and l.
Parts are sessions of fixture, view, region and series from fixtures.json, so --dry checks a part with no display.
--dry also prints, per session, where the pointer goes for a wheel or a stroke and which row an Enter series lands on.

Journey filter. run PART --only REGEX keeps the series with a journey that matches REGEX (re.search over the journey
ids journeys_of lists) and drops a session left with none, so a re-measure of a few journeys costs only those sessions.
A series is kept whole: a cycle that times enter and back runs both. gate JSONL... --only REGEX plans only the
journeys that match, so a filtered run is not reported MISSING for the journeys it left out; rows in the files are
judged as they are. The manifest of a filtered run names its filter under plan.only.

Targets. A wheel notch or a stroke goes to the centre of the left pane in dual: the content centre falls on the one
pixel gap between the two panes (or within a pixel of it), which belongs to neither, so no view received the input.
Enter in a cycle goes to the first folder of the session fixture, found by Home alone with no Down press, because the
ui-state patch pins foldersFirst true and folders sort first, so the first folder is row 0. A window that is not this session's Flea (class com.thisisgm.flea under the launched process)
mapping during a timed sample ends the series with an error row, "an external window opened", never a miss; nothing
that Flea did not start is closed.

Loads (p7-load). The right pane carries one of: a, a fresh hard-link copy of 2,000 cold 1280x720 JPEGs, whose
thumbnails Flea writes for the rows the pane shows and for no others; b, a copy of a large file to the USB stick; c,
the 5,000-entry NAS folder listing. A load is proved per sample, never per series: the probe (a: thumbnail files in the
cache; b: bytes copied; c: the right pane's list request count and whether its listing is in flight, read by the dualState
IPC) is read right before each timed input and right after its quiet, and the row carries load_probe and load_active
(it moved, or the load was busy at the read before). Loads a and c are renewed before every sample, a by replacing the
previous batch of linked pictures in the right folder with a batch of never-thumbnailed ones under names that sort just after its first row
(the folder keeps its size), c by changing the NAS folder's mtime, and the input is sent only after the probe moved since
the renewal; a probe that never moves is an error row, "the load did not start", never a timed row. A renewal waits out
the load's gap since the last one (a 0.3 s, c 1 s). A paced series renews after a key only when that gap has passed, so
at the 400 ms pace c renews on about every third key, and reads the probe before each key. The
gate passes a load group only when every sample's load_active is true.
"""
import json
import math
import os
import re
import shutil
import signal
import socket
import statistics
import subprocess
import sys
import tempfile
import time
from importlib.machinery import SourceFileLoader

HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(HERE)
N = SourceFileLoader("flea_field_nav", os.path.join(HERE, "flea-field-nav")).load_module()
S = N.S
NS, MS = S.NS, S.MS

# The budget: p95 of each series, key press or click to the first changed pixel.

REFRESH_MS = 1000.0 / 120  # minipc DP-2, one frame
BUDGET = {
    1: {"name": "cursor", "target_ms": 8.33, "ceiling_ms": 16.7,
        "rule": "p95 first_ms at or under target, else at or under ceiling"},
    2: {"name": "held key", "target_ms": 25.0, "ceiling_ms": 0,
        "rule": "target: every repeat drawn before the next (no missed slot); ceiling: overshoot 0 rows at release"},
    3: {"name": "navigate", "target_ms": 50, "ceiling_ms": 100, "rule": "p95 first_ms"},
    4: {"name": "open", "target_ms": 50, "ceiling_ms": 100, "rule": "p95 first_ms"},
    5: {"name": "heavy", "target_ms": 100, "ceiling_ms": 100, "content_ceiling_ms": 1000,
        "rule": "target: p95 settled_ms at or under 100; ceiling: p95 first_ms at or under 100 and p95 settled_ms "
                "at or under 1000 (content or progress); a video's settle is its playback, so it is judged on first_ms"},
    6: {"name": "long ops", "target_ms": 50, "ceiling_ms": 100,
        "rule": "p95 first_ms; percent and time left past 10 s read from the 11 s evidence frame"},
}

# Journey id patterns (PERF-PROGRAM-20261005.md), each with its budget row.
JOURNEYS = [
    (r"^nav\.[a-z]+\.(down|up|j|k|pagedown|pageup|ctrl-d|ctrl-u|home|end|g|G|left|right|h|l|down-scroll|wheel|fling)$", 1),
    (r"^nav\.[a-z]+\.(pre|after)-rename-(j|k)$", 1),
    (r"^nav\.[a-z]+\.held-j-sweep$", 2),
    (r"^nav\.[a-z]+\.(enter|back|shift-right|shift-left)$", 3),
    (r"^view\.switch\.[a-z]+-[a-z]+$", 3),
    (r"^tab\.(new|switch)$", 3),
    (r"^pane\.switch\.dual$", 3),
    (r"^open\.(menu\.key|menu\.click|rename|settings|palette|pathjump)$", 4),
    (r"^search\.type-to-first-result$", 4),
    (r"^open\.(ql|cols-preview)\.(txt|jpg|md|code)(\.first)?$", 4),
    (r"^open\.(ql|cols-preview)\.(video|pdf|large-image)(\.first)?$", 5),
    (r"^op\.(copy\.many|copy\.large|copy\.long|move\.cross-fs|trash|size\.recursive|archive)$", 6),
]


def row_of(journey, fixture="local"):
    """The budget row of a journey on a fixture, or None when no pattern names it."""
    for pat, row in JOURNEYS:
        if re.match(pat, journey):
            return 5 if fixture in ("usb", "nas") and row != 2 else row
    return None


def p95(vals):
    """Nearest-rank p95; a miss (None) sorts above every number."""
    if not vals:
        return None
    ordered = sorted(vals, key=lambda v: math.inf if v is None else v)
    return ordered[max(0, math.ceil(0.95 * len(ordered)) - 1)]


def p50(vals):
    if not vals:
        return None
    ordered = sorted(vals, key=lambda v: math.inf if v is None else v)
    return ordered[(len(ordered) - 1) // 2]


def judge(row, firsts, settled=None, video=False, held=None):
    """TARGET, CEILING-OK or BREACH for one series, from the BUDGET row."""
    b = BUDGET[row]
    if row == 2:
        if not held or any(h.get("overshoot") is None for h in held):
            return "NO-READ"
        if any(h["overshoot"] > 0 for h in held):
            return "BREACH"
        return "TARGET" if all(h["slots_missed"] == 0 for h in held) else "CEILING-OK"
    f95 = p95(firsts)
    if f95 is None:
        return "BREACH" if firsts else "NO-READ"  # a miss ranks above every reading; empty stays NO-READ
    if row == 5:
        s95 = None if video else p95(settled or [])
        if f95 > b["ceiling_ms"] or (not video and (s95 is None or s95 > b["content_ceiling_ms"])):
            return "BREACH"
        if video:
            return "TARGET" if f95 <= b["target_ms"] else "CEILING-OK"
        return "TARGET" if s95 <= b["target_ms"] else "CEILING-OK"
    if f95 <= b["target_ms"]:
        return "TARGET"
    return "CEILING-OK" if f95 <= b["ceiling_ms"] else "BREACH"


# Held-key analysis: pure functions over the recorder's (presentation ns, changed rows, ink) frames.


def repeats_due(t_down, t_up, delay_ms, interval_ms):
    """Repeats a client schedules strictly before the release: due at delay, delay + interval, ..."""
    n, t = 0, t_down + delay_ms * MS
    while t < t_up:
        n += 1
        t += interval_ms * MS
    return n


def held_analysis(frames, t_down, t_up, delay_ms, interval_ms):
    """Repeat slots drawn, the largest gap between changed frames in the hold, and the first repeat's latency."""
    changes = S.changes_of(frames)
    n_rep = repeats_due(t_down, t_up, delay_ms, interval_ms)
    press = S.first_change_after(changes, t_down)
    anchor = S.first_change_after(changes, t_down + delay_ms * MS)
    out = {"repeats_due": n_rep, "press_first_ms": None if press is None else round((press - t_down) / MS, 1),
           "first_repeat_ms": None, "slots_missed": n_rep, "max_gap_ms": None}
    if anchor is None or anchor > t_up + interval_ms * MS:
        return out
    out["first_repeat_ms"] = round((anchor - t_down - delay_ms * MS) / MS, 1)
    half = interval_ms * MS / 2
    missed = 0
    for k in range(1, n_rep):
        lo, hi = anchor + k * interval_ms * MS - half, anchor + k * interval_ms * MS + half
        if not any(lo <= t < hi for t in changes):
            missed += 1
    out["slots_missed"] = missed
    inside = [t for t in changes if anchor <= t <= t_up]
    gaps = [(b - a) / MS for a, b in zip(inside, inside[1:])]
    out["max_gap_ms"] = round(max(gaps), 1) if gaps else None
    return out


def overshoot(start_idx, final_idx, stride, max_events):
    """(rows past the idle client's event count, rows short of it); None when a read failed."""
    if start_idx is None or final_idx is None or not stride or max_events is None:
        return None, None
    steps = (final_idx - start_idx) // stride
    return max(0, steps - max_events), max(0, max_events - steps)


# Input beyond keys: pointer, buttons, wheel and a touchpad, all through uinput.

N.EXTRA_KEYCODES.update({"pageup": 104, "pagedown": 109, "f2": 60, "delete": 111, "[": 26, "]": 27, ";": 39,
                         "rightctrl": 97, "super": 125})
EV_REL, EV_SYN = 2, 0
REL_X, REL_Y, REL_WHEEL = 0, 1, 8
BTN_LEFT, BTN_RIGHT, BTN_MIDDLE = 0x110, 0x111, 0x112
MODIFIERS = (29, 42, 56, 97, 125)


def release_all(keys):
    """Every button and modifier up: a held one left by an earlier run makes the kernel drop the next press."""
    for code in (BTN_LEFT, BTN_RIGHT, BTN_MIDDLE) + MODIFIERS:
        keys._key(code, 0)


def rel(keys, code, value):
    keys._event(EV_REL, code, value)
    keys._event(EV_SYN, 0, 0)


def pointer_to(keys, x, y):
    """Warp to an output point, then a one pixel uinput wiggle so the client gets a motion with its frame."""
    S.dispatch("hl.dsp.cursor.move({ x = %d, y = %d })" % (x, y))
    time.sleep(0.05)
    rel(keys, REL_X, 1)
    time.sleep(0.02)
    rel(keys, REL_X, -1)
    time.sleep(0.05)
    # Sample input, hyprctl cursorpos: "1280, 720"
    pos = S.hypr("cursorpos").strip()
    px, py = (int(float(v)) for v in pos.split(","))
    if abs(px - x) > 3 or abs(py - y) > 3:
        raise RuntimeError("pointer at %s, wanted %d,%d" % (pos, x, y))


def click(keys, button):
    """One press and release back to back; the CLOCK_MONOTONIC ns just before the press."""
    t = time.monotonic_ns()
    keys._key(button, 1)
    keys._key(button, 0)
    return t


def wheel(keys, notches):
    """One wheel report, negative scrolls down; the CLOCK_MONOTONIC ns before the write."""
    t = time.monotonic_ns()
    rel(keys, REL_WHEEL, notches)
    return t


def touchpad():
    """tools/flea-touchpad as a module, or None with the reason when /dev/uinput is not writable."""
    T = SourceFileLoader("flea_touchpad", os.path.join(HERE, "flea-touchpad")).load_module()
    if not os.access(T.UINPUT, os.W_OK):
        return None, "%s is not writable, so no touchpad can be created" % T.UINPUT
    return T, None


def stroke(T, fd, dy_mm, ms):
    """A two-finger stroke released with no hold (a fling); the CLOCK_MONOTONIC ns of its first motion report."""
    dy = int(dy_mm * T.RES)
    x0 = y0 = T.SPAN // 2
    steps = max(2, ms // T.EVENT_STEP_MS)
    T.finger(fd, 0, 1, x0, y0)
    T.finger(fd, 1, 2, x0, y0)
    T.emit(fd, T.EV_KEY, T.BTN_TOUCH, 1)
    T.emit(fd, T.EV_KEY, T.BTN_TOOL_FINGER, 0)
    T.emit(fd, T.EV_KEY, T.BTN_TOOL_DOUBLETAP, 1)
    T.report(fd)
    t_first = None
    for step in range(1, steps + 1):
        y = y0 + dy * step // steps
        T.finger(fd, 0, 1, x0, y)
        T.finger(fd, 1, 2, x0, y)
        if t_first is None:
            t_first = time.monotonic_ns()
        T.report(fd)
        time.sleep(ms / 1000 / steps)
    T.finger(fd, 0, -1, x0, y0 + dy)
    T.finger(fd, 1, -1, x0, y0 + dy)
    T.emit(fd, T.EV_KEY, T.BTN_TOUCH, 0)
    T.emit(fd, T.EV_KEY, T.BTN_TOOL_DOUBLETAP, 0)
    T.report(fd)
    return t_first


def drop_touchpad(T, fd):
    """Destroy the uinput touchpad fd names and close it."""
    import fcntl
    try:
        fcntl.ioctl(fd, T.UI_DEV_DESTROY)
    except OSError:
        pass
    os.close(fd)


# Recorder with a dump on the first change after an armed input, keeping the first drawn state.


class EvRecorder(S.Recorder):
    armed_ns = None

    # Sample input, one recorder line: "F 660389861257350 660389874933379 32 95 126 300 2400 61" (see tools/flea-framecap.c).
    def _read(self):
        for line in self.proc.stdout:
            f = line.split()
            with self.lock:
                if f and f[0] == "F" and len(f) in (8, 9):
                    self.frames.append((int(f[1]), int(f[3]), int(f[8]) if len(f) == 9 else -1))
                    self.detail.append(tuple(int(v) for v in f[1:]))
                    if self.armed_ns is not None and int(f[3]) > 0 and int(f[1]) > self.armed_ns:
                        self.armed_ns = None
                        self.proc.send_signal(signal.SIGUSR1)
                elif f and f[0] == "H":
                    self.header = f[1:]
                elif f and f[0] == "D":
                    self.dumps.append(f[1])


# Sandbox: every root the run writes into carries the marker, and nothing outside a marked root is ever removed.

MARKER = ".flea-test-sandbox"
LOCAL_PARENT = os.environ.get("FLEA_INSTANT_PARENT", "/home/flea-sandbox")  # where ROOT must live; a smoke points it at /tmp
MIN_ROOT_LEN = len("/home/x/y")  # a marked root must run deeper than this path


def marked(root):
    return os.path.isfile(os.path.join(root, MARKER))


def inside_marked(path, roots):
    """True when path is strictly inside one of roots and that root carries the marker."""
    real = os.path.realpath(path) if not os.path.islink(path) else os.path.join(
        os.path.realpath(os.path.dirname(path)), os.path.basename(path))
    for r in roots:
        rr = os.path.realpath(r)
        if real.startswith(rr + "/") and marked(rr) and len(rr) > MIN_ROOT_LEN:
            return True
    return False


def sandbox_rm(path, roots):
    """Remove path (file, link or tree) only when it is inside a marked root; refuse anything else loudly."""
    if not os.path.isabs(path) or not path.strip("/"):
        raise SystemExit("refusing to remove a relative or empty path: %r" % path)
    if not inside_marked(path, roots):
        raise SystemExit("refusing to remove %s: not inside a marked sandbox root" % path)
    if os.path.islink(path) or os.path.isfile(path):
        os.unlink(path)
    elif os.path.isdir(path):
        shutil.rmtree(path)


def empty_dir(path, roots):
    for name in os.listdir(path):
        sandbox_rm(os.path.join(path, name), roots)


def reflink(src, dst):
    subprocess.run(["cp", "-r", "--reflink=auto", src, dst], check=True)


# Fixtures.

CODE_TEXT = '''"""A small module the code preview renders."""


def walk(root, depth=0):
    for name in sorted(root):
        yield depth, name


class Tree:
    def __init__(self, names):
        self.names = list(names)

    def count(self):
        return len(self.names)
'''
KINDS = {"jpg": "a_photo.jpg", "pdf": "c_doc.pdf", "md": "d_notes.md", "txt": "e_text.txt", "code": "g_code.py",
         "large-image": "h_large.jpg", "video": "i_hd.mp4"}
REMOTE_KINDS = ("jpg", "pdf", "md", "txt")  # what the USB and NAS preview folders hold
COPY_MANY = 1000  # files of op.copy.many, 4 KiB each
MOVE_FILES = 200  # files of op.move.cross-fs, 64 KiB each
TREE_FILES = 20000  # files under op.size.recursive's folder, 100 per directory
ARCHIVE_FILES = 200  # files of op.archive's source folder, 16 KiB each
LARGE_MIB = 1024  # op.copy.large
LONG_MIB = 4000  # op.copy.long, past 10 s onto the USB stick and under vfat's 4 GiB - 1 byte file limit
if os.environ.get("FLEA_INSTANT_SMALL"):  # a builder smoke off minipc: the same layout at a few MiB
    LARGE_MIB, LONG_MIB = 4, 8
LOAD_ROWS = 3000  # rows of the load folder besides its folder, its text file and its big file
COLD_IMAGES = 2000  # images of the cold thumbnail folder
NAS_STREAM = 5000  # entries of the NAS folder a load-c listing streams


def run(cmd):
    subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL)


def files(d, n, fmt, size):
    os.makedirs(d, exist_ok=True)
    blob = os.urandom(size) if size else b""
    for i in range(n):
        with open(os.path.join(d, fmt % i), "wb") as f:
            f.write(blob)


def random_file(path, mib):
    with open(path, "wb") as f:
        for _ in range(mib):
            f.write(os.urandom(1 << 20))


def new_root(parent):
    """A fresh mktemp -d under parent with the marker; the shared flea-perf bench fixture is refused."""
    if "flea-perf" in os.path.realpath(parent):
        raise SystemExit("refusing %s: the flea-perf bench fixture is shared and never written" % parent)
    root = tempfile.mkdtemp(prefix="flea-instant-", dir=parent)
    open(os.path.join(root, MARKER), "w").close()
    return root


def remote_tree(root):
    files(os.path.join(root, "enter", "big"), 1000, "g_%04d.txt", 0)
    prev = os.path.join(root, "preview")
    os.makedirs(prev, exist_ok=True)
    return prev


def build_fixtures(root, usb=None, nas=None):
    """Everything a census reads, under ROOT (a mktemp -d under /home/flea-sandbox) and the stick and share roots."""
    real = os.path.realpath(root)
    if not real.startswith(LOCAL_PARENT + "/"):
        raise SystemExit("ROOT must be a mktemp -d under %s, not %s" % (LOCAL_PARENT, root))
    if os.listdir(real) and not marked(real):
        raise SystemExit("refusing to fill %s: not empty and carries no %s" % (real, MARKER))
    open(os.path.join(real, MARKER), "w").close()
    for d in ("home", "data", "cache", "states", "ops", "load"):
        os.makedirs(os.path.join(real, d), exist_ok=True)
    # Read-only views of the operator's fonts and icons, so the sandboxed data home draws the same glyphs.
    for name in ("fonts", "icons"):
        src = os.path.expanduser("~/.local/share/" + name)
        if os.path.isdir(src) and not os.path.lexists(os.path.join(real, "data", name)):
            os.symlink(src, os.path.join(real, "data", name))
    nav = os.path.join(real, "nav")
    builder = SourceFileLoader("flea_nav_fixture", os.path.join(HERE, "flea-nav-fixture")).load_module()
    facts = {"root": real, "nav": builder.build(nav)}
    prev = os.path.join(nav, "preview")
    with open(os.path.join(prev, KINDS["code"]), "w") as f:
        f.write(CODE_TEXT)
    magick = shutil.which("magick") or shutil.which("convert")
    gaps = list(facts["nav"].get("gaps", []))
    if magick:
        run([magick, "-size", "6000x4000", "xc:gray", "+noise", "Random", "-quality", "85",
             os.path.join(prev, KINDS["large-image"])])
    else:
        gaps.append("large-image skipped: no ImageMagick")
    if shutil.which("ffmpeg"):
        run(["ffmpeg", "-y", "-v", "error", "-f", "lavfi", "-i", "testsrc2=duration=5:size=1920x1080:rate=30",
             "-pix_fmt", "yuv420p", os.path.join(prev, KINDS["video"])])
    else:
        gaps.append("video skipped: no ffmpeg")
    ops = os.path.join(real, "ops")
    files(os.path.join(ops, "src-many"), COPY_MANY, "m_%04d.dat", 4096)
    files(os.path.join(ops, "src-move"), MOVE_FILES, "v_%03d.dat", 65536)
    files(os.path.join(ops, "src-trash"), 50, "t_%02d.txt", 1024)
    files(os.path.join(ops, "src-archive", "src-dir"), ARCHIVE_FILES, "a_%03d.dat", 16384)
    for d in range(TREE_FILES // 100):
        files(os.path.join(ops, "size", "tree", "d%02d" % (d // 20), "e%03d" % d), 100, "s_%03d.txt", 512)
    random_file(os.path.join(ops, "large.bin"), LARGE_MIB)
    random_file(os.path.join(ops, "long.bin"), LONG_MIB)
    load = os.path.join(real, "load", "L")
    files(os.path.join(load, "a_dir"), 1000, "g_%04d.txt", 0)
    with open(os.path.join(load, "b_text.txt"), "w") as f:
        f.write("plain text for the load journeys\n" * 40)
    files(load, LOAD_ROWS, "f_%04d.txt", 0)
    reflink(os.path.join(ops, "long.bin"), os.path.join(load, "zzz_big.bin"))
    cold = os.path.join(real, "load", "cold-src")
    os.makedirs(cold, exist_ok=True)
    if shutil.which("ffmpeg"):
        run(["ffmpeg", "-y", "-v", "error", "-f", "lavfi", "-i", "testsrc2=size=1280x720:rate=30",
             "-frames:v", str(COLD_IMAGES), os.path.join(cold, "c_%05d.jpg")])
    else:
        gaps.append("cold images skipped: no ffmpeg")
    for name, parent in (("usb", usb), ("nas", nas)):
        if not parent:
            continue
        r = new_root(parent)
        rp = remote_tree(r)
        for kind in REMOTE_KINDS:
            if os.path.exists(os.path.join(prev, KINDS[kind])):
                shutil.copyfile(os.path.join(prev, KINDS[kind]), os.path.join(rp, KINDS[kind]))
            else:
                gaps.append("%s %s skipped: not built locally" % (name, kind))
        os.makedirs(os.path.join(r, "dst"), exist_ok=True)
        if name == "nas":
            files(os.path.join(r, "stream"), NAS_STREAM, "n_%05d.txt", 0)
        facts[name] = r
    facts["preview"] = sorted(n for n in os.listdir(prev) if not n.startswith("."))
    facts["gaps"] = gaps
    with open(os.path.join(real, "fixtures.json"), "w") as f:
        json.dump(facts, f, indent=1)
    return facts


def drop_root(path):
    """Remove one fixture root: absolute, not a link, at least three levels deep, and carrying the marker."""
    if not path or not os.path.isabs(path) or os.path.realpath(path) != path or path.rstrip("/").count("/") < 3:
        raise SystemExit("refusing to remove %r: not an absolute, real path three levels deep" % path)
    if not marked(path):
        raise SystemExit("refusing to remove %s: carries no %s" % (path, MARKER))
    shutil.rmtree(path)
    print("removed", path)


def clean(root):
    """Remove the local root and the stick and share roots its fixtures.json names, each only if marked."""
    real = os.path.realpath(root)
    with open(os.path.join(real, "fixtures.json")) as f:
        facts = json.load(f)
    for key in ("usb", "nas"):
        r = facts.get(key)
        if r and os.path.basename(r).startswith("flea-instant-"):
            drop_root(r)
    if not real.startswith(LOCAL_PARENT + "/"):
        print("refusing to remove local root %s: not under %s" % (real, LOCAL_PARENT), file=sys.stderr)
        return 1
    drop_root(real)
    return 0


# A session: one Flea window on one fixture and view, a recorder over one region, and the IPC seam for setup reads.

QUIET = {1: 1.0, 2: 2.0, 3: 1.0, 4: 1.0, 5: 2.0, 6: 2.0}  # s of quiet that ends a sample, per budget row
CAP = {1: 3, 2: 15, 3: 3, 4: 3, 5: 8, 6: 30}  # s after the input a sample may keep changing
PACE_MS = N.CURSOR_PACE_MS  # cursor series, a steady hand pace
HOLD_REPEATS = 70  # repeat periods a held sweep spans after the delay
CHROME_TOP = 56  # Flea's content region margins, the ones tools/flea-field-search measured
CHROME_BOTTOM = 34
RAIL = 192
PANE_SPLIT = 0.6  # share of the content width kept left of the preview column


def window_at(address):
    """The mapped client at address, or a RuntimeError when it has gone, so a window lost mid-setup is an error row."""
    for c in S.clients():
        if c.get("address") == address:
            return c
    raise RuntimeError("the window %s closed before its geometry was read" % address)


def win_rect(win, kind):
    """A watched rectangle (x, y, w, h) of a window (x, y, w, h): the whole window, the preview column, dual's left pane, else the content."""
    wx, wy, ww, wh = win
    content = (wx + RAIL, wy + CHROME_TOP, ww - RAIL, wh - CHROME_TOP - CHROME_BOTTOM)
    if kind == "window":
        return (wx, wy, ww, wh)
    if kind == "pane":  # the preview column, the right 40 percent of the content region
        cx, cy, cw, ch = content
        return (cx + int(cw * PANE_SPLIT), cy, cw - int(cw * PANE_SPLIT), ch)
    if kind == "left":  # the left pane of dual
        cx, cy, cw, ch = content
        return (cx, cy, cw // 2, ch)
    return content


def scroll_point(view, region, left):
    """The pointer point of a wheel or stroke: dual's left pane centre, because the content centre is the hairline no pane owns."""
    x, y, w, h = left if view == "dual" else region
    return x + w // 2, y + h // 2


def external_clients(clients, owned):
    """The mapped clients that are not Flea's: Flea's are of class S.FLEA_CLASS and owned by a pid under the launched process."""
    return [c for c in clients if c.get("mapped", True) and not (c.get("class") == S.FLEA_CLASS and c.get("pid") in owned)]


class ExternalWindow(Exception):
    """A window that is not this session's Flea mapped during a sample; row is the sample's row, with its error."""

    def __init__(self, row):
        super().__init__(row["error"])
        self.row = row


class Session:
    def __init__(self, ctx, fixture, view, region="content", right=None, log=None, ui=None):
        self.ctx, self.view, self.fixture = ctx, view, fixture
        self.keys = ctx["keys"]
        release_all(self.keys)
        self.patch, self.taken = dict(ui_patch(view, fixture, right), **(ui or {})), []
        state = flea_state(ctx, self.patch)
        tree = ctx["tree"]
        env = dict(S.ENV, FLEA_BIN=tree + "/target/release/flea", FLEA_UI=tree + "/ui", XDG_STATE_HOME=state,
                   QML_DISK_CACHE_PATH=tree + "/target/qmlcache", HOME=ctx["sbx"] + "/home",
                   XDG_DATA_HOME=ctx["sbx"] + "/data", XDG_CACHE_HOME=ctx["sbx"] + "/cache",
                   XDG_CONFIG_HOME=ctx["config_home"])
        os.makedirs(env["QML_DISK_CACHE_PATH"], exist_ok=True)
        self.log = open(log or os.devnull, "a")
        self.proc = subprocess.Popen([env["FLEA_BIN"], "--gui", fixture], env=env, stdout=self.log,
                                     stderr=subprocess.STDOUT, start_new_session=True)
        self.rec = None
        try:
            t0 = time.monotonic()
            win = None
            while win is None and time.monotonic() - t0 < 60:
                owned = S.descendants(self.proc.pid)
                win = next((c for c in S.clients() if c.get("class") == S.FLEA_CLASS and c.get("pid") in owned), None)
                time.sleep(0.05)
            if win is None:
                raise RuntimeError("no Flea window mapped in 60 s")
            self.address = win["address"]
            S.focus_window(self.address)
            S.park_cursor()
            time.sleep(0.5)
            win = window_at(self.address)
            (wx, wy), (ww, wh) = win["at"], win["size"]
            self.win = (wx, wy, ww, wh)
            self.known = set(self.outsiders())
            self.qs_pid = qs_pid(ctx, self.proc)
            self.set_region(region)
        except BaseException:
            self.close()
            raise

    def rect(self, kind):
        return win_rect(self.win, kind)

    def set_region(self, kind):
        if self.rec:
            self.rec.stop()
        self.region = self.rect(kind)
        dump_dir = os.path.join(self.ctx["out_dir"], "dumps")
        os.makedirs(dump_dir, exist_ok=True)
        self.rec = EvRecorder(self.ctx["rec_bin"], self.region, (), os.path.join(dump_dir, "d"), self.win)
        self.rec.wait_quiet(int(1.5 * NS), time.monotonic_ns(), 30 * NS)

    def ipc(self, fn, *args):
        out = subprocess.run(["qs", "ipc", "--pid", str(self.qs_pid), "call", "flea", fn] + [str(a) for a in args],
                             env=S.ENV, capture_output=True, text=True, timeout=10)
        return out.stdout.strip()

    def ipc_int(self, fn, *args):
        try:
            return int(self.ipc(fn, *args))
        except ValueError:
            return None

    def point(self, target):
        """Output coordinates of an IPC target: ("chrome", glyph) or ("row", index)."""
        kind, arg = target
        text = self.ipc("chromeButtonCentre" if kind == "chrome" else "rowCentre", arg)
        if not text:
            raise RuntimeError("IPC gave no centre for %s %s" % target)
        # Sample input, the IPC centre text: "412 96"
        x, y = (int(v) for v in text.split())
        return self.win[0] + x, self.win[1] + y

    def send(self, inp):
        """One untimed or timed input; the CLOCK_MONOTONIC ns before its write."""
        kind = inp[0]
        if kind == "key":
            return self.keys.chord(*inp[1])
        if kind in ("click", "rclick"):
            x, y = self.point(inp[1])
            pointer_to(self.keys, x, y)
            self.rec.wait_quiet(int(0.5 * NS), time.monotonic_ns(), 3 * NS)
            return click(self.keys, BTN_LEFT if kind == "click" else BTN_RIGHT)
        raise ValueError(inp)

    def settle(self, quiet_s, cap_s, since=None):
        return self.rec.wait_quiet(int(quiet_s * NS), since or time.monotonic_ns(), int(cap_s * NS))

    def outsiders(self):
        """The addresses of the mapped windows that are not this session's Flea."""
        owned = S.descendants(self.proc.pid)
        return {c["address"]: c for c in external_clients(S.clients(), owned)}

    load = None  # the Load beside the series, or None
    load_reads = None  # the probe read before each paced input

    def load_read(self):
        return self.load.probe() if self.load else None

    def load_ready(self):
        """The probe read that shows a renewed load moving (a load that cannot be renewed: its read now), or None with no load; a probe that never moves raises."""
        ld = self.load
        if ld is None:
            return None
        if ld.renew_fn is None:
            return ld.probe()
        base = ld.probe()
        ld.renew(True)
        t0 = time.monotonic()
        while True:
            now = ld.probe()
            if load_moved(base, now):
                return now
            if time.monotonic() - t0 > LOAD_START_S:
                raise RuntimeError("the load did not start")
            time.sleep(LOAD_POLL_S)

    def sample(self, journey, inp, row, evidence=None, extra=None):
        """One timed input and its columns (pre_quiet: the region was quiet before it), with the load's probe read before it and after its quiet; a window mapped since the baseline that is not Flea's raises ExternalWindow."""
        quiet_s, cap_s = QUIET[row], CAP[row]
        if inp[0] == "key":
            S.hide_cursor(self.keys, 0)
        pre = self.settle(0.3, 2)
        if inp[0] in ("click", "rclick"):
            x, y = self.point(inp[1])
            pointer_to(self.keys, x, y)
            pre = self.settle(0.5, 3)
            p_in = self.load_ready()
            t = click(self.keys, BTN_LEFT if inp[0] == "click" else BTN_RIGHT)
        else:
            p_in = self.load_ready()
            t = self.send(inp)
        if evidence:
            self.rec.armed_ns = t
        self.rec.wait_quiet(int(quiet_s * NS), t, int(cap_s * NS))
        p_out = self.load_read()
        out = self.columns(journey, t, row, pre, evidence, extra)
        if self.load:
            out.update(load_evidence(p_in, p_out))
        opened = {a: c for a, c in self.outsiders().items() if a not in self.known}
        if opened:
            out.update(error="an external window opened",
                       external=[{"class": c.get("class"), "title": c.get("title"), "pid": c.get("pid")}
                                 for c in opened.values()])
            raise ExternalWindow(out)
        return out

    def columns(self, journey, t, row, pre=True, evidence=None, extra=None, quiet_s=None, cap_s=None):
        quiet_ns, cap_ns = int((quiet_s or QUIET[row]) * NS), int((cap_s or CAP[row]) * NS)
        frames = self.rec.snapshot()
        cols = S.search_columns(frames, [t], time.monotonic_ns(), quiet_ns, cap_ns)
        any_first = S.first_change_after(S.changes_of(frames), t)
        out = {"journey": journey, "row": row, "view": self.view, "first_ms": cols["first_ms"],
               "first_any_ms": None if any_first is None else round((any_first - t) / MS, 1),
               "settled_ms": cols["settled_ms"], "capped": cols["capped"], "pre_quiet": pre,
               "blank_frames_skipped": cols["blank_frames_skipped"]}
        if evidence:
            out["evidence"] = self.keep_dump(evidence)
        if extra:
            out.update(extra)
        return out

    def keep_dump(self, stem):
        """The newest dump as a PNG with the watched region outlined, or None."""
        t0 = time.monotonic()
        while not self.rec.dumps and time.monotonic() - t0 < 2:
            time.sleep(0.05)
        if not self.rec.dumps:
            return None
        ppm = self.rec.dumps.pop()
        png = os.path.join(self.ctx["out_dir"], stem + ".png")
        S.ppm_to_png(ppm, png, self.rec.boxes(self.region, ()))
        return png

    def require_window(self):
        if not any(c.get("address") == self.address for c in S.clients()):
            raise RuntimeError("the Flea window was gone before the series ended")

    def close(self):
        if self.rec:
            self.rec.stop()
        N.end_proc(self.proc)
        release_all(self.keys)
        self.log.close()


# The shipped menu.hidden (src/uischema.rs DEFAULTS) hides Properties; op.size.recursive's session shows it and hides the rest.
MENU_HIDDEN = ["delete", "openTerminal", "placeMenu", "runScript", "moveto", "copyto", "properties", "permissions",
               "copyAs", "pasteAs", "invertSelection", "extThumbs", "run", "openAsText"]
SIZE_UI = {"menu": {"hidden": [a for a in MENU_HIDDEN if a != "properties"]}}


def ui_patch(view, fixture, right=None):
    """The ui-state patch of a session: name sort, folders first and the view; dual names both pane paths."""
    patch = {"sort": {"key": "name", "reverse": False}, "foldersFirst": True, "view": view}
    if view == "dual":
        patch["dual"] = {"paths": [fixture, right or fixture], "focus": 0}
    return patch


def flea_state(ctx, patch):
    """A state directory with Flea's shipped defaults and the patch applied."""
    key = json.dumps(patch, sort_keys=True)
    if key not in ctx["states"]:
        d = tempfile.mkdtemp(prefix="state-", dir=os.path.join(ctx["sbx"], "states"))
        os.makedirs(d + "/no-config", exist_ok=True)
        subprocess.run([ctx["tree"] + "/target/release/flea", "--ui-state", json.dumps(patch)],
                       env=dict(S.ENV, XDG_STATE_HOME=d, XDG_CONFIG_HOME=d + "/no-config"),
                       capture_output=True, check=True)
        ctx["states"][key] = d
    return ctx["states"][key]


def qs_pid(ctx, proc):
    """The pid of the one qs running this tree's ui under the launched process, the only exact IPC selector."""
    want = ctx["tree"] + "/ui/boot/shell.qml"
    t0 = time.monotonic()
    while time.monotonic() - t0 < 10:
        out = subprocess.run(["qs", "list", "--all"], env=S.ENV, capture_output=True, text=True, timeout=10).stdout
        pid, found = None, []
        # Sample input: "  Process ID: 1318508" (qs list --all indents two spaces).
        for line in out.splitlines():
            line = line.strip()
            if line.startswith("Instance "):
                pid = None
            elif line.startswith("Process ID:"):
                pid = int(line.split(":", 1)[1])
            elif line.startswith("Config path:") and line.split(":", 1)[1].strip() == want and pid:
                found.append(pid)
        owned = S.descendants(proc.pid)
        mine = [p for p in found if p in owned]
        if len(mine) == 1:
            return mine[0]
        time.sleep(0.2)
    raise RuntimeError("no single qs instance of %s under pid %d" % (want, proc.pid))


# Series. Each takes a session and a spec and returns rows; a spec is plain data so the parts below stay a table.

CHORD_NAMES = {("down",): "down", ("up",): "up", ("j",): "j", ("k",): "k", ("pagedown",): "pagedown",
               ("pageup",): "pageup", ("ctrl", "d"): "ctrl-d", ("ctrl", "u"): "ctrl-u", ("home",): "home",
               ("end",): "end", ("g",): "g", ("shift", "g"): "G", ("left",): "left", ("right",): "right",
               ("h",): "h", ("l",): "l"}
# Cursor-moving chords of the default preset; each pair leaves row 0, then returns.
PAIRS = [(("down",), ("up",)), (("j",), ("k",)), (("pagedown",), ("pageup",)), (("ctrl", "d"), ("ctrl", "u")),
         (("end",), ("home",)), (("shift", "g"), ("g",))]
# The grid alone steps sideways on the bare arrows and on h and l (ui/js/Focus.js lookup, ui/js/Grid.js).
GRID_PAIRS = [(("right",), ("left",)), (("l",), ("h",))]
VIEW_KEYS = {"list": ("ctrl", "1"), "columns": ("ctrl", "2"), "grid": ("ctrl", "3")}  # keys.toml preset default
# An Euler circuit of the 12 ordered view pairs; dual has no chord, so it is reached by its chrome button.
VIEW_CIRCUIT = ["list", "columns", "grid", "dual", "list", "grid", "columns", "dual", "grid", "list", "dual",
                "columns", "list"]


def paced(sess, chords, pace_ms, times=None):
    """Chords on an absolute schedule; their CLOCK_MONOTONIC ns, in times as each is sent, the load probe read before each in sess.load_reads, the load renewed after each when its gap has passed."""
    times = [] if times is None else times
    reads = sess.load_reads = []
    first = sess.load_ready()
    start = time.monotonic_ns()
    for i, c in enumerate(chords):
        due = start + int(i * pace_ms * MS)
        while time.monotonic_ns() < due - 2 * MS:
            time.sleep(0.001)
        while time.monotonic_ns() < due:
            pass
        reads.append(first if i == 0 else sess.load_read())
        times.append(c() if callable(c) else sess.keys.chord(*c))
        if sess.load:
            sess.load.renew()
    return times


def per_key_rows(sess, times, names, extra=None):
    """One row per input from the per-key first change; names labels each input's journey."""
    sess.settle(N.NAV_QUIET_NS / NS, 15, times[-1])
    frames = sess.rec.snapshot()
    end = time.monotonic_ns()
    cols = S.search_columns(frames, times, end, N.NAV_QUIET_NS, N.NAV_CAP_NS)
    changes = S.changes_of(frames)
    reads = sess.load_reads[:len(times)] if sess.load else []
    ends = reads[1:] + [sess.load_read()]
    rows = []
    for i, (t, name) in enumerate(zip(times, names)):
        nxt = times[i + 1] if i + 1 < len(times) else end
        a = S.first_change_after(changes, t)
        rows.append(dict({"journey": name, "row": 1, "view": sess.view, "first_ms": cols["per_key_first_ms"][i],
                          "first_any_ms": None if a is None or a > nxt else round((a - t) / MS, 1)}, **(extra or {})))
        if sess.load:
            rows[-1].update(load_evidence(reads[i], ends[i]))
    return rows


def paced_rows(sess, chords, names, pace_ms, extra=None):
    """The rows of paced inputs; an error mid-run keeps the rows of the inputs already sent in sess.taken."""
    times = []
    try:
        paced(sess, chords, pace_ms, times)
    except (RuntimeError, OSError, ValueError, subprocess.CalledProcessError):
        if times:
            sess.taken += per_key_rows(sess, times, names[:len(times)], extra)
        raise
    return per_key_rows(sess, times, names, extra)


def series_pairs(sess, spec, n):
    rows = sess.taken
    for a, b in spec["pairs"]:
        sess.keys.chord("home")
        sess.settle(1.0, 5)
        chords = [a, b] * (n * SERIES_ROWS_PER_SAMPLE["pairs"])
        names = ["nav.%s.%s" % (sess.view, CHORD_NAMES[c]) for c in chords]
        rows += paced_rows(sess, chords, names, PACE_MS)
    return rows


def series_scroll(sess, spec, n):
    sess.keys.chord("home")
    sess.settle(1.0, 5)
    pre = N.preroll_count(sess.region[3])
    paced(sess, [("down",)] * pre, N.SCROLL_PREROLL_PACE_MS)
    sess.settle(2.0, 15)
    count = n * SERIES_ROWS_PER_SAMPLE["scroll"]
    return paced_rows(sess, [("down",)] * count, ["nav.%s.down-scroll" % sess.view] * count, PACE_MS, {"preroll": pre})


def series_wheel(sess, spec, n):
    pointer_to(sess.keys, *scroll_point(sess.view, sess.region, sess.rect("left")))
    sess.settle(1.0, 5)
    half = SERIES_ROWS_PER_SAMPLE["wheel"] // 2
    sends = [lambda: wheel(sess.keys, -1)] * (n * half) + [lambda: wheel(sess.keys, 1)] * (n * half)
    rows = sess.taken
    rows += paced_rows(sess, sends, ["nav.%s.wheel" % sess.view] * (n * SERIES_ROWS_PER_SAMPLE["wheel"]), PACE_MS)
    S.park_cursor()
    return rows


def series_fling(sess, spec, n):
    T, why = touchpad()
    if T is None:
        return [{"journey": "nav.%s.fling" % sess.view, "row": 1, "view": sess.view, "first_ms": "no instrument",
                 "note": why}]
    pointer_to(sess.keys, *scroll_point(sess.view, sess.region, sess.rect("left")))
    fd = T.create()
    rows = sess.taken
    try:
        time.sleep(T.CREATE_SETTLE_S)
        sess.settle(1.0, 5)
        for i in range(n * SERIES_ROWS_PER_SAMPLE["fling"]):
            t = stroke(T, fd, -40 if i % 2 == 0 else 40, 120)
            sess.rec.wait_quiet(N.NAV_QUIET_NS, t, N.NAV_CAP_NS)
            rows.append(sess.columns("nav.%s.fling" % sess.view, t, 1, quiet_s=2, cap_s=15))
    finally:
        drop_touchpad(T, fd)
        S.park_cursor()
    return rows


def series_held(sess, spec, n):
    """Held j sweeps: the repeat slots drawn, the gap between drawn frames, and overshoot against the idle bound."""
    delay, rate = sess.ctx["repeat"]["delay_ms"], sess.ctx["repeat"]["rate"]
    interval = 1000.0 / rate
    hold_ns = int((delay + interval * HOLD_REPEATS + interval / 2) * MS)
    stride = sess.ipc_int("gridColumns") if sess.view == "grid" else 1
    bound = sess.ctx["repeat"].get("max_events")
    rows = sess.taken
    code = N.keycode("j")
    for i in range(n * SERIES_ROWS_PER_SAMPLE["held"]):
        sess.keys.chord("home")
        sess.settle(1.0, 5)
        start = sess.ipc_int("cursor")
        sess.settle(0.5, 3)
        t_down = time.monotonic_ns()
        sess.keys._key(code, 1)
        while time.monotonic_ns() < t_down + hold_ns - 2 * MS:
            time.sleep(0.001)
        while time.monotonic_ns() < t_down + hold_ns:
            pass
        t_up = time.monotonic_ns()
        sess.keys._key(code, 0)
        sess.rec.wait_quiet(N.NAV_QUIET_NS, t_up, N.NAV_CAP_NS)
        final = sess.ipc_int("cursor")
        h = held_analysis(sess.rec.snapshot(), t_down, t_up, delay, interval)
        over, behind = overshoot(start, final, stride, bound)
        h.update({"start_index": start, "final_index": final, "stride": stride, "idle_max_events": bound,
                  "overshoot": over, "steps_behind": behind, "hold_ms": round((t_up - t_down) / MS, 1)})
        rows.append({"journey": "nav.%s.held-j-sweep" % sess.view, "row": 2, "view": sess.view,
                     "first_ms": h["max_gap_ms"], "held": h})
    return rows


def folder_setup(patch, path):
    """The keys that put the cursor on the first folder of path."""
    if patch.get("foldersFirst") is not True:
        raise RuntimeError("the Enter target assumes folders first, and this session does not set foldersFirst to true")
    if not any(e.is_dir() for e in os.scandir(path) if not e.name.startswith(".")):
        raise RuntimeError("no folder in %s for Enter to enter" % path)
    return [k("home")]  # folders first puts the first folder at row 0


def names_off(fixture, want):
    """Names that differ between a fixture directory and its expected names, sorted."""
    return sorted(set(os.listdir(fixture)) ^ set(want))


def put_back(fixture, want, typed):
    """Rename each file whose stem grew by typed back to the expected name it came from."""
    have, want = set(os.listdir(fixture)), set(want)
    for name in sorted(have - want):
        stem, ext = os.path.splitext(name)
        old = name
        if typed and stem.endswith(typed):
            old = stem[:len(stem) - len(typed)] + ext
        elif typed:
            for w in sorted(want):
                wstem, wext = os.path.splitext(w)
                if w not in have and ext == wext and stem.startswith(wstem) and 0 < len(stem) - len(wstem) <= len(typed):
                    old = w
                    break
        if old in want and old not in have:
            os.rename(os.path.join(fixture, name), os.path.join(fixture, old))


def series_cycle(sess, spec, n):
    """Repeated steps, a step with a journey timed; to_folder aims Enter at a folder, first_suffix renames cycle 0's journeys."""
    # restores: the fixture's expected names; typed: what the cycle appends to a stem, to put a failed run back.
    want = spec.get("restores")
    if want is not None and names_off(sess.fixture, want):
        raise RuntimeError("the fixture %s starts off its expected names: %s"
                           % (sess.fixture, ", ".join(names_off(sess.fixture, want)[:4])))
    rows = sess.taken
    try:
        for inp in folder_setup(sess.patch, sess.fixture) if spec.get("to_folder") else spec.get("setup", []):
            sess.send(inp)
            sess.settle(1.0, CAP[3])
        for i in range(n * SERIES_ROWS_PER_SAMPLE["cycle"]):
            for step in spec["steps"]:
                j = step.get("journey")
                if j is None:
                    sess.send(step["in"])
                    sess.settle(step.get("quiet", 1.0), step.get("cap", 3))
                    continue
                if i < FIRST_ROWS and spec.get("first_suffix"):
                    j += spec["first_suffix"]
                ev = "%s-%s-%d" % (sess.ctx["tag"], j, i) if i < FIRST_ROWS and sess.ctx.get("evidence") else None
                rows.append(sess.sample(j, step["in"], row_of(j, spec.get("fixture", "local")) or 4, ev,
                                        {"cycle": i}))
        sess.require_window()
    finally:
        left = names_off(sess.fixture, want) if want is not None else []
        if left:
            put_back(sess.fixture, want, spec["typed"])
    if left:
        raise RuntimeError("the cycle left these names changed in %s: %s" % (sess.fixture, ", ".join(left[:4])))
    return rows


def series_views(sess, spec, n):
    """The view circuit, every ordered pair timed once per lap."""
    rows = sess.taken
    for lap in range(n * SERIES_ROWS_PER_SAMPLE["views"]):
        for a, b in zip(VIEW_CIRCUIT, VIEW_CIRCUIT[1:]):
            inp = ("click", ("chrome", "dual")) if b == "dual" else ("key", VIEW_KEYS[b])
            rows.append(sess.sample("view.switch.%s-%s" % (a, b), inp, 3, None, {"lap": lap}))
            S.park_cursor()
    sess.require_window()
    return rows


def series_land(sess, spec, n):
    """Columns preview: the cursor key that lands on the file, from its neighbour, cycle 0 named .first."""
    names = sorted(x for x in os.listdir(sess.fixture) if not x.startswith("."))
    rows = sess.taken
    for kind in spec["kinds"]:
        idx = names.index(KINDS[kind])
        go, back = (("down",), ("up",)) if idx > 0 else (("up",), ("down",))
        sess.keys.chord("home")
        time.sleep(0.2)
        for _ in range(idx - 1 if idx > 0 else 1):
            sess.keys.chord("down")
            time.sleep(0.15)
        sess.settle(2.0, 10)
        for i in range(n * SERIES_ROWS_PER_SAMPLE["land"] + FIRST_ROWS):
            j = "open.cols-preview.%s%s" % (kind, ".first" if i < FIRST_ROWS else "")
            ev = "%s-%s" % (sess.ctx["tag"], j) if i < FIRST_ROWS and sess.ctx.get("evidence") else None
            rows.append(sess.sample(j, ("key", go), row_of(j, spec.get("fixture", "local")), ev, {"cycle": i}))
            sess.send(("key", back))
            sess.settle(QUIET[5], CAP[5])
    return rows


def series_ql(sess, spec, n):
    """Quick Look of each kind: Space opens, Space closes; cycle 0 is the uncached open, named .first."""
    names = sorted(x for x in os.listdir(sess.fixture) if not x.startswith("."))
    rows = sess.taken
    for kind in spec["kinds"]:
        idx = names.index(KINDS[kind])
        sess.keys.chord("home")
        time.sleep(0.2)
        for _ in range(idx):
            sess.keys.chord("down")
            time.sleep(0.15)
        sess.settle(2.0, 10)
        for i in range(n * SERIES_ROWS_PER_SAMPLE["ql"] + FIRST_ROWS):
            j = "open.ql.%s%s" % (kind, ".first" if i < FIRST_ROWS else "")
            row = row_of(j, spec.get("fixture", "local"))
            ev = "%s-%s" % (sess.ctx["tag"], j) if i < FIRST_ROWS and sess.ctx.get("evidence") else None
            rows.append(sess.sample(j, ("key", ("space",)), row, ev, {"cycle": i}))
            sess.send(("key", ("space",)))
            sess.settle(QUIET[row], CAP[row])
    return rows


def series_search(sess, spec, n):
    """search.type-to-first-result: f opens the search, the query is typed at 150 ms a key, the final key is timed."""
    rows = sess.taken
    for i in range(n * SERIES_ROWS_PER_SAMPLE["search"]):
        sess.send(("key", ("f",)))
        sess.settle(1.0, 3)
        times = S.type_keys(sess.keys, spec["query"], 0.15)
        sess.rec.wait_quiet(int(QUIET[4] * NS), times[-1], int(CAP[4] * NS))
        rows.append(sess.columns("search.type-to-first-result", times[-1], 4, extra={"query": spec["query"]}))
        sess.send(("key", ("esc",)))
        sess.settle(1.0, 3)
    return rows


def menu_to(sess, label):
    """Open the context menu on the cursor row and walk its cursor to label; False when the row is absent."""
    sess.send(("key", ("m",)))
    sess.settle(1.0, 3)
    entries = sess.ipc("contextMenuEntries").split("|")
    if label not in entries:
        sess.send(("key", ("esc",)))
        sess.settle(1.0, 3)
        return False
    want = entries.index(label)
    for _ in range(len(entries) + 2):
        if sess.ipc_int("contextMenuCursor") == want:
            return True
        sess.keys.chord("down")
        time.sleep(0.1)
    return sess.ipc_int("contextMenuCursor") == want


def wait_until(pred, cap_s):
    t0 = time.monotonic()
    while time.monotonic() - t0 < cap_s:
        if pred():
            return round(time.monotonic() - t0, 2)
        time.sleep(0.1)
    return None


# Op isolation: a sample starts on a status bar with no queued error and no activity, or it is an error row.
ISOLATE_PASTE_CAP_S = 20  # s a sample may spend dismissing queued errors before its input
ISOLATE_CANCEL_CAP_S = 60  # s a cancelled transfer may take to leave the activity list
ISOLATE_POLL_S = 0.3  # s between status reads, at most one Esc per read
ISOLATE_CANCEL_GAP_S = 5.0  # s before a cancel that left the transfer running is pressed again


def activity_state(sess):
    """statusActivityState (ui/Ipc.qml) as a dict, or None when the call gave nothing parseable."""
    try:
        got = json.loads(sess.ipc("statusActivityState"))
    except (ValueError, subprocess.TimeoutExpired):
        return None
    return got if isinstance(got, dict) else None


def busy_reason(state, escs):
    """What an unclear status bar still holds, for the error row."""
    if state is None:
        return "the status bar is not idle: statusActivityState gave no reading"
    if state.get("errors"):
        return "the status bar is not idle: %d queued errors after %d Esc" % (state["errors"], escs)
    acts = state.get("activities") or []
    return "the status bar is not idle: %d activities after %d Esc, the first %r" % (
        len(acts), escs, acts[0].get("text") if acts else "")


def isolate(read, esc, cap_s, poll_s=ISOLATE_POLL_S, now=None, sleep=None):
    """Esc until read() shows no queued error and no activity; (Esc count, last reading, why), why None when clear.

    Esc dismisses the oldest queued error before it cancels anything (ui/StatusBar.qml escapePressed), so errors are
    pressed away first, one Esc per read; a running transfer that is not already cancelling gets one cancel Esc, repeated
    after ISOLATE_CANCEL_GAP_S if it is still running. A read of None (the IPC call failed) is waited out like a busy bar."""
    now, sleep = now or time.monotonic, sleep or time.sleep
    t0, escs, cancelled_at, st = now(), 0, None, None
    while True:
        st = read()
        if st is not None:
            acts, errs = st.get("activities") or [], st.get("errors") or 0
            if not acts and not errs:
                return escs, st, None
            if errs:
                esc()
                escs += 1
            elif any(a.get("running") and not a.get("cancelling") for a in acts) and (
                    cancelled_at is None or now() - cancelled_at >= ISOLATE_CANCEL_GAP_S):
                esc()
                escs += 1
                cancelled_at = now()
        if now() - t0 >= cap_s:
            return escs, st, busy_reason(st, escs)
        sleep(poll_s)


def wait_idle(sess, cap_s):
    """(Esc count, idle reading) once the status bar is clear; a RuntimeError naming what stays when it is not."""
    escs, st, why = isolate(lambda: activity_state(sess), lambda: sess.keys.chord("esc"), cap_s)
    if why:
        raise RuntimeError(why)
    return escs, st


def cancel_op(sess):
    """Esc a transfer that outlived its wait and give it ISOLATE_CANCEL_CAP_S to leave the activity list."""
    escs, _, why = isolate(lambda: activity_state(sess), lambda: sess.keys.chord("esc"), ISOLATE_CANCEL_CAP_S)
    return dict({"cancel_escs": escs}, **({"cancel_failed": why} if why else {}))


def op_state_fields(before, after):
    """errors_before, errors_after, activity_after, notice_before, notice_after and refused of an op row, from two
    statusActivityState readings (None: no reading). refused is True when the paste left more queued errors than it found;
    notice_before is what the copy or cut before it said ("Copied 0 items" is a paste of nothing)."""
    eb = None if before is None else before.get("errors")
    ea = None if after is None else after.get("errors")
    acts = None if after is None else [{k: a.get(k) for k in ("text", "running", "cancelling")}
                                       for a in after.get("activities") or []]
    return {"errors_before": eb, "errors_after": ea, "activity_after": acts,
            "notice_before": None if before is None else before.get("notice"),
            "notice_after": None if after is None else after.get("notice"),
            "refused": None if eb is None or ea is None else ea > eb}


# Op navigation: a sample starts in the folder sample 0 started in, read off the pane, never assumed from keys sent blind.
LISTING_CAP_S = 10  # s a listing may take to land before a key is judged lost
NAV_STEPS = 6  # keys a walk back to the source or the destination may spend
TAKE_CAP_S = 5.0  # s the clipboard may take to hold the copied or cut paths
HARNESS_FAULTS_STOP = 3  # consecutive harness faults after which an op series stops
ROW_SCAN_CAP = 64  # rows row_index reads off the pane before it calls a row missing


class HarnessFault(RuntimeError):
    """The tool could not give a sample the state sample 0 started from: a row of its own, never a refusal or a miss."""


def listing_landed(sess, cap_s=None):
    """Wait until the pane has no listing in flight (listInFlight), since a key pressed before is refused (ui/js/Nav.js)."""
    cap_s = LISTING_CAP_S if cap_s is None else cap_s
    t0 = time.monotonic()
    while sess.ipc("listInFlight").strip().lower() != "false":
        if time.monotonic() - t0 >= cap_s:
            raise HarnessFault("the listing did not land in %g s" % cap_s)
        time.sleep(0.1)


def nav_key(sess, *chord):
    """One navigation key, sent only on a landed listing and followed by the listing it starts."""
    listing_landed(sess)
    sess.keys.chord(*chord)
    time.sleep(0.25)
    listing_landed(sess)


def same_dir(a, b):
    return bool(a) and os.path.realpath(a) == os.path.realpath(b)


def row_index(sess, name):
    """The listing index of the row called name, or None: a symlink to a folder is listed after the folders, so find it by name."""
    for i in range(min(sess.ipc_int("total") or 0, ROW_SCAN_CAP)):
        if sess.ipc("visibleRowName", i) == name:
            return i
    return None


def enter_row(sess, name):
    """Put the cursor on the row called name and press Enter, checking the cursor before it."""
    i = row_index(sess, name)
    if i is None:
        raise HarnessFault("no row %r in %s" % (name, sess.ipc("path")))
    nav_key(sess, "home")
    for _ in range(i):
        nav_key(sess, "down")
    if sess.ipc_int("cursor") != i:
        raise HarnessFault("the cursor is on row %s, not on %r (row %d)" % (sess.ipc("cursor"), name, i))
    nav_key(sess, "enter")


def reach(sess, want, base, via, kids):
    """Walk the pane to want: from a folder in kids Backspace climbs to base (Nav.js parent cuts the pane path), Enter on via goes down."""
    base_real = os.path.realpath(base)
    for _ in range(NAV_STEPS):
        here = sess.ipc("path")
        if same_dir(here, want):
            return
        if same_dir(here, base):
            enter_row(sess, via)
        elif os.path.realpath(here) in kids or os.path.dirname(os.path.realpath(here)) == base_real:
            nav_key(sess, "backspace")
        else:
            raise HarnessFault("the pane is at %r, outside %s" % (here, base))
    raise HarnessFault("the pane did not reach %s in %d keys, it is at %r" % (want, NAV_STEPS, sess.ipc("path")))


def listed_total(sess, count, cap_s=None):
    """Wait until the pane lists exactly count rows: Flea has seen the folder the tool just filled or emptied."""
    cap_s = LISTING_CAP_S if cap_s is None else cap_s
    t0 = time.monotonic()
    while sess.ipc_int("total") != count:
        if time.monotonic() - t0 >= cap_s:
            raise HarnessFault("%s lists %s rows, not %d" % (sess.ipc("path"), sess.ipc("total"), count))
        time.sleep(0.1)


def clipboard_holds(sess, paths, moving, cap_s=None):
    """Wait until the pane clipboard (fileClipboard) holds exactly these paths, real, with the right verb."""
    cap_s = TAKE_CAP_S if cap_s is None else cap_s
    t0, got = time.monotonic(), None
    while True:
        try:
            got = json.loads(sess.ipc("fileClipboard"))
        except ValueError:
            got = None
        held = sorted(os.path.realpath(p) for p in got.get("paths") or []) if isinstance(got, dict) else None
        if held == sorted(os.path.realpath(p) for p in paths) and bool(got.get("moving")) == moving:
            return
        if time.monotonic() - t0 >= cap_s:
            raise HarnessFault("the clipboard holds %s paths (moving %s), not the %d of this sample (moving %s)" % (
                len(held) if held is not None else "no reading", got.get("moving") if held is not None else "-",
                len(paths), moving))
        time.sleep(0.1)


# Each op: its source under ROOT/ops, the chord that starts it, and whether its destination is on the USB stick.
OPS = {
    "op.copy.many": {"src": "src-many", "take": ("ctrl", "c"), "usb": False, "count": COPY_MANY},
    "op.copy.large": {"src": None, "file": "large.bin", "take": ("ctrl", "c"), "usb": False, "count": 1},
    "op.copy.long": {"src": None, "file": "long.bin", "take": ("ctrl", "c"), "usb": True, "count": 1, "long": True},
    "op.move.cross-fs": {"src": "src-move", "take": ("ctrl", "x"), "usb": True, "count": MOVE_FILES},
}


def series_op(sess, spec, n):
    """One op journey, n samples, each from a reset source and an idle status bar: the key that starts it to the first changed pixel."""
    name, roots, ctx = spec["op"], sess.ctx["roots"], sess.ctx
    base = ctx["op_base"]  # ROOT/ops/run/<op>, holding a-dst and b-src; Flea opens in b-src
    dst, src = os.path.join(base, "a-dst"), os.path.join(base, "b-src")
    rows = sess.taken
    for i in range(n * SERIES_ROWS_PER_SAMPLE["op"]):
        ev = "%s-%s-%d" % (ctx["tag"], name, i) if i < FIRST_ROWS and ctx.get("evidence") else None
        try:
            escs, before = wait_idle(sess, ISOLATE_PASTE_CAP_S)  # before any folder is emptied under a job that may still run
            if name in OPS:
                o = OPS[name]
                real_dst, real_src = os.path.realpath(dst), os.path.realpath(src)
                kids = (real_src, real_dst)
                reach(sess, real_src, base, "b-src", kids)  # sample 0 starts here; later samples are walked back
                empty_dir(real_dst, roots)
                empty_dir(src, roots)
                if o["src"]:
                    reflink(os.path.join(ctx["root"], "ops", o["src"], "."), src)
                else:
                    reflink(os.path.join(ctx["root"], "ops", o["file"]), src)
                sess.settle(2.0, 10)
                listed_total(sess, o["count"])  # the pane lists what the tool just put there
                for c in (("ctrl", "a"), o["take"]):
                    nav_key(sess, *c)
                clipboard_holds(sess, [os.path.join(real_src, n) for n in os.listdir(real_src)], o["take"] == ("ctrl", "x"))
                reach(sess, real_dst, base, "a-dst", kids)
                listed_total(sess, 0)  # the destination is clear in the pane and on the disk
                if os.listdir(real_dst):
                    raise HarnessFault("the destination %s is not clear" % real_dst)
                sess.settle(1.0, 5)
                more, before = wait_idle(sess, ISOLATE_PASTE_CAP_S)  # the keys above may have queued an error
                escs += more
                t = sess.keys.chord("ctrl", "v")
                if ev:
                    sess.rec.armed_ns = t
                extra = {"sample": i}
                if o.get("long"):
                    while time.monotonic_ns() < t + 11 * NS:
                        time.sleep(0.05)
                    extra["evidence_11s"] = (sess.keep_dump_now("%s-%s-11s" % (ctx["tag"], name))
                                             if sess.rec.proc.poll() is None else None)
                    extra["activity_11s"] = sess.ipc("statusActivityState")
                want = os.path.getsize(os.path.join(ctx["root"], "ops", o["file"])) if o.get("file") else None

                def done():
                    names = os.listdir(real_dst)
                    if len(names) < o["count"]:
                        return False
                    if want is not None:
                        return os.path.getsize(os.path.join(real_dst, names[0])) >= want
                    return name != "op.move.cross-fs" or not os.listdir(src)
                waited = wait_until(done, 900 if o.get("long") else 120)
                extra["op_s"] = None if waited is None else round((time.monotonic_ns() - t) / NS, 2)
                sess.rec.wait_quiet(int(QUIET[6] * NS), t, int(CAP[6] * NS))
                extra.update(op_state_fields(before, activity_state(sess)), esc_before=escs)
                row = sess.columns(name, t, 6, evidence=ev, extra=extra)
                if extra["op_s"] is None:
                    row.update(cancel_op(sess))  # the next sample must not start under this job
                rows.append(row)
            elif name == "op.trash":
                empty_dir(src, roots)
                reflink(os.path.join(ctx["root"], "ops", "src-trash", "t_00.txt"), src)
                sess.settle(2.0, 10)
                sess.keys.chord("home")
                sess.settle(1.0, 3)
                row = sess.sample(name, ("key", ("delete",)), 6, ev, {"sample": i})
                row["op_s"] = wait_until(lambda: not os.path.exists(os.path.join(src, "t_00.txt")), 30)
                row.update(op_state_fields(before, activity_state(sess)), esc_before=escs)
                rows.append(row)
            elif name in ("op.size.recursive", "op.archive"):
                label = "Properties" if name == "op.size.recursive" else "Compress"
                present = set(os.listdir(src))
                sess.keys.chord("home")
                sess.settle(1.0, 3)
                if not menu_to(sess, label):
                    rows.append({"journey": name, "row": 6, "view": sess.view, "first_ms": "no instrument",
                                 "note": "the context menu holds no %s row here" % label})
                    break
                if name == "op.archive":
                    sess.send(("key", ("right",)))
                    sess.settle(1.0, 3)
                row = sess.sample(name, ("key", ("enter",)), 6, ev, {"sample": i})
                if name == "op.archive":
                    row["op_s"] = wait_until(lambda: set(os.listdir(src)) - present, 60)
                    for extra_name in set(os.listdir(src)) - present:
                        sess.settle(2.0, 60)
                        sandbox_rm(os.path.join(src, extra_name), roots)
                row.update(op_state_fields(before, activity_state(sess)), esc_before=escs)
                rows.append(row)
                sess.send(("key", ("esc",)))
                sess.settle(1.0, 5)
        except HarnessFault as e:
            rows.append({"journey": name, "row": 6, "view": sess.view, "sample": i, "harness_fault": True,
                         "error": "%s: %s" % (type(e).__name__, e)})
            if all(r.get("harness_fault") for r in rows[-HARNESS_FAULTS_STOP:]) and len(rows) >= HARNESS_FAULTS_STOP:
                break  # the tool cannot reach the state sample 0 had: more samples would only repeat it
        except (RuntimeError, OSError, subprocess.SubprocessError) as e:
            rows.append({"journey": name, "row": 6, "view": sess.view, "error": "%s: %s" % (type(e).__name__, e)})
    return rows


def keep_dump_now(self, stem):
    """Dump the newest frame now, as a PNG."""
    got = self.rec.dump()
    if not got:
        return None
    self.rec.dumps.remove(got)
    png = os.path.join(self.ctx["out_dir"], stem + ".png")
    S.ppm_to_png(got, png, self.rec.boxes(self.region, ()))
    return png


Session.keep_dump_now = keep_dump_now

SERIES = {"pairs": series_pairs, "scroll": series_scroll, "wheel": series_wheel, "fling": series_fling,
          "held": series_held, "cycle": series_cycle, "views": series_views, "land": series_land, "ql": series_ql,
          "search": series_search, "op": series_op}
# Rows per sample for each series type, read by the series functions and planned_counts.
SERIES_ROWS_PER_SAMPLE = {"pairs": 1, "scroll": 1, "wheel": 2, "fling": 1, "held": 1, "cycle": 1,
                          "views": 1, "land": 1, "ql": 1, "search": 1, "op": 1}
# Rows for a one-shot .first journey, read by the cycle, land and ql series and planned_counts.
FIRST_ROWS = 1


# Parts: sessions of fixture, view, region and series from fixtures.json; --dry checks one with no display.

def k(*chord):
    return ("key", chord)


def cyc(steps, **kw):
    return dict({"type": "cycle", "steps": steps}, **kw)


def open_close(journey, key_in, close=("esc",)):
    return cyc([{"journey": journey, "in": key_in}, {"in": k(*close), "quiet": 1.0, "cap": 3}])


RENAME_PAIRS = 3  # j and k pairs timed in a row on each side of the rename
RENAME_LETTERS = ("a", "b")  # typed at the end of the stem, so the new name sorts into the old name's slot
NAV500_NAME = "f_%04d.txt"  # tools/flea-nav-fixture's name for file i of nav500


def rename_cycle(view, names):
    """p8-rename for one view: j and k pairs, the cursor row renamed, the same pairs again, the rename undone."""
    def pairs(side):
        return [{"journey": "nav.%s.%s-rename-%s" % (view, side, key), "in": k(key)}
                for _ in range(RENAME_PAIRS) for key in ("j", "k")]

    def edit(*keys):
        typed = [{"in": k(c), "quiet": 0.3, "cap": 2} for c in keys]
        return [{"in": k("r")}, {"in": k("right")}] + typed + [{"in": k("enter")}]
    undo = ("backspace",) * len(RENAME_LETTERS)
    return cyc(pairs("pre") + edit(*RENAME_LETTERS) + pairs("after") + edit(*undo),
               setup=[k("ctrl", "d")] * 3, restores=names, typed="".join(RENAME_LETTERS))


def enter_back(view, fixture="local"):
    return cyc([{"journey": "nav.%s.enter" % view, "in": k("enter")},
                {"journey": "nav.%s.back" % view, "in": k("backspace")}], to_folder=True, fixture=fixture)


def parts(fx, samples):
    """Every part's sessions from the fixtures facts; samples is n per series (cursor pairs: per chord)."""
    nav = os.path.join(fx["root"], "nav")
    n500, n10k, enter, prev = (os.path.join(nav, d) for d in ("nav500", "nav10k", "enter", "preview"))
    ops = os.path.join(fx["root"], "ops")
    L = os.path.join(fx["root"], "load", "L")
    views = ["list", "columns", "grid", "dual"]
    out = {}
    out["p1-cursor"] = [{"fixture": n10k, "view": v, "region": "content", "series": [
        {"type": "pairs", "pairs": PAIRS + (GRID_PAIRS if v == "grid" else [])},
        {"type": "scroll"}, {"type": "wheel"}, {"type": "fling"}]} for v in views]
    out["p2-held"] = [{"fixture": n10k, "view": v, "region": "content", "series": [{"type": "held"}]} for v in views]
    out["p3-navigate"] = [{"fixture": enter, "view": v, "region": "content", "series": [enter_back(v)]} for v in views]
    out["p3-navigate"] += [
        {"fixture": enter, "view": "columns", "region": "content", "series": [cyc(
            [{"journey": "nav.columns.shift-right", "in": k("right")},
             {"journey": "nav.columns.shift-left", "in": k("left")}], setup=[k("down")])]},
        {"fixture": n500, "view": "list", "region": "window", "series": [{"type": "views"}]},
        {"fixture": n500, "view": "list", "region": "window", "series": [
            cyc([{"journey": "tab.new", "in": k("t")}, {"in": k("w"), "quiet": 1.0, "cap": 3}]),
            cyc([{"journey": "tab.switch", "in": k("]")}, {"journey": "tab.switch", "in": k("[")}],
                setup=[k("t")])]},
        {"fixture": n500, "view": "dual", "region": "content", "series": [
            cyc([{"journey": "pane.switch.dual", "in": k("tab")}])]}]
    out["p4-open"] = [
        {"fixture": prev, "view": "list", "region": "window", "series": [
            open_close("open.menu.key", k("m")),
            open_close("open.menu.click", ("rclick", ("row", 0))),
            open_close("open.rename", k("r")),
            open_close("open.settings", k(",")),
            open_close("open.palette", k("shift", "/")),
            open_close("open.pathjump", k("shift", ";")),
            {"type": "ql", "kinds": ["txt", "jpg", "md", "code"]}]},
        {"fixture": prev, "view": "columns", "region": "pane", "series": [
            {"type": "land", "kinds": ["txt", "jpg", "md", "code"]}]},
        {"fixture": n10k, "view": "list", "region": "content", "series": [{"type": "search", "query": "h_0123"}]}]
    out["p5-heavy"] = [
        {"fixture": prev, "view": "list", "region": "window", "series": [
            {"type": "ql", "kinds": ["video", "pdf", "large-image"]}]},
        {"fixture": prev, "view": "columns", "region": "pane", "series": [
            {"type": "land", "kinds": ["video", "pdf", "large-image"]}]}]
    for where in ("usb", "nas"):
        r = fx.get(where)
        if not r:
            out["p5-heavy"].append({"missing": where})
            continue
        out["p5-heavy"] += [
            {"fixture": os.path.join(r, "enter"), "view": "list", "region": "content", "where": where,
             "series": [enter_back("list", where)]},
            {"fixture": os.path.join(r, "preview"), "view": "list", "region": "window", "where": where,
             "series": [{"type": "ql", "kinds": list(REMOTE_KINDS), "fixture": where}]}]
    out["p6-ops"] = [{"fixture": os.path.join(ops, "run", name.split(".", 1)[1].replace(".", "-")), "view": "list",
                      "region": "window", "op": name, "series": [{"type": "op", "op": name}],
                      **({"n": 3} if name == "op.copy.long" else {}),
                      **({"ui": SIZE_UI} if name == "op.size.recursive" else {})}
                     for name in ("op.copy.many", "op.copy.large", "op.move.cross-fs", "op.trash",
                                  "op.size.recursive", "op.archive", "op.copy.long")]
    load_series = [
        {"type": "pairs", "pairs": [(("down",), ("up",))]},
        enter_back("dual"),
        open_close("open.menu.key", k("m")),
        cyc([{"journey": "open.ql.txt", "in": k("space")}, {"in": k("space"), "quiet": 2.0, "cap": 8}],
            setup=[k("home"), k("down")], first_suffix=".first")]
    out["p7-load"] = []
    for load in ("a", "b", "d", "c"):
        # Load d is raw dd with no Flea copy, so it takes the nav pair only, to tell a slow frame in the stick's own writes from one in Flea's copy.
        for s in (load_series[:1] if load == "d" else load_series):
            out["p7-load"].append({"fixture": L, "view": "dual", "region": "left", "load": load, "series": [s]})
    names = sorted(NAV500_NAME % i for i in range(fx["nav"]["nav500"]))
    out["p8-rename"] = [{"fixture": n500, "view": v, "region": "content", "series": [rename_cycle(v, names)]}
                        for v in views]
    out["smoke"] = [
        {"fixture": n10k, "view": "list", "region": "content", "series": [
            {"type": "pairs", "pairs": [(("down",), ("up",))]}, {"type": "wheel"}, {"type": "fling"},
            {"type": "held"}], "n": 3},
        {"fixture": enter, "view": "list", "region": "content", "series": [enter_back("list")], "n": 3},
        {"fixture": n500, "view": "list", "region": "window", "series": [
            {"type": "views"}, cyc([{"journey": "tab.new", "in": k("t")}, {"in": k("w")}])], "n": 1},
        {"fixture": prev, "view": "list", "region": "window", "series": [
            open_close("open.menu.key", k("m")), open_close("open.menu.click", ("rclick", ("row", 0))),
            {"type": "ql", "kinds": ["txt", "video"]}], "n": 2},
        {"fixture": prev, "view": "columns", "region": "pane", "series": [{"type": "land", "kinds": ["txt"]}],
         "n": 2},
        {"fixture": os.path.join(ops, "run", "copy-many"), "view": "list", "region": "window", "op": "op.copy.many",
         "series": [{"type": "op", "op": "op.copy.many"}], "n": 2},
        {"fixture": os.path.join(ops, "run", "trash"), "view": "list", "region": "window", "op": "op.trash",
         "series": [{"type": "op", "op": "op.trash"}], "n": 2}]
    if fx.get("usb"):
        out["smoke"].append({"fixture": os.path.join(fx["usb"], "enter"), "view": "list", "region": "content",
                             "where": "usb", "series": [enter_back("list", "usb")], "n": 2})
    for sessions in out.values():
        for s in sessions:
            s.setdefault("n", samples)
    return out


PARTS = ["smoke", "p1-cursor", "p2-held", "p3-navigate", "p4-open", "p5-heavy", "p6-ops", "p7-load", "p8-rename"]
# Minutes of minipc per part at 20 samples, from the series' own pacing (keys at 400 ms, quiet gaps, launches).
ESTIMATE_MIN = {"smoke": 5, "p1-cursor": 16, "p2-held": 8, "p3-navigate": 14, "p4-open": 14, "p5-heavy": 13,
                "p6-ops": 13, "p7-load": 17, "p8-rename": 34}


def journeys_of(sessions):
    """Every journey id a part's sessions time, for the dry check and the selftest."""
    out = set()
    for s in sessions:
        v = s.get("view")
        for sp in s.get("series", []):
            t = sp["type"]
            if t == "pairs":
                for a, b in sp["pairs"]:
                    out |= {"nav.%s.%s" % (v, CHORD_NAMES[a]), "nav.%s.%s" % (v, CHORD_NAMES[b])}
            elif t in ("scroll", "wheel", "fling"):
                out.add("nav.%s.%s" % (v, {"scroll": "down-scroll"}.get(t, t)))
            elif t == "held":
                out.add("nav.%s.held-j-sweep" % v)
            elif t == "cycle":
                for st in sp["steps"]:
                    if st.get("journey"):
                        out.add(st["journey"])
                        if sp.get("first_suffix"):
                            out.add(st["journey"] + sp["first_suffix"])
            elif t == "views":
                out |= {"view.switch.%s-%s" % p for p in zip(VIEW_CIRCUIT, VIEW_CIRCUIT[1:])}
            elif t == "land":
                out |= {"open.cols-preview.%s%s" % (kk, x) for kk in sp["kinds"] for x in ("", ".first")}
            elif t == "ql":
                out |= {"open.ql.%s%s" % (kk, x) for kk in sp["kinds"] for x in ("", ".first")}
            elif t == "search":
                out.add("search.type-to-first-result")
            elif t == "op":
                out.add(sp["op"])
    return out


def journey_filter(pattern):
    """A predicate over journey ids for --only REGEX (re.search); None keeps every journey. A bad regex is a usage error."""
    if pattern is None:
        return lambda journey: True
    try:
        rx = re.compile(pattern)
    except re.error as e:
        print("--only %r is not a regular expression: %s" % (pattern, e), file=sys.stderr)
        raise SystemExit(2)
    return lambda journey: rx.search(journey) is not None


def only_series(sessions, pattern):
    """The sessions cut to the series with a journey that matches pattern; a session left with none is dropped."""
    keep = journey_filter(pattern)
    out = []
    for s in sessions:
        series = [sp for sp in s.get("series", []) if any(keep(j) for j in journeys_of([dict(s, series=[sp])]))]
        if series:
            out.append(dict(s, series=series))
    return out


# Synthetic fixtures facts for the plan: only usb/nas presence shapes the journeys gate expects.
PLAN_FX = {"root": "/home/flea-sandbox/x", "nav": {"nav500": 500}, "usb": "/run/media/u/s/flea-instant-1",
           "nas": "/run/user/1/gvfs/n/f-2"}


def planned_counts(part, samples=20, fx=None, only=None):
    """Rows per (journey, view, fixture) one part writes at samples; the minima gate checks. only: a --only REGEX."""
    out = {}

    def add(journey, view, fixture, n):
        out[(journey, view, fixture)] = out.get((journey, view, fixture), 0) + n

    for s in parts(fx or PLAN_FX, samples).get(part, []):
        if s.get("missing") or not s.get("series"):
            continue
        view, n, where = s.get("view"), s.get("n", samples), s.get("where", "local")
        fixture = where + ("+load-" + s["load"] if s.get("load") else "")
        for sp in s["series"]:
            t = sp["type"]
            r = SERIES_ROWS_PER_SAMPLE[t]
            if t == "pairs":
                for a, b in sp["pairs"]:
                    add("nav.%s.%s" % (view, CHORD_NAMES[a]), view, fixture, n * r)
                    add("nav.%s.%s" % (view, CHORD_NAMES[b]), view, fixture, n * r)
            elif t == "scroll":
                add("nav.%s.down-scroll" % view, view, fixture, n * r)
            elif t == "wheel":
                add("nav.%s.wheel" % view, view, fixture, n * r)
            elif t == "fling":
                add("nav.%s.fling" % view, view, fixture, n * r)
            elif t == "held":
                add("nav.%s.held-j-sweep" % view, view, fixture, n * r)
            elif t == "search":
                add("search.type-to-first-result", view, fixture, n * r)
            elif t == "views":
                for a, b in zip(VIEW_CIRCUIT, VIEW_CIRCUIT[1:]):
                    add("view.switch.%s-%s" % (a, b), view, fixture, n * r)
            elif t == "cycle":
                for st in sp["steps"]:
                    if not st.get("journey"):
                        continue
                    if sp.get("first_suffix"):
                        add(st["journey"], view, fixture, n * r - FIRST_ROWS)
                        add(st["journey"] + sp["first_suffix"], view, fixture, FIRST_ROWS)
                    else:
                        add(st["journey"], view, fixture, n * r)
            elif t == "land":
                for kk in sp["kinds"]:
                    add("open.cols-preview.%s" % kk, view, fixture, n * r)
                    add("open.cols-preview.%s.first" % kk, view, fixture, FIRST_ROWS)
            elif t == "ql":
                for kk in sp["kinds"]:
                    add("open.ql.%s" % kk, view, fixture, n * r)
                    add("open.ql.%s.first" % kk, view, fixture, FIRST_ROWS)
            elif t == "op":
                add(sp["op"], view, fixture, n * r)
    keep = journey_filter(only)
    return {key: n for key, n in out.items() if keep(key[0])}


# Running a part.


def repeat_settings(plan):
    """Hyprland's repeat delay and rate, read from hyprctl; the plan's values (minipc: 250 ms, 40 a second) else."""
    got = {"delay_ms": plan.get("repeat_delay_ms", 250), "rate": plan.get("repeat_rate", 40), "source": "plan"}
    try:
        d = json.loads(S.hypr("-j", "getoption", "input:repeat_delay"))
        r = json.loads(S.hypr("-j", "getoption", "input:repeat_rate"))
        got.update(delay_ms=int(d["int"]), rate=int(r["int"]), source="hyprctl getoption")
    except (ValueError, KeyError, TypeError):
        pass
    return got


# Height of the calibration client's header, kept out of the results region.
CALIB_HEADER_PX = 48


def calib_window(work, delay):
    """tests/search-calib.qml under qs: (client, results region, query mask, log path)."""
    cfg = os.path.join(work, "cfg")
    os.makedirs(cfg, exist_ok=True)
    shutil.copyfile(os.path.join(REPO, "tests", "search-calib.qml"), os.path.join(cfg, "shell.qml"))
    logp = os.path.join(work, "client-%d-%d.log" % (delay, time.monotonic_ns()))
    logf = open(logp, "w")
    client = subprocess.Popen(["qs", "-p", cfg, "-n"], env=dict(S.ENV, CALIB_DELAY_MS=str(delay),
                              QT_FORCE_STDERR_LOGGING="1"), stdout=logf, stderr=subprocess.STDOUT)
    logf.close()
    win = S.window_by_pid({client.pid} | S.descendants(client.pid), 20)
    if win is None:
        client.terminate()
        raise RuntimeError("the calibration window never mapped")
    S.focus_window(win["address"])
    S.park_cursor()
    time.sleep(1.0)
    win = window_at(win["address"])
    (x, y), (w, h) = win["at"], win["size"]
    return client, (x, y + CALIB_HEADER_PX, w, h - CALIB_HEADER_PX), [(x, y, w, CALIB_HEADER_PX)], logp


def end_client(client):
    client.terminate()
    try:
        client.wait(5)
    except subprocess.TimeoutExpired:
        client.kill()
    time.sleep(0.5)


def repeat_probe(keys, rep, runs=5):
    """The key events an idle Qt client receives for the held sweep's exact hold: the overshoot bound."""
    interval = 1000.0 / rep["rate"]
    hold_ns = int((rep["delay_ms"] + interval * HOLD_REPEATS + interval / 2) * MS)
    counts = []
    with tempfile.TemporaryDirectory(prefix="flea-instant-repeat-") as work:
        for _ in range(runs):
            client, region, masks, logp = calib_window(work, 0)
            try:
                code = N.keycode("a")
                t = time.monotonic_ns()
                keys._key(code, 1)
                while time.monotonic_ns() < t + hold_ns:
                    pass
                keys._key(code, 0)
                time.sleep(1.0)
            finally:
                end_client(client)
            with open(logp) as f:
                counts.append(sum(1 for line in f if "CALIB key" in line))
    return {"counts": counts, "max_events": max(counts), "hold_ms": round(hold_ns / MS, 1),
            "expected": 1 + repeats_due(0, hold_ns, rep["delay_ms"], interval)}


def negative_control(runs, cap_s=3):
    """No key sent: the recorder over the calibration client's results region must see no change within cap_s."""
    out = []
    with tempfile.TemporaryDirectory(prefix="flea-instant-neg-") as work:
        rec_bin = S.build_recorder(os.path.join(work, "rec"))
        for _ in range(runs):
            client, region, masks, _ = calib_window(work, 0)
            try:
                rec = S.Recorder(rec_bin, region, masks)
                time.sleep(0.6)
                t0 = time.monotonic_ns()
                time.sleep(cap_s)
                changed = [f for f in rec.snapshot() if f[0] > t0 and f[1] > 0]
                rec.stop()
            finally:
                end_client(client)
            out.append(len(changed))
    return {"runs": runs, "cap_s": cap_s, "changed_frames": out, "clean": all(c == 0 for c in out)}


FLING_MM, FLING_MS = -40, 120  # the stroke series_fling plays first
FLING_AIM_QUIET_CAP_NS = 5 * NS  # longest wait for the recorder to go quiet after the pointer is aimed
FLING_QUIET_CAP_NS = 3 * NS  # longest wait for the recorder to go quiet after the stroke
CLOCK_OFFSET_READS = 20  # back to back clock reads clock_offset_ns takes the tightest of
WHEEL_LINE = re.compile(r"CALIB wheel at=(\d+) px=(-?\d+)")


def clock_offset_ns():
    """CLOCK_REALTIME minus CLOCK_MONOTONIC in ns, from the tightest of CLOCK_OFFSET_READS back to back reads."""
    best = None
    for _ in range(CLOCK_OFFSET_READS):
        m0 = time.monotonic_ns()
        w = time.time_ns()
        m1 = time.monotonic_ns()
        if best is None or m1 - m0 < best[0]:
            best = (m1 - m0, w - (m0 + m1) // 2)
    return best[1]


def wheel_ms_of(lines, t_first_ns, offset_ns):
    """ms from the stroke's first motion report (CLOCK_MONOTONIC ns) to the client's first pixel-carrying wheel event, read
    from its log lines on the wall clock; None when it logged none. The client's clock floors to a ms, so this reads up to 1 ms early."""
    for line in lines:
        m = WHEEL_LINE.search(line)
        if m:
            return round(int(m.group(1)) - (t_first_ns + offset_ns) / MS, 1)
    return None


def spread(vals):
    """Count, median and range of the numbers in vals; None entries are reads that failed."""
    nums = [v for v in vals if v is not None]
    if not nums:
        return {"n": 0, "median": None, "min": None, "max": None}
    return {"n": len(nums), "median": round(statistics.median(nums), 1), "min": round(min(nums), 1),
            "max": round(max(nums), 1)}


def fling_floor(wheel_ms, frame_ms):
    """The fling floor of a calibrate run: the medians and ranges of the stroke to the idle client's first wheel event and first frame."""
    return {"stroke": {"dy_mm": FLING_MM, "ms": FLING_MS}, "runs": len(frame_ms), "first_wheel_ms": spread(wheel_ms),
            "first_frame_ms": spread(frame_ms), "wheel_ms": wheel_ms, "frame_ms": frame_ms}


def fling_probe(runs, keys=None):
    """What a fling costs before Flea: the census stroke over tests/search-calib.qml, an idle Qt client, read to the client's
    first pixel-carrying wheel event (its log) and to its first changed frame (the recorder, as the census reads it)."""
    T, why = touchpad()
    if T is None:
        return {"skipped": why}
    keys = keys or N.NavKeys()
    work = tempfile.mkdtemp(prefix="flea-instant-fling-")
    rec_bin = S.build_recorder(os.path.join(work, "rec"))
    wheel_ms, frame_ms = [], []
    fd = T.create()
    try:
        time.sleep(T.CREATE_SETTLE_S)
        for _ in range(runs):
            client, region, masks, logp = calib_window(work, 0)
            rec = None
            try:
                rec = S.Recorder(rec_bin, region, masks)
                pointer_to(keys, region[0] + region[2] // 2, region[1] + region[3] // 2)
                rec.wait_quiet(NS, time.monotonic_ns(), FLING_AIM_QUIET_CAP_NS)
                offset = clock_offset_ns()
                t = stroke(T, fd, FLING_MM, FLING_MS)
                rec.wait_quiet(NS, t, FLING_QUIET_CAP_NS)
                frame_ms.append(S.search_columns(rec.snapshot(), [t], time.monotonic_ns(), NS, FLING_QUIET_CAP_NS)["first_ms"])
                with open(logp) as f:
                    wheel_ms.append(wheel_ms_of(f.read().splitlines(), t, offset))
            finally:
                if rec is not None:
                    rec.stop()
                end_client(client)
    finally:
        drop_touchpad(T, fd)
        S.park_cursor()
    return dict(fling_floor(wheel_ms, frame_ms), workdir=work)


# Loads: the probe of each is read per sample, see the docstring.

LOAD_START_S = 30  # how long a renewed load may take to show its first progress
LOAD_POLL_S = 0.02  # between probe reads while a renewed load is awaited
COLD_BATCH = 48  # pictures linked per renewal of load a, a little over the rows the right pane shows at 2560x1440
COLD_SEQ = 1000000  # the numbers in load a's link names count down from here, so no name is used twice
LOAD_GAP_S = {"a": 0.3, "c": 1.0}  # least time between two renewals, a forced one waits it out; c's folder mtime has second resolution


def load_progress(p):
    """The count of a probe value: a bare number, or the first of a (count, busy) pair."""
    return p[0] if isinstance(p, (tuple, list)) else p


def load_busy(p):
    """True when a probe value says work was in flight at that read; only a (count, busy) pair can."""
    return bool(p[1]) if isinstance(p, (tuple, list)) else False


def load_moved(p0, p1):
    """True when the load progressed since the read p0: its count changed, or it is busy at p1; an unread probe moved nothing."""
    a, b = load_progress(p0), load_progress(p1)
    return a is not None and b is not None and (a != b or load_busy(p1))


def load_active(p_in, p_out):
    """True when the load ran during a sample: it progressed between the read before the input and the read after its quiet, or was busy at the first."""
    return load_moved(p_in, p_out) or (load_progress(p_in) is not None and load_busy(p_in))


def load_evidence(p_in, p_out):
    """The two columns a row carries for its load."""
    return {"load_probe": [p_in, p_out], "load_active": load_active(p_in, p_out)}


class Load:
    """A load beside a series: probe() reads its progress, renew() starts it again (renew_fn None: it runs on its own)."""

    def __init__(self, name, note, probe, renew_fn=None, gap_s=0.0):
        self.name, self.note, self.probe, self.renew_fn, self.gap_s, self.at = name, note, probe, renew_fn, gap_s, None

    def renew(self, force=False):
        """Start the load again; False when gap_s has not passed since the last time, unless force: that waits out the rest of the gap."""
        if self.renew_fn is None:
            return False
        wait = 0.0 if self.at is None else self.gap_s - (time.monotonic() - self.at)
        if wait > 0:
            if not force:
                return False
            time.sleep(wait)
        self.at = time.monotonic()
        self.renew_fn()
        return True


def link_cold(src, names, dst, seq, n, linked=None):
    """Link n cold pictures into dst under names that sort just after the first name there; seq[0] counts down across calls so no name repeats. Returns the new paths; linked, when given, is that list and names only the links that exist when a link fails."""
    stem = os.path.splitext(min(os.listdir(dst)))[0]  # the right pane's cursor stays on its first row and a re-read keeps that row's screen y, so the new rows must fall below it
    linked = [] if linked is None else linked
    for _ in range(n):
        seq[0] -= 1
        name = names[seq[0] % len(names)]
        path = os.path.join(dst, "%s.k%06d%s" % (stem, seq[0], os.path.splitext(name)[1]))
        os.link(os.path.join(src, name), path)
        linked.append(path)
    return linked


def load_renew(sess, load, ctx):
    """The function that starts load a or c again, or None: a swaps the batch of pictures linked into the right folder for a fresh one, c changes the NAS folder's mtime."""
    if load == "a":
        cold = os.path.join(ctx["root"], "load", "cold-src")
        names, seq, batch = sorted(os.listdir(cold)), [COLD_SEQ], []

        def drop():
            while batch:  # the right folder keeps its size, so every sample re-reads the same load; batch names only what is left
                sandbox_rm(batch[-1], ctx["roots"])
                batch.pop()

        def renew():
            if not names:
                raise RuntimeError("no cold images in %s" % cold)
            drop()
            try:
                link_cold(cold, names, sess.right, seq, COLD_BATCH, batch)
            except OSError:  # a link that failed midway leaves no partial batch in the folder
                drop()
                raise
        return renew
    if load == "c":
        mark = os.path.join(sess.right, ".restart")

        def renew():
            open(mark, "w").close()
            sandbox_rm(mark, ctx["roots"])
        return renew
    return None


LOAD_D_NAME = "load-d.bin"
LOAD_D_MIB = 4000  # bs=4M count=1000, the size of load b's file


def load_d_command(sink):
    """Load d's writer: direct 4 MiB writes, so the stick's own rate is the load and the page cache holds none of it."""
    return ["dd", "if=/dev/zero", "of=" + sink, "bs=4M", "count=1000", "oflag=direct"]


def load_stop(sess):
    """End a load-d writer still running, so no write reaches the stick once the session's series is over."""
    proc = getattr(sess, "load_proc", None)
    if proc is None or proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(10)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait()


def load_start(sess, load, ctx):
    """Start a load in the right pane; (probe, note). probe() changes while the load runs."""
    roots = ctx["roots"]
    if load == "a":
        thumbs = os.path.join(ctx["sbx"], "cache", "thumbnails")

        def probe():
            return sum(len(fs) for _, _, fs in os.walk(thumbs)) if os.path.isdir(thumbs) else 0
        return probe, "thumbnails of a cold image folder in the right pane"
    if load == "b":
        dst = os.path.realpath(sess.right)
        empty_dir(dst, roots)
        for c in (("end",), ("ctrl", "c"), ("tab",), ("ctrl", "v"), ("tab",), ("home",)):
            sess.keys.chord(*c)
            time.sleep(0.3)

        def probe():
            return sum(os.path.getsize(os.path.join(dst, f)) for f in os.listdir(dst))
        return probe, "copy of a %d MiB file to the USB stick from the right pane" % LONG_MIB
    if load == "d":
        sink = os.path.join(os.path.realpath(sess.right), LOAD_D_NAME)
        sess.load_proc = subprocess.Popen(load_d_command(sink), stdin=subprocess.DEVNULL,
                                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

        def probe():
            return os.path.getsize(sink) if os.path.exists(sink) else 0
        return probe, "dd of %d MiB to the USB stick, no Flea copy, the right pane lists its folder" % LOAD_D_MIB

    def right_listing():
        """(list requests, listing in flight) of the right pane, from the dualState IPC; (None, False) when unreadable."""
        try:
            # Sample input: {"active": true, "focused": 0, "panes": [{"path": "/a", "total": 3, "cursor": 0, "loading": false, "selected": [], "focused": true, "listRequests": 4}, null]} (ui/Ipc.qml dualState)
            panes = json.loads(sess.ipc("dualState") or "{}").get("panes") or []
            right = panes[1] or {}
            return right.get("listRequests"), bool(right.get("loading"))
        except (ValueError, IndexError, AttributeError, OSError, subprocess.SubprocessError):
            return None, False
    return right_listing, "NAS listing streaming in the right pane"


def run_series(sess, sp, n):
    """A series' rows; an error or an outside window ends it with the rows already built plus its error row; under a load every row names it and says whether it ran during that sample."""
    sess.taken = []  # the series build their rows in this list, so a failure keeps them
    try:
        rows = SERIES[sp["type"]](sess, sp, n)
    except ExternalWindow as e:
        rows = sess.taken + [e.row]
    except (RuntimeError, OSError, ValueError, subprocess.SubprocessError) as e:
        rows = sess.taken + [{"journey": sp["type"], "view": sess.view, "error": "%s: %s" % (type(e).__name__, e)}]
    for r in rows if sess.load else ():
        r.update({"load": sess.load.name, "load_note": sess.load.note})
        r.setdefault("load_probe", None)
        r.setdefault("load_active", False)
    return rows


def run_session(spec, ctx, out):
    """One session's series; rows go to out as they finish. A failure is an error row, never a gap."""
    where = spec.get("where", "local")
    right = None
    try:
        if spec.get("op"):
            setup_op_dirs(spec, ctx)
        fixture = spec["fixture"]
        if spec.get("load"):
            right = load_right(spec["load"], ctx)
            if right is None:
                raise RuntimeError("load %s needs a %s fixture" % (spec["load"], "USB" if spec["load"] in ("b", "d") else "NAS"))
        sess = Session(ctx, fixture, spec["view"], spec["region"], right,
                       os.path.join(ctx["out_dir"], "flea-%s.log" % ctx["tag"]), ui=spec.get("ui"))
    except (RuntimeError, OSError, subprocess.SubprocessError) as e:
        emit(out, {"journey": "session", "view": spec.get("view"), "fixture": where,
                   "error": "%s: %s" % (type(e).__name__, e)})
        if spec.get("load") in ("b", "d") and right is not None:
            drop_load_dst(right, ctx["roots"])
        return
    sess.right = right
    try:
        if spec.get("load"):
            probe, note = load_start(sess, spec["load"], ctx)
            sess.load = Load(spec["load"], note, probe, load_renew(sess, spec["load"], ctx), LOAD_GAP_S.get(spec["load"], 0.0))
            sess.settle(1.0, 5)
        for sp in spec["series"]:
            for r in run_series(sess, sp, spec["n"]):
                r.setdefault("fixture", where)
                emit(out, r)
    finally:
        load_stop(sess)
        sess.close()
        if spec.get("load") in ("b", "d") and right is not None:
            drop_load_dst(right, ctx["roots"])


def setup_op_dirs(spec, ctx):
    """ROOT/ops/run/<op>/{a-dst,b-src}; a-dst links to a fresh dir on the USB stick for a cross-fs op."""
    base = spec["fixture"]
    os.makedirs(os.path.join(base, "b-src"), exist_ok=True)
    name = spec["op"]
    dst = os.path.join(base, "a-dst")
    if name in OPS and OPS[name]["usb"]:
        if not ctx["fx"].get("usb"):
            raise RuntimeError("%s needs the USB fixture" % name)
        if not os.path.lexists(dst):
            os.symlink(tempfile.mkdtemp(prefix="dst-", dir=os.path.join(ctx["fx"]["usb"], "dst")), dst)
    else:
        os.makedirs(dst, exist_ok=True)
    if name == "op.size.recursive" and not os.path.exists(os.path.join(base, "b-src", "tree")):
        reflink(os.path.join(ctx["root"], "ops", "size", "tree"), os.path.join(base, "b-src"))
    if name == "op.archive" and not os.path.exists(os.path.join(base, "b-src", "src-dir")):
        reflink(os.path.join(ctx["root"], "ops", "src-archive", "src-dir"), os.path.join(base, "b-src"))
    spec["fixture"] = os.path.join(base, "b-src")
    ctx["op_base"] = base


def drop_load_dst(path, roots):
    """Remove a load-b or load-d session's own dst directory, by the same checks as every sandbox removal."""
    if not os.path.basename(path).startswith("dst-"):
        raise SystemExit("refusing to remove %s: not a load-b or load-d dst directory" % path)
    sandbox_rm(path, roots)


def load_right(load, ctx):
    """The right pane's path for a load: a fresh cold image copy, a USB destination, or the NAS stream folder."""
    if load == "a":
        d = tempfile.mkdtemp(prefix="cold-", dir=os.path.join(ctx["root"], "load"))
        for f in os.listdir(os.path.join(ctx["root"], "load", "cold-src")):
            os.link(os.path.join(ctx["root"], "load", "cold-src", f), os.path.join(d, f))
        return d
    if load in ("b", "d"):
        return tempfile.mkdtemp(prefix="dst-", dir=os.path.join(ctx["fx"]["usb"], "dst")) if ctx["fx"].get("usb") else None
    return os.path.join(ctx["fx"]["nas"], "stream") if ctx["fx"].get("nas") else None


def emit(out, row):
    out.write(json.dumps(row) + "\n")
    out.flush()


def dry_targets(sessions):
    """Per session, the pointer of a wheel or stroke and the Enter row of a folder cycle, resolved on a nominal 2560x1440 window."""
    nominal = (0, 0, 2560, 1440)
    out, bad = [], 0
    for s in sessions:
        if s.get("missing"):
            continue
        view, region = s.get("view"), win_rect(nominal, s.get("region", "content"))
        for sp in s["series"]:
            t = {"view": view, "fixture": os.path.basename(s["fixture"]), "series": sp["type"]}
            if sp["type"] in ("wheel", "fling"):
                t["pointer"] = list(scroll_point(view, region, win_rect(nominal, "left")))
                t["aim"] = "centre of the left pane" if view == "dual" else "centre of the watched region"
            elif sp["type"] == "cycle" and sp.get("to_folder"):
                try:
                    keys = folder_setup(ui_patch(view, s["fixture"]), s["fixture"])
                    t["enter_index"] = len(keys) - 1
                    t["aim"] = "the first folder, row 0 with folders first"
                    t["setup"] = [c[1][0] for c in keys]
                except (RuntimeError, OSError) as e:
                    t["enter_index"], t["error"] = None, "%s: %s" % (type(e).__name__, e)
                    bad += 1
            else:
                continue
            if s.get("load"):
                t["load"] = s["load"]
            out.append(t)
    return out, bad


def run_part(part, root, tree, out_dir, samples=20, dry=False, evidence=True, only=None):
    with open(os.path.join(os.path.realpath(root), "fixtures.json")) as f:
        fx = json.load(f)
    sessions = parts(fx, samples)[part]
    if only is not None:
        sessions = only_series(sessions, only)
        if not sessions:
            print("--only %r matches no journey of part %s" % (only, part), file=sys.stderr)
            return 2
    if dry:
        js = sorted(journeys_of(sessions))
        unmapped = [j for j in js if row_of(j) is None]
        targets, bad = dry_targets(sessions)
        print(json.dumps({"part": part, "only": only, "sessions": len(sessions), "journeys": js, "unmapped": unmapped,
                          "targets": targets, "estimate_min": ESTIMATE_MIN[part]}, indent=1))
        return 1 if unmapped or bad else 0
    os.makedirs(out_dir, exist_ok=True)
    S.ENV = S.session_env()
    os.environ.update(S.ENV)
    S.UI_TOKENS[:] = [tree + "/ui"]
    S.require_idle()
    keys = N.NavKeys()
    ctx = {"tree": tree, "root": fx["root"], "fx": fx, "sbx": fx["root"], "out_dir": out_dir, "keys": keys,
           "states": {}, "evidence": evidence, "config_home": os.environ.get("XDG_CONFIG_HOME") or
           os.path.expanduser("~/.config"),
           "roots": [fx["root"]] + [fx[x] for x in ("usb", "nas") if fx.get(x)]}
    ctx["rec_bin"] = S.build_recorder(os.path.join(out_dir, "rec"))
    plan = {"part": part, "root": root, "flea": tree, "out": out_dir, "samples": samples}
    if only is not None:
        plan["only"] = only
    ctx["repeat"] = repeat_settings(plan)
    if part in ("p2-held", "smoke"):
        ctx["repeat"].update(repeat_probe(keys, ctx["repeat"]))
    mon = S.monitor()
    rev = subprocess.run(["git", "-C", tree, "rev-parse", "HEAD"], capture_output=True, text=True)
    head = rev.stdout.strip()
    if rev.returncode != 0 or not head:
        raise SystemExit("flea tree %s is not a git tree, git rev-parse HEAD says: %s" % (tree, rev.stderr.strip() or "nothing"))
    out_path = os.path.join(out_dir, "census-%s.jsonl" % part)
    with open(out_path, "x") as out:
        emit(out, {"manifest": {"date": time.strftime("%Y-%m-%d %H:%M:%S"), "host": socket.gethostname(),
                                "plan": plan, "flea_head": head, "repeat": ctx["repeat"],
                                "monitor": {x: mon.get(x) for x in ("name", "width", "height", "refreshRate", "scale")},
                                "quiet_s": QUIET, "cap_s": CAP, "pace_ms": PACE_MS, "budget": BUDGET}})
        for i, spec in enumerate(sessions):
            if spec.get("missing"):
                emit(out, {"journey": "fixture", "fixture": spec["missing"], "first_ms": "no instrument",
                           "note": "no %s fixture in fixtures.json" % spec["missing"]})
                continue
            ctx["tag"] = "%s-s%02d" % (part, i)
            t0 = time.monotonic()
            run_session(dict(spec), ctx, out)
            print("%s session %d/%d %s %s done in %.0f s" % (part, i + 1, len(sessions), spec.get("view"),
                                                            os.path.basename(spec["fixture"]), time.monotonic() - t0),
                  flush=True)
    print(out_path)
    return 0


# Report.


def fmt(v):
    return "miss" if v is None else ("%.1f" % v)


GATE_MIN_SAMPLES = 20  # rows per journey at 20 samples when no manifest names the part


def aggregate(paths):
    """Census JSONL paths into grouped numeric rows, notes, errors and manifests, shared by report and gate."""
    rows, manifests = [], []
    for p in paths:
        with open(p) as f:
            for line in f:
                d = json.loads(line)
                (manifests.append(d["manifest"]) if "manifest" in d else rows.append(d))
    groups, notes, errors = {}, [], []
    for r in rows:
        if "error" in r:
            errors.append(r)
            continue
        if isinstance(r.get("first_ms"), str):
            notes.append(r)
            continue
        fixture = r.get("fixture", "local") + ("+load-" + r["load"] if r.get("load") else "")
        groups.setdefault((r["journey"], r.get("view", "-"), fixture), []).append(r)
    return groups, notes, errors, manifests


def case_of(key, g):
    """One grouped journey with its budget verdict, the numbers both report and gate print."""
    journey, view, fixture = key
    row = row_of(journey, fixture.split("+")[0]) or g[0].get("row")
    firsts = [r.get("first_ms") for r in g]
    settled = [r.get("settled_ms") for r in g]
    held = [r["held"] for r in g if r.get("held")]
    video = journey.split(".")[-1] == "video" or journey.endswith("video.first")
    return {"key": key, "journey": journey, "view": view, "fixture": fixture, "row": row, "firsts": firsts,
            "settled": settled, "held": held, "video": video,
            "verdict": judge(row, firsts, settled, video, held)}


def fling_floor_of(path):
    """(floor, why) from a calibrate output of concatenated JSON objects: its last "fling" object, kept when a later part does not
    parse; floor is None and why names the file and the cause when the file is unreadable, holds no fling object or does not parse."""
    try:
        with open(path, errors="replace") as f:
            text = f.read()
    except OSError as e:
        return None, "%s: %s" % (path, e.strerror or e)
    dec, i, found, bad = json.JSONDecoder(), 0, None, None
    while True:
        while i < len(text) and text[i].isspace():
            i += 1
        if i >= len(text):
            break
        try:
            obj, i = dec.raw_decode(text, i)
        except json.JSONDecodeError as e:
            bad = e.pos
            break
        if isinstance(obj, dict) and isinstance(obj.get("fling"), dict):
            found = obj["fling"]
    if found is not None:
        return found, None
    return None, "%s: %s" % (path, "no fling object" if bad is None else "parse error at offset %d" % bad)


def fling_floor_note(floor):
    """The fling floor as a report note, beside the fling rows; the ceiling less the floor is the room left for Flea. Never judged."""
    if not floor:
        return None
    if "first_frame_ms" not in floor:
        return "fling floor not read (%s)" % (floor.get("skipped") or floor.get("error") or "no runs")
    w, f = floor["first_wheel_ms"], floor["first_frame_ms"]
    if f["median"] is None:
        return "fling floor not read: no run drew a frame"
    left = BUDGET[1]["ceiling_ms"] - f["median"]
    return ("idle client floor, stroke to first wheel event p50 %s (%s to %s), to first frame p50 %s (%s to %s) over %d runs, "
            "%.1f left under the %s ms ceiling, not subtracted" % (
                fmt(w["median"]), fmt(w["min"]), fmt(w["max"]), fmt(f["median"]), fmt(f["min"]), fmt(f["max"]),
                floor["runs"], left, BUDGET[1]["ceiling_ms"]))


def report(paths, out=None, floor=None):
    """The census table: one line per (journey, view, fixture, load), with the budget row's target and verdict; floor is a calibrate "fling" object."""
    groups, notes, errors, manifests = aggregate(paths)
    lines = ["# Instant budget census", ""]
    for m in manifests:
        lines.append("- %s on %s, Flea %s, part %s, repeat %s ms delay at %s a second (%s)" % (
            m.get("date"), m.get("host"), (m.get("flea_head") or "?")[:10], m["plan"]["part"],
            m["repeat"]["delay_ms"], m["repeat"]["rate"], m["repeat"].get("source")))
    lines += ["", "Times are ms from the input to the first frame whose watched region changes with ink (first_ms), "
              "p50 and p95 nearest rank, a miss (nothing drawn before the next input or the cap) ranked above "
              "every number. Row 2 lists the largest gap between drawn frames in each sweep; its verdict reads the "
              "repeat slots and the overshoot. Row 5 lists first_ms; its verdict also reads settled_ms.", "",
              "| row | journey | view | fixture | n | p50 | p95 | max | target | ceiling | verdict | notes |",
              "|---|---|---|---|---|---|---|---|---|---|---|---|"]
    order = sorted(groups, key=lambda g: (row_of(g[0], g[2].split("+")[0]) or 9, g[0], g[1], g[2]))
    tally = {}
    for key in order:
        g = groups[key]
        c = case_of(key, g)
        journey, view, fixture = c["journey"], c["view"], c["fixture"]
        row, firsts, settled, held, video, verdict = c["row"], c["firsts"], c["settled"], c["held"], c["video"], c["verdict"]
        tally[verdict] = tally.get(verdict, 0) + 1
        b = BUDGET[row]
        note = []
        missed = sum(1 for v in firsts if v is None)
        if missed:
            note.append("%d miss" % missed)
        if any(r.get("refused") for r in g):
            note.append("%d refused" % sum(1 for r in g if r.get("refused")))
        if any(r.get("capped") for r in g):
            note.append("%d capped" % sum(1 for r in g if r.get("capped")))
        if any(r.get("pre_quiet") is False for r in g):
            note.append("%d not quiet before" % sum(1 for r in g if r.get("pre_quiet") is False))
        if row == 2:
            note.append("slots missed %s, overshoot %s, behind %s" % (
                ",".join(str(h["slots_missed"]) for h in held), ",".join(str(h["overshoot"]) for h in held),
                ",".join(str(h["steps_behind"]) for h in held)))
        if row in (5, 6):
            st = [v for v in settled if v is not None]
            if st:
                note.append("settled p95 %s" % fmt(p95(settled)))
        if journey.endswith(".fling") and fling_floor_note(floor):
            note.append(fling_floor_note(floor))
        if any(r.get("load") for r in g):
            note.append("load active %d/%d" % (sum(1 for r in g if r.get("load_active")), len(g)))
        if any(r.get("evidence") for r in g):
            note.append("evidence %s" % os.path.basename(next(r["evidence"] for r in g if r.get("evidence"))))
        if any(r.get("evidence_11s") for r in g):
            note.append("11 s frame %s" % os.path.basename(next(r["evidence_11s"] for r in g if r.get("evidence_11s"))))
        target = b["target_ms"] if row != 2 else "25 slots"
        ceiling = b["ceiling_ms"] if row not in (2, 5) else ("0 overshoot" if row == 2 else "100 / 1000")
        lines.append("| %d | %s | %s | %s | %d | %s | %s | %s | %s | %s | %s | %s |" % (
            row, journey, view, fixture, len(g), fmt(p50(firsts)), fmt(p95(firsts)),
            fmt(max(firsts, key=lambda v: math.inf if v is None else v)), target, ceiling, verdict, "; ".join(note)))
    lines += ["", "Verdicts: " + ", ".join("%s %d" % kv for kv in sorted(tally.items()))]
    for r in notes:
        lines.append("- %s %s %s: %s (%s)" % (r.get("journey"), r.get("view", ""), r.get("fixture", ""),
                                             r["first_ms"], r.get("note", "")))
    for r in errors:
        lines.append("- %s %s %s %s: %s" % ("harness fault" if r.get("harness_fault") else "error", r.get("journey"),
                                          r.get("view", ""), r.get("fixture", ""), r["error"]))
    text = "\n".join(lines) + "\n"
    if out:
        with open(out, "w") as f:
            f.write(text)
    print(text, end="")
    return 0


def gate_ceiling(row, case):
    """Numeric ceiling of a BREACH line: row 5 names the settled limit when only it is over."""
    if row == 2:
        return 0
    if row == 5:
        f95 = p95(case["firsts"])
        if f95 is not None and f95 <= BUDGET[5]["ceiling_ms"]:
            return BUDGET[5]["content_ceiling_ms"]
        return BUDGET[5]["ceiling_ms"]
    return BUDGET[row]["ceiling_ms"]


def gate_shown_p95(row, case):
    """p95 of a BREACH line: row 5 names the settled p95 when only it is over."""
    if row == 5:
        f95 = p95(case["firsts"])
        if f95 is not None and f95 <= BUDGET[5]["ceiling_ms"]:
            return p95(case["settled"])
    return p95(case["firsts"])


def resolve_part(name):
    """A part name from --part: a full PARTS name or a pN short form; None when unknown."""
    if name in PARTS:
        return name
    hits = [p for p in PARTS if p.startswith(name + "-")]
    return hits[0] if len(hits) == 1 else None


def gate(paths, out=None, only=None, pattern=None):
    """Pass or fail the census against the budget ceilings; a missing journey never passes, except one a --only pattern drops."""
    if not paths:
        print("need census .jsonl files, got none", file=sys.stderr)
        return 2
    # A part named twice (p1 and p1-cursor) is planned once, so its minima never double.
    expected = PARTS if only is None else list(dict.fromkeys(resolve_part(p) or p for p in only))
    groups, notes, errors, manifests = aggregate(paths)
    order = sorted(groups, key=lambda g: (row_of(g[0], g[2].split("+")[0]) or 9, g[0], g[1], g[2]))
    need = {}
    for part in expected:
        for key, n in planned_counts(part, GATE_MIN_SAMPLES, PLAN_FX, pattern).items():
            need[key] = need.get(key, 0) + n
    seen_parts, sample_lines = set(), []
    part_manifests = {}
    for m in manifests:
        plan = m.get("plan", {})
        part = plan.get("part")
        if part in PARTS:
            seen_parts.add(part)
            part_manifests[part] = part_manifests.get(part, 0) + 1
            if not m.get("flea_head"):
                sample_lines.append("GATE MISSING %s manifest has an empty flea_head" % part)
            s = plan.get("samples", GATE_MIN_SAMPLES)
            if s < GATE_MIN_SAMPLES:
                sample_lines.append("GATE MISSING %s plan samples %s, under the defined %s" % (part, s, GATE_MIN_SAMPLES))
    for part in sorted(part_manifests):
        if part_manifests[part] > 1:
            sample_lines.append("GATE MISSING part %s has %d manifests, so its rows came from more than one run" % (part, part_manifests[part]))
    breaches, missing, seen = [], list(sample_lines), set()
    for part in expected:
        if part not in seen_parts:
            missing.append("GATE MISSING part %s" % part)
    for key in order:
        g = groups[key]
        seen.add(key)
        c = case_of(key, g)
        want = need.get(key, GATE_MIN_SAMPLES)
        if len(g) < want:
            missing.append("GATE MISSING %s %s %s n=%d needs %d" % (
                c["journey"], c["view"], c["fixture"], len(g), want))
            continue
        if c["verdict"] == "NO-READ":
            missing.append("GATE MISSING %s %s %s no readable samples n=%d" % (
                c["journey"], c["view"], c["fixture"], len(g)))
            continue
        idle = sum(1 for r in g if not r.get("load_active")) if "+load-" in c["fixture"] else 0
        if idle:
            missing.append("GATE MISSING %s %s %s load inactive in %d of %d samples" % (
                c["journey"], c["view"], c["fixture"], idle, len(g)))
        if c["verdict"] == "BREACH":
            breaches.append("GATE BREACH %s %s %s p95=%s ceiling=%s" % (
                c["journey"], c["view"], c["fixture"], fmt(gate_shown_p95(c["row"], c)), gate_ceiling(c["row"], c)))
    for r in notes:
        key = (r.get("journey", "?"), r.get("view", ""), r.get("fixture", ""))
        seen.add(key)
        missing.append("GATE MISSING %s %s %s %s (%s)" % (key[0], key[1], key[2], r.get("first_ms"), r.get("note", "")))
    for r in errors:
        key = (r.get("journey", "?"), r.get("view", ""), r.get("fixture", ""))
        seen.add(key)
        missing.append("GATE MISSING %s %s %s error %s" % (key[0], key[1], key[2], r.get("error")))
    for key in sorted(need, key=lambda g: (row_of(g[0], g[2].split("+")[0]) or 9, g[0], g[1], g[2])):
        if key not in seen:
            missing.append("GATE MISSING %s %s %s n=0 needs %d" % (key[0], key[1], key[2], need[key]))
    if breaches or missing:
        summary = "GATE FAIL breaches=%d journeys=%d" % (len(breaches), len(seen))
        code = 2 if missing else 1
    else:
        summary = "GATE PASS journeys=%d" % len(seen)
        code = 0
    text = "\n".join(breaches + missing + [summary]) + "\n"
    if out:
        with open(out, "w") as f:
            f.write(text)
    print(text, end="")
    return code


def budget_json():
    print(json.dumps({"budget": {str(k): v for k, v in BUDGET.items()},
                      "journeys": [{"pattern": p, "row": r} for p, r in JOURNEYS],
                      "remote_row": 5, "metric": "p95 first_ms, nearest rank, a miss above every number"}, indent=1))
    return 0


# calibrate and selftest.


def calibrate_gate(neg, rep):
    # Pass on a clean negative control, the 50 ms row, an unspent delay-0 capture and every row median in one frame.
    rows = rep.get("rows", [])
    if "frame_ms" not in rep:
        print("no frame_ms", file=sys.stderr)
        return 1
    frame = rep["frame_ms"]
    fifty = next((r for r in rows if r.get("delay_ms") == 50), None)
    if fifty is None:
        print("no 50 ms row", file=sys.stderr)
        return 1
    if "within_one_frame" not in fifty:
        print("50 ms row unscored", file=sys.stderr)
        return 1
    ok = bool(neg.get("clean", False)) and bool(fifty.get("within_one_frame", False))
    if not fifty.get("within_one_frame", False):
        errs = fifty.get("error_vs_client_ms", [])
        med = "%.1f" % statistics.median(errs) if errs else "none"
        print("fifty_outside_one_frame delay=50 median=%s runs=%s" % (
            med, [i + 1 for i, e in enumerate(errs) if abs(e) > frame]), file=sys.stderr)
    # A delay-0 result after another commit was read a recorder round trip late, so the floor it sets is not the floor.
    zero = next((r for r in rows if r.get("delay_ms") == 0), None)
    spent = [i + 1 for i, n in enumerate((zero or {}).get("frames_before_result", [])) if n > 0]
    if spent:
        print("capture_spent delay=0 runs=%s" % spent, file=sys.stderr)
        ok = False
    for r in rows:
        if r.get("delay_ms") == 50 or r.get("within_one_frame", False):
            continue
        if "error_vs_client_ms" not in r or "within_one_frame" not in r:
            print("unscored delay=%s" % (r.get("delay_ms"),), file=sys.stderr)
            continue
        errs = r.get("error_vs_client_ms", [])
        # A row whose median misses by more than a frame is an instrument error, not a stray run.
        if errs and abs(statistics.median(errs)) > frame:
            print("median_outside_one_frame delay=%s median=%.1f" % (r.get("delay_ms"), statistics.median(errs)),
                  file=sys.stderr)
            ok = False
        bad = [i + 1 for i, e in enumerate(errs) if abs(e) > frame]
        print("outside_one_frame delay=%s runs=%s error_vs_client_ms=%s" % (
            r.get("delay_ms"), bad, [errs[i - 1] for i in bad]), file=sys.stderr)
    return 0 if ok else 1


def calibrate(runs):
    """Also prints the fling floor (fling_probe) under "fling", which never changes the verdict. Pass when the negative control is clean, the 50 ms row is within one frame, no delay-0 result followed another commit and every scored row's median error is within one frame; each other run outside one frame prints to stderr as outside_one_frame with its delay, runs and error_vs_client_ms without failing the gate."""
    S.ENV = S.session_env()
    os.environ.update(S.ENV)
    neg = negative_control(3)
    print(json.dumps({"negative_control": neg}, indent=1), flush=True)
    rep = S.calibrate([0, 50, 200], runs)
    keys = N.NavKeys()
    rp = repeat_probe(keys, repeat_settings({}), 3)
    print(json.dumps({"repeat_probe": rp}, indent=1), flush=True)
    # The fling floor is reported, never judged: no failure here changes the exit code.
    try:
        fling = fling_probe(runs, keys)
    except Exception as e:
        fling = {"error": "%s: %s" % (type(e).__name__, e)}
    print(json.dumps({"fling": fling}, indent=1))
    return calibrate_gate(neg, rep)


def selftest():
    fail = []

    def check(name, got, want):
        if got != want:
            fail.append("%s: got %r, wanted %r" % (name, got, want))

    ms = MS
    check("row_cursor", row_of("nav.grid.l"), 1)
    check("row_held", row_of("nav.dual.held-j-sweep"), 2)
    check("row_enter", row_of("nav.columns.enter"), 3)
    check("row_view", row_of("view.switch.grid-dual"), 3)
    check("row_open", row_of("open.pathjump"), 4)
    check("row_ql_txt", row_of("open.ql.txt"), 4)
    check("row_ql_first", row_of("open.ql.md.first"), 4)
    check("row_heavy", row_of("open.cols-preview.video"), 5)
    check("row_usb", row_of("nav.list.enter", "usb"), 5)
    check("row_op", row_of("op.archive"), 6)
    check("row_none", row_of("theme.flip"), None)
    check("p95_20", p95(list(range(1, 21))), 19)
    check("p95_miss", p95([1] * 19 + [None]), 1)
    check("p95_two_miss", p95([1] * 18 + [None, None]), None)
    check("p50", p50([5, 1, 3, 2, 4]), 3)
    check("judge_target", judge(1, [5.0] * 20), "TARGET")
    check("judge_ceiling", judge(1, [5.0] * 18 + [12.0, 12.0]), "CEILING-OK")
    check("judge_breach", judge(3, [40.0] * 18 + [140.0, 150.0]), "BREACH")
    check("judge_miss", judge(1, [5.0] * 18 + [None, None]), "BREACH")
    check("judge_heavy_target", judge(5, [30.0] * 20, [80.0] * 20), "TARGET")
    check("judge_heavy_ok", judge(5, [30.0] * 20, [600.0] * 20), "CEILING-OK")
    check("judge_heavy_slow", judge(5, [30.0] * 20, [1600.0] * 20), "BREACH")
    check("judge_heavy_late", judge(5, [130.0] * 20, [130.0] * 20), "BREACH")
    check("judge_video", judge(5, [60.0] * 20, [5000.0] * 20, video=True), "TARGET")
    check("judge_held_target", judge(2, [], held=[{"overshoot": 0, "slots_missed": 0}] * 3), "TARGET")
    check("judge_held_slots", judge(2, [], held=[{"overshoot": 0, "slots_missed": 2}]), "CEILING-OK")
    check("judge_held_over", judge(2, [], held=[{"overshoot": 1, "slots_missed": 0}]), "BREACH")
    check("judge_held_noread", judge(2, [], held=[{"overshoot": None, "slots_missed": 0}]), "NO-READ")
    # Repeats due before a release half a period past repeat 70: delay 250, every 25 ms, so 71 of them.
    hold = int((250 + 25 * HOLD_REPEATS + 12.5) * ms)
    check("repeats_due", repeats_due(0, hold, 250, 25), 71)
    # A sweep drawn every repeat 6 ms late, one frame per repeat: no slot missed, gaps of 25 ms.
    frames = [(0, -1, -1), (6 * ms, 40, 60)]
    frames += [(int((250 + 25 * kk + 6) * ms), 40, 60) for kk in range(71)]
    h = held_analysis(frames, 0, hold, 250, 25)
    check("held_slots", h["slots_missed"], 0)
    check("held_gap", h["max_gap_ms"], 25.0)
    check("held_first", h["first_repeat_ms"], 6.0)
    check("held_press", h["press_first_ms"], 6.0)
    # Every fifth repeat not drawn: 14 slots missed and a 50 ms gap.
    frames2 = [f for i, f in enumerate(frames) if i < 3 or (i - 2) % 5 != 0]
    h2 = held_analysis(frames2, 0, hold, 250, 25)
    check("held_missed", h2["slots_missed"], 14)
    check("held_gap_missed", h2["max_gap_ms"], 50.0)
    # Nothing drawn after the press: every slot missed, no gap.
    h3 = held_analysis([(0, -1, -1), (6 * ms, 40, 60)], 0, hold, 250, 25)
    check("held_none", (h3["slots_missed"], h3["max_gap_ms"]), (71, None))
    # Overshoot: 72 events at most, 75 steps is 3 past; a grid stride of 6 divides.
    check("over_exact", overshoot(0, 72, 1, 72), (0, 0))
    check("over_backlog", overshoot(0, 75, 1, 72), (3, 0))
    check("over_behind", overshoot(0, 60, 1, 72), (0, 12))
    check("over_grid", overshoot(0, 6 * 73, 6, 72), (1, 0))
    check("over_noread", overshoot(None, 70, 1, 72), (None, None))
    # Every chord the parts send has a keycode, and every pair names its two journeys.
    for a, b in PAIRS + GRID_PAIRS:
        for c in (a, b):
            for key in c:
                check("key %s" % key, key in S.KEYCODES or key in N.EXTRA_KEYCODES, True)
            check("name %s" % (c,), c in CHORD_NAMES, True)
    for key in ("t", "w", "[", "]", "m", "r", ",", "/", ";", "f", "space", "esc", "tab", "delete", "backspace",
                "enter", "right", "left", "home", "end", "1", "2", "3", "v", "c", "x", "a"):
        check("key %s" % key, key in S.KEYCODES or key in N.EXTRA_KEYCODES, True)
    # The view circuit holds each of the 12 ordered pairs exactly once.
    pairs_seen = list(zip(VIEW_CIRCUIT, VIEW_CIRCUIT[1:]))
    check("circuit_len", len(pairs_seen), 12)
    check("circuit_distinct", len(set(pairs_seen)), 12)
    check("circuit_no_self", all(a != b for a, b in pairs_seen), True)
    # Every part builds and maps every journey to a row, from a synthetic fixtures facts.
    fx = {"root": "/home/flea-sandbox/x", "nav": {"nav500": 500}, "usb": "/run/media/u/s/flea-instant-1",
          "nas": "/run/user/1/gvfs/n/f-2"}
    built = parts(fx, 20)
    check("parts", sorted(built), sorted(PARTS))
    for part, sessions in built.items():
        js = journeys_of(sessions)
        check("part %s has journeys" % part, bool(js) or part == "", True)
        check("part %s unmapped" % part, [j for j in js if row_of(j) is None], [])
    cover = set().union(*(journeys_of(s) for s in built.values()))
    for want in ("nav.list.down", "nav.grid.left", "nav.dual.wheel", "nav.columns.fling", "nav.grid.held-j-sweep",
                 "nav.columns.shift-right", "view.switch.dual-columns", "tab.new", "tab.switch", "pane.switch.dual",
                 "open.menu.key", "open.menu.click", "open.rename", "open.settings", "open.palette", "open.pathjump",
                 "search.type-to-first-result", "open.ql.code", "open.cols-preview.md", "open.ql.video",
                 "open.cols-preview.large-image", "op.copy.many", "op.copy.large", "op.move.cross-fs", "op.trash",
                 "op.size.recursive", "op.archive", "op.copy.long"):
        check("covers %s" % want, want in cover, True)
    check("cover_size", len(cover) >= 100, True)
    # p8-rename: a session per view, the four rename journeys of each, the rename keys in order, the names restored.
    rn, views = built["p8-rename"], ["list", "columns", "grid", "dual"]
    check("rename_views", [s["view"] for s in rn], views)
    check("rename_journeys", journeys_of(rn), {"nav.%s.%s-rename-%s" % (v, side, key) for v in views
                                              for side in ("pre", "after") for key in ("j", "k")})
    check("rename_row", row_of("nav.columns.after-rename-k"), 1)
    want = sorted(NAV500_NAME % i for i in range(500))
    check("rename_restores", [(sp.get("restores"), sp.get("typed")) for s in rn for sp in s["series"]],
          [(want, "ab")] * 4)
    steps = rn[0]["series"][0]["steps"]
    untimed = [st["in"][1] for st in steps if "journey" not in st]
    check("rename_keys", untimed, [("r",), ("right",), ("a",), ("b",), ("enter",), ("r",), ("right",), ("backspace",),
                                   ("backspace",), ("enter",)])
    check("rename_pairs", [st["journey"].rsplit(".", 1)[1] for st in steps if "journey" in st],
          ["pre-rename-j", "pre-rename-k"] * RENAME_PAIRS + ["after-rename-j", "after-rename-k"] * RENAME_PAIRS)
    for chord in untimed + [i[1] for i in rn[0]["series"][0]["setup"]]:
        check("rename key %s" % (chord,), all(c in S.KEYCODES or c in N.EXTRA_KEYCODES for c in chord), True)
    check("rename_plan", planned_counts("p8-rename", 20)[("nav.grid.after-rename-j", "grid", "local")],
          20 * RENAME_PAIRS)
    # Kinds sort in the cursor order the series walk.
    check("kinds_sorted", sorted(KINDS.values())[:3], ["a_photo.jpg", "c_doc.pdf", "d_notes.md"])
    # Sandbox guard: a path outside a marked root is refused.
    with tempfile.TemporaryDirectory() as td:
        root = os.path.join(td, "r" * 12)
        os.makedirs(os.path.join(root, "a"))
        check("unmarked", inside_marked(os.path.join(root, "a"), [root]), False)
        open(os.path.join(root, MARKER), "w").close()
        check("marked", inside_marked(os.path.join(root, "a"), [root]), True)
        check("root_itself", inside_marked(root, [root]), False)
        check("outside", inside_marked(td, [root]), False)
        os.symlink(td, os.path.join(root, "link"))
        check("link_itself", inside_marked(os.path.join(root, "link"), [root]), True)
        sandbox_rm(os.path.join(root, "link"), [root])
        check("link_removed_target_kept", (os.path.lexists(os.path.join(root, "link")), os.path.isdir(td)),
              (False, True))
        try:
            sandbox_rm(td, [root])
            fail.append("sandbox_rm removed an unmarked path")
        except SystemExit:
            pass
    # Report: a cursor series at target, a navigate breach, a held sweep, a heavy row and a no-instrument row.
    import io
    from contextlib import redirect_stderr, redirect_stdout
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(json.dumps({"manifest": {"date": "d", "host": "h", "flea_head": "c7e23485e", "plan": {"part": "x"},
                                          "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}) + "\n")
        for i in range(20):
            tf.write(json.dumps({"journey": "nav.list.down", "row": 1, "view": "list", "first_ms": 5.0 + i * 0.1,
                                 "fixture": "local"}) + "\n")
            tf.write(json.dumps({"journey": "nav.grid.enter", "row": 3, "view": "grid", "first_ms": 140.0 + i,
                                 "fixture": "local"}) + "\n")
            tf.write(json.dumps({"journey": "open.ql.pdf", "row": 5, "view": "list", "first_ms": 40.0,
                                 "settled_ms": 300.0, "fixture": "local"}) + "\n")
        tf.write(json.dumps({"journey": "nav.list.held-j-sweep", "row": 2, "view": "list", "first_ms": 25.0,
                             "fixture": "local", "held": {"slots_missed": 0, "overshoot": 0, "steps_behind": 1}}) + "\n")
        tf.write(json.dumps({"journey": "nav.list.fling", "row": 1, "view": "list", "first_ms": "no instrument",
                             "note": "why", "fixture": "local"}) + "\n")
        tf.write(json.dumps({"journey": "op.trash", "view": "list", "error": "RuntimeError: x"}) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        report([tf.name])
    os.unlink(tf.name)
    text = buf.getvalue()
    check("rep_header", "| row | journey | view | fixture | n | p50 | p95 | max | target | ceiling | verdict | notes |"
          in text, True)
    check("rep_cursor", "| 1 | nav.list.down | list | local | 20 | 5.9 | 6.8 | 6.9 | 8.33 | 16.7 | TARGET |" in text, True)
    check("rep_breach", "| 3 | nav.grid.enter | grid | local | 20 | 149.0 | 158.0 | 159.0 | 50 | 100 | BREACH |"
          in text, True)
    check("rep_held", "| 2 | nav.list.held-j-sweep | list | local | 1 | 25.0 | 25.0 | 25.0 | 25 slots | 0 overshoot "
          "| TARGET |" in text, True)
    check("rep_heavy", "| 5 | open.ql.pdf | list | local | 20 | 40.0 | 40.0 | 40.0 | 100 | 100 / 1000 | CEILING-OK |"
          in text, True)
    check("rep_noinst", "nav.list.fling list local: no instrument (why)" in text, True)
    check("rep_error", "error op.trash list : RuntimeError: x" in text, True)
    check("rep_order", text.index("nav.list.down") < text.index("held-j-sweep") < text.index("nav.grid.enter"), True)
    # The rename rows sit among their own view's cursor rows, and a view's rows never interleave with another's.
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(json.dumps({"manifest": {"date": "d", "host": "h", "flea_head": "c7e23485e", "plan": {"part": "x"},
                                          "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}) + "\n")
        for v, name in (("grid", "down"), ("list", "pre-rename-k"), ("list", "down"), ("list", "after-rename-j"),
                        ("grid", "after-rename-k")):
            tf.write(json.dumps({"journey": "nav.%s.%s" % (v, name), "row": 1, "view": v, "first_ms": 5.0,
                                 "fixture": "local"}) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        report([tf.name])
    os.unlink(tf.name)
    check("rep_rename_order", re.findall(r"\| (nav\.[a-z]+\.[a-z-]+) \|", buf.getvalue()),
          ["nav.grid.after-rename-k", "nav.grid.down", "nav.list.after-rename-j", "nav.list.down",
           "nav.list.pre-rename-k"])
    # Gate: the same verdicts as report, one BREACH line per journey over ceiling, missing never passes.
    def run_gate(rows, part="x", samples=20, only=None, pattern=None, head="c7e23485e"):
        mani2 = {"manifest": {"date": "d", "host": "h", "flea_head": head,
                              "plan": {"part": part, "samples": samples},
                              "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}
        with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
            tf.write(json.dumps(mani2) + "\n")
            for r in rows:
                tf.write(json.dumps(r) + "\n")
        buf = io.StringIO()
        want_only = only if only is not None else ([part] if part in PARTS else [])
        with redirect_stdout(buf):
            code = gate([tf.name], only=want_only, pattern=pattern)
        os.unlink(tf.name)
        return code, buf.getvalue()

    under = [{"journey": "nav.list.down", "row": 1, "view": "list", "first_ms": 5.0, "fixture": "local"}
             for _ in range(20)]
    code, gtext = run_gate(under)
    check("gate_pass_code", code, 0)
    check("gate_pass_line", "GATE PASS journeys=1" in gtext, True)
    check("gate_pass_quiet", ("GATE BREACH" in gtext, "GATE MISSING" in gtext), (False, False))
    buf = io.StringIO()
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(json.dumps({"manifest": {"date": "d", "host": "h", "flea_head": "c7e23485e",
                                           "plan": {"part": "x", "samples": 20},
                                           "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}) + "\n")
        for r in under:
            tf.write(json.dumps(r) + "\n")
    with redirect_stdout(buf):
        report([tf.name])
    os.unlink(tf.name)
    check("gate_pass_report", "TARGET" in buf.getvalue(), True)
    over = under + [{"journey": "nav.grid.enter", "row": 3, "view": "grid", "first_ms": 140.0, "fixture": "local"}
                    for _ in range(20)]
    code, gtext = run_gate(over)
    check("gate_breach_code", code, 1)
    check("gate_breach_line", "GATE BREACH nav.grid.enter grid local p95=140.0 ceiling=100" in gtext, True)
    check("gate_breach_fail", "GATE FAIL breaches=1 journeys=2" in gtext, True)
    short = under + [{"journey": "nav.list.up", "row": 1, "view": "list", "first_ms": 5.0, "fixture": "local"}
                     for _ in range(3)]
    short.append({"journey": "nav.list.fling", "row": 1, "view": "list", "first_ms": "no instrument",
                  "note": "why", "fixture": "local"})
    code, gtext = run_gate(short)
    check("gate_missing_code", code, 2)
    check("gate_missing_short", "GATE MISSING nav.list.up list local n=3 needs 20" in gtext, True)
    check("gate_missing_noinst", "GATE MISSING nav.list.fling list local no instrument" in gtext, True)
    check("gate_missing_fail", ("GATE FAIL" in gtext, "GATE PASS" in gtext), (True, False))
    # Minima from the sampling plan: .first journeys need one row, plain ones n (n-1 by a .first).
    p7 = planned_counts("p7-load", 20)
    check("plan_p7_first", p7[("open.ql.txt.first", "dual", "local+load-a")], 1)
    check("plan_p7_plain", p7[("open.ql.txt", "dual", "local+load-a")], 19)
    check("plan_p7_pairs", p7[("nav.dual.down", "dual", "local+load-b")], 20)
    check("plan_p7_pairs_d", (p7[("nav.dual.down", "dual", "local+load-d")], p7[("nav.dual.up", "dual", "local+load-d")]), (20, 20))
    check("plan_p7_d_pairs_only", sorted(k[0] for k in p7 if k[2] == "local+load-d"), ["nav.dual.down", "nav.dual.up"])
    check("plan_p7_d_beside_b", "".join(sp["load"] for sp in parts(PLAN_FX, 20)["p7-load"]), "aaaabbbbdcccc")
    check("load_d_command", load_d_command("/u/dst-x/load-d.bin"),
          ["dd", "if=/dev/zero", "of=/u/dst-x/load-d.bin", "bs=4M", "count=1000", "oflag=direct"])
    class Holder:
        pass
    writer = Holder()
    writer.load_proc = subprocess.Popen(["sleep", "60"])
    load_stop(writer)
    check("load_d_stopped", writer.load_proc.poll() is not None, True)
    load_stop(writer)
    load_stop(Holder())
    p4 = planned_counts("p4-open", 20)
    check("plan_p4_first", p4[("open.ql.txt.first", "list", "local")], 1)
    check("plan_p4_plain", p4[("open.ql.txt", "list", "local")], 20)
    check("plan_p6_long", planned_counts("p6-ops", 20)[("op.copy.long", "list", "local")], 3)

    def plan_rows(counts):
        rows = []
        for (journey, view, fixture), n in counts.items():
            row = row_of(journey, fixture.split("+")[0])
            for _ in range(n):
                r = {"journey": journey, "view": view, "fixture": fixture, "first_ms": 5.0}
                if "+load-" in fixture:  # a load row names its load apart from its fixture, as run_series writes it
                    r.update(fixture=fixture.split("+")[0], load=fixture[-1], load_active=True)
                if row == 2:
                    r["held"] = {"slots_missed": 0, "overshoot": 0, "steps_behind": 0}
                if row == 5:
                    r["settled_ms"] = 50.0
                rows.append(r)
        return rows

    full = plan_rows(planned_counts("p7-load", 20))

    def where(r):
        return r["fixture"] + ("+load-" + r["load"] if r.get("load") else "")
    code, gtext = run_gate(full, "p7-load", 20)
    check("gate_plan_pass_code", code, 0)
    check("gate_plan_pass_line", "GATE PASS" in gtext, True)
    code, gtext = run_gate(full, "p7-load", 20, only=["p7", "p7-load"])
    check("gate_plan_part_twice", (code, "GATE PASS" in gtext), (0, True))
    code, gtext = run_gate(full[:-1], "p7-load", 20)
    check("gate_plan_short_code", code, 2)
    check("gate_plan_short_line", "GATE MISSING" in gtext, True)
    kept = [r for r in full if (r["journey"], where(r)) != ("open.ql.txt.first", "local+load-a")]
    code, gtext = run_gate(kept, "p7-load", 20)
    check("gate_plan_first_gone_code", code, 2)
    check("gate_plan_first_gone_line",
          "GATE MISSING open.ql.txt.first dual local+load-a n=0 needs 1" in gtext, True)
    code, gtext = run_gate(full[:5], "p7-load", 20)
    check("gate_truncated_code", code, 2)
    check("gate_truncated_line", ("GATE MISSING" in gtext, "GATE PASS" in gtext), (True, False))
    try:
        with redirect_stderr(io.StringIO()):
            main(["gate", "census-p9-typ0.jsonl"])
        fail.append("typo_gate: a typo'd path was dropped instead of refused")
    except SystemExit as e:
        check("typo_gate_code", e.code, 2)
    miss2 = [dict(r) for r in full]
    n = 0
    for r in miss2:
        if r["journey"] == "open.ql.txt" and where(r) == "local+load-a" and n < 2:
            r["first_ms"] = None
            n += 1
    code, gtext = run_gate(miss2, "p7-load", 20)
    check("gate_miss_code", code, 1)
    check("gate_miss_line", "GATE BREACH open.ql.txt dual local+load-a p95=miss ceiling=100" in gtext, True)
    # Expected set from the plan: a part without a manifest is MISSING, no paths is a usage error.
    p1rows = plan_rows(planned_counts("p1-cursor", GATE_MIN_SAMPLES, PLAN_FX))
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(json.dumps({"manifest": {"date": "d", "host": "h", "flea_head": "c",
                                            "plan": {"part": "p1-cursor", "samples": GATE_MIN_SAMPLES},
                                            "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}) + "\n")
        for r in p1rows:
            tf.write(json.dumps(r) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        code = gate([tf.name])
    os.unlink(tf.name)
    check("gate_expected_code", code, 2)
    check("gate_expected_part", "GATE MISSING part p4-open" in buf.getvalue(), True)
    with redirect_stderr(io.StringIO()):
        check("gate_nopaths_code", gate([]), 2)
    check("resolve_short", resolve_part("p1"), "p1-cursor")
    check("resolve_full", resolve_part("p4-open"), "p4-open")
    check("resolve_unknown", resolve_part("p9"), None)
    # Defined counts bind: a manifest with samples=2 is MISSING against the defined 20.
    code, gtext = run_gate(full, "p7-load", 2)
    check("gate_samples_code", code, 2)
    check("gate_samples_line", "plan samples 2, under the defined 20" in gtext, True)
    # A census whose manifest has no flea_head never passes: it cannot name the tree it measured.
    code, gtext = run_gate(full, "p7-load", 20, head="")
    check("gate_empty_head_code", code, 2)
    check("gate_empty_head_line", "GATE MISSING p7-load manifest has an empty flea_head" in gtext, True)
    # A timed-out ipc call is an error row for its series, never an escape that kills the part.
    def hung(sess, sp, n):
        raise subprocess.TimeoutExpired("ipc", 10)

    class Stub:
        taken, view, load = [], "list", None
    SERIES["hung"] = hung
    try:
        trows = run_series(Stub(), {"type": "hung"}, 1)
    finally:
        del SERIES["hung"]
    check("series_timeout_row", [r.get("error", "")[:14] for r in trows], ["TimeoutExpired"])
    # A part census with two manifests never passes: a rerun must not pool with the old rows.
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        for head in ("c7e23485e", "d8f34596f"):
            tf.write(json.dumps({"manifest": {"date": "d", "host": "h", "flea_head": head,
                                "plan": {"part": "p1-cursor", "samples": GATE_MIN_SAMPLES},
                                "repeat": {"delay_ms": 250, "rate": 40, "source": "plan"}}}) + "\n")
        for r in p1rows:
            tf.write(json.dumps(r) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        code = gate([tf.name])
    os.unlink(tf.name)
    check("gate_two_manifests_code", code, 2)
    check("gate_two_manifests_line", "GATE MISSING part p1-cursor has 2 manifests" in buf.getvalue(), True)
    # clean outside the parent names both paths, exits non-zero, and still removes the stick root.
    parent = tempfile.mkdtemp(prefix="ig2a2-parent-")
    outer = tempfile.mkdtemp(prefix="ig2a2-outer-")
    open(os.path.join(parent, MARKER), "w").close()
    open(os.path.join(outer, MARKER), "w").close()
    usb = os.path.join(outer, "u", "v", "flea-instant-1")
    os.makedirs(os.path.join(usb, "dst"))
    open(os.path.join(usb, MARKER), "w").close()
    local = os.path.join(outer, "local-root")
    os.makedirs(local)
    open(os.path.join(local, MARKER), "w").close()
    with open(os.path.join(local, "fixtures.json"), "w") as f:
        json.dump({"usb": usb}, f)
    old_parent, ran = globals()["LOCAL_PARENT"], False
    globals()["LOCAL_PARENT"] = parent
    try:
        err = io.StringIO()
        with redirect_stderr(err):
            check("clean_outside_code", clean(local), 1)
        ran = True
        check("clean_outside_names", (os.path.realpath(local) in err.getvalue(),
                                      parent in err.getvalue()), (True, True))
        check("clean_outside_usb", os.path.exists(usb), False)
    finally:
        globals()["LOCAL_PARENT"] = old_parent
        for p in (parent, outer):
            if os.path.isabs(p) and p.strip("/") and marked(p):
                shutil.rmtree(p, ignore_errors=True)
    check("clean_outside_ran", ran, True)
    # A load-b session's own dst directory is removed; anything not named dst- is refused.
    with tempfile.TemporaryDirectory() as td:
        usb = os.path.join(td, "u", "flea-instant-9")
        dst = os.path.join(usb, "dst", "dst-abc")
        os.makedirs(dst)
        open(os.path.join(usb, MARKER), "w").close()
        open(os.path.join(dst, "zzz_big.bin"), "w").close()
        drop_load_dst(dst, [usb])
        check("load_dst_removed", (os.path.exists(dst), os.path.isdir(usb)), (False, True))
        try:
            drop_load_dst(os.path.join(usb, "dst"), [usb])
            fail.append("load_dst_guard: a non-dst directory was removed")
        except SystemExit:
            pass
    # Census calibration on the 23:17 numbers (literals from the cal1 input).
    m0 = [38.4, 27.3, 32.7, 23.5, 23.3, 24.5, 22.9, 27.6, 28.3, 27.9]
    c0 = [11.0, 3.0, 7.0, 3.0, 3.0, 5.0, 2.0, 2.0, 8.0, 2.0]
    m50 = [77.9, 73.4, 73.0, 74.8, 74.9, 72.7, 78.5, 77.0, 73.0, 76.7]
    c50 = [52.0, 52.0, 52.0, 52.0, 53.0, 52.0, 53.0, 52.0, 52.0, 52.0]
    m200 = [223.7, 228.3, 222.6, 227.3, 227.8, 222.5, 227.6, 223.1, 224.9, 215.5]
    c200 = [202.0, 203.0, 202.0, 202.0, 203.0, 202.0, 202.0, 202.0, 202.0, 203.0]
    row_fn = getattr(S, "calibrate_row", None)
    gate_fn = globals().get("calibrate_gate")
    check("calib_row_present", row_fn is not None, True)
    check("calib_gate_present", gate_fn is not None, True)
    if row_fn is not None and gate_fn is not None:
        r0 = row_fn(0, m0, c0, 27.4, 22.6, 8.33)
        r50 = row_fn(50, m50, c50, 27.4, 22.6, 8.33)
        r200 = row_fn(200, m200, c200, 27.4, 22.6, 8.33)
        check("calib_d0_pass", r0["within_one_frame"], True)
        check("calib_d50_pass", r50["within_one_frame"], True)
        check("calib_d200_fail", r200["within_one_frame"], False)
        rep = {"frame_ms": 8.33, "rows": [r0, r50, r200]}
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, rep)
        check("calib_gate_pass", code, 0)
        check("calib_outside_names", "outside_one_frame delay=200 runs=[10]" in err.getvalue(), True)
        # One 50 ms run 9 ms late against its own client swap, scored by the row rule itself.
        bad50 = row_fn(50, [52.0 + 22.6 + 9.0] + m50[1:], c50, 27.4, 22.6, 8.33)
        check("calib_50_off_row", bad50["within_one_frame"], False)
        with redirect_stderr(io.StringIO()):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [r0, bad50, r200]})
        check("calib_50_off_fails", code, 1)
        # An unscored row prints unscored, never outside_one_frame.
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [r0, r50, {"delay_ms": 200}]})
        check("calib_unscored_code", code, 0)
        check("calib_unscored_names", ("unscored delay=200" in err.getvalue(),
                                       "outside_one_frame delay=200" in err.getvalue()), (True, False))
        # A report with no frame_ms fails named.
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"rows": [r0, r50, r200]})
        check("calib_no_frame_code", code, 1)
        check("calib_no_frame_names", "no frame_ms" in err.getvalue(), True)
        # No 50 ms row, or an unscored one, fails named.
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [r0, r200]})
        check("calib_no_fifty_code", code, 1)
        check("calib_no_fifty_names", "no 50 ms row" in err.getvalue(), True)
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [r0, {"delay_ms": 50}, r200]})
        check("calib_fifty_unscored_code", code, 1)
        check("calib_fifty_unscored_names", "50 ms row unscored" in err.getvalue(), True)
        # A row whose median misses by more than a frame fails the gate, named.
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [r0, r50, {"delay_ms": 200,
                           "error_vs_client_ms": [12.0, 11.5, 12.4], "within_one_frame": False}]})
        check("calib_median_off_code", code, 1)
        check("calib_median_off_names", "median_outside_one_frame delay=200" in err.getvalue(), True)
        # Census calibration on the 02:37 numbers: delay 0 read a recorder round trip late, after the cursor hide.
        m0 = [31.8, 29.9, 29.2, 36.9, 31.5, 31.2, 31.5, 26.4, 36.7, 25.6]
        c0 = [2.0, 7.0, 6.0, 3.0, 7.0, 2.0, 11.0, 2.0, 2.0, 7.0]
        m50 = [54.6, 54.9, 54.6, 55.4, 55.6, 55.1, 55.1, 55.6, 54.9, 54.9]
        c50 = [52.0, 52.0, 52.0, 52.0, 53.0, 52.0, 52.0, 53.0, 52.0, 52.0]
        m200 = [205.7, 205.8, 205.7, 205.2, 205.2, 205.5, 205.6, 205.7, 204.7, 205.2]
        c200 = [203.0, 203.0, 202.0, 202.0, 202.0, 203.0, 203.0, 203.0, 202.0, 202.0]
        share0 = statistics.median(m - c for m, c in zip(m0, c0))
        q0, q50, q200 = (row_fn(d, m, c, 31.4, share0, 8.33)
                         for d, m, c in ((0, m0, c0), (50, m50, c50), (200, m200, c200)))
        check("calib_0237_shares", (q0.get("median_share_ms"), q50.get("median_share_ms"), q200.get("median_share_ms")),
              (24.45, 2.9, 2.75))
        check("calib_0237_offset", [round(statistics.median(q["error_vs_client_ms"]), 1) for q in (q50, q200)],
              [-21.6, -21.7])
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33, "rows": [q0, q50, q200]})
        check("calib_0237_code", code, 1)
        names = ("fifty_outside_one_frame delay=50 median=-21.6", "median_outside_one_frame delay=200 median=-21.7",
                 "outside_one_frame delay=0 runs=[4, 9] error_vs_client_ms=[9.4, 10.3]")
        check("calib_0237_names", [n in err.getvalue() for n in names], [True, True, True])
        # Run 1 at delay 0: the cursor hide at +1.0 ms took the pending frame, so the result came at +31.8 ms.
        spent_fn = getattr(S, "frames_before_result", None)
        check("calib_spent_present", spent_fn is not None, True)
        if spent_fn is not None:
            key, first = 1000 * MS, (-600 * MS, -1, -1)
            hid = [(key + t, rows, ink) for t, rows, ink in (first, (1 * MS, 0, -1), (31.8 * MS, 652, 300))]
            bare = [(key + t, rows, ink) for t, rows, ink in (first, (4.6 * MS, 652, 300))]
            check("calib_spent_counts", (spent_fn(hid, key, hid[-1][0]), spent_fn(bare, key, bare[-1][0])), (1, 0))
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33,
                                             "rows": [dict(q0, frames_before_result=[1] * 10), r50, r200]})
        check("calib_spent_code", code, 1)
        check("calib_spent_names", "capture_spent delay=0 runs=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]" in err.getvalue(), True)
        # The same rows with no frame before any delay-0 result pass, so the spent branch alone fails the gate.
        err = io.StringIO()
        with redirect_stderr(err):
            code = gate_fn({"clean": True}, {"frame_ms": 8.33,
                                             "rows": [dict(q0, frames_before_result=[0] * 10), r50, r200]})
        check("calib_unspent", (code, "capture_spent" in err.getvalue()), (0, False))
    # Targets and filters (CB01): the dual pointer, the Enter row, the external window guard, the journey filter.
    win = (0, 0, 2560, 1440)
    region, left = win_rect(win, "content"), win_rect(win, "left")
    dual_x, dual_y = scroll_point("dual", region, left)
    check("scroll_dual_in_left", (left[0] <= dual_x <= left[0] + left[2] - 8, left[1] <= dual_y < left[1] + left[3]),
          (True, True))
    check("scroll_dual_off_the_gap", dual_x < region[0] + region[2] // 2 - 100, True)
    for view in ("list", "columns", "grid"):
        check("scroll_%s_centre" % view, scroll_point(view, region, left),
              (region[0] + region[2] // 2, region[1] + region[3] // 2))
    # The p7-load Enter lands on a_dir (index 0): the fixtures.json root holds L as the builder lays it out.
    with tempfile.TemporaryDirectory() as td:
        L = os.path.join(td, "load", "L")
        os.makedirs(os.path.join(L, "a_dir"))
        os.makedirs(os.path.join(L, ".hidden_dir"))
        for name in ("b_text.txt", "f_0000.txt", "f_0001.txt", "zzz_big.bin"):
            open(os.path.join(L, name), "w").close()
        os.makedirs(os.path.join(td, "nav", "enter", "big"))
        p7 = parts({"root": td, "nav": {"nav500": 500}}, 20)["p7-load"]
        enters = [sp for s in p7 for sp in s["series"] if sp["type"] == "cycle" and sp["steps"][0]["journey"] == "nav.dual.enter"]
        check("p7_enter_specs", (len(enters), all(sp.get("to_folder") for sp in enters), any(sp.get("setup") for sp in enters)),
              (3, True, False))
        check("ui_patch_pins_folders_first", ui_patch("dual", L).get("foldersFirst"), True)
        check("p7_enter_index", folder_setup(ui_patch("dual", p7[0]["fixture"]), p7[0]["fixture"]), [k("home")])
        check("p3_enter_index", folder_setup(ui_patch("list", os.path.join(td, "nav", "enter")),
                                             os.path.join(td, "nav", "enter")), [k("home")])
        # The first folder is the fourth name: folders first still puts it at row 0, so the keys are Home alone.
        shutil.rmtree(os.path.join(L, "a_dir"))
        os.makedirs(os.path.join(L, "g_dir"))
        check("folder_setup_fourth_by_name", folder_setup(ui_patch("dual", L), L), [k("home")])
        targets, bad = dry_targets(parts({"root": td, "nav": {"nav500": 500}}, 20)["p7-load"])
        check("dry_enter_index", (bad, sorted({(t["enter_index"], t["aim"]) for t in targets if t["series"] == "cycle"})),
              (0, [(0, "the first folder, row 0 with folders first")]))
        try:
            folder_setup(dict(ui_patch("dual", L), foldersFirst=False), L)
            fail.append("folder_setup_folders_last: a session with foldersFirst false was accepted")
        except RuntimeError as e:
            check("folder_setup_folders_last_names_it", "assumes folders first" in str(e), True)
        try:
            folder_setup({key: v for key, v in ui_patch("dual", L).items() if key != "foldersFirst"}, L)
            fail.append("folder_setup_unpinned: a patch that does not set foldersFirst was accepted")
        except RuntimeError:
            pass
        os.rmdir(os.path.join(L, "g_dir"))
        try:
            folder_setup(ui_patch("dual", L), L)
            fail.append("folder_setup_none: a listing with no folder was accepted")
        except RuntimeError:
            pass
    # --only: the dual wheel and fling are all that is left of p1-cursor, and the gate plans only them.
    only = r"nav\.dual\.(wheel|fling)"
    check("only_p1_journeys", journeys_of(only_series(built["p1-cursor"], only)), {"nav.dual.wheel", "nav.dual.fling"})
    check("only_p1_sessions", [(s["view"], [sp["type"] for sp in s["series"]]) for s in only_series(built["p1-cursor"], only)],
          [("dual", ["wheel", "fling"])])
    check("only_p1_plan", planned_counts("p1-cursor", 20, fx, only),
          {("nav.dual.wheel", "dual", "local"): 40, ("nav.dual.fling", "dual", "local"): 20})
    only7 = r"nav\.dual\.(enter|back)"
    check("only_p7_journeys", journeys_of(only_series(built["p7-load"], only7)), {"nav.dual.enter", "nav.dual.back"})
    check("only_p7_sessions", len(only_series(built["p7-load"], only7)), 3)
    check("only_p7_plan", sorted(planned_counts("p7-load", 20, fx, only7)),
          sorted((j, "dual", "local+load-" + x) for j in ("nav.dual.enter", "nav.dual.back") for x in "abc"))
    check("only_none_keeps_all", planned_counts("p1-cursor", 20, fx), planned_counts("p1-cursor", 20, fx, None))
    check("only_no_match", only_series(built["p1-cursor"], r"^no such journey$"), [])
    try:
        with redirect_stderr(io.StringIO()):
            journey_filter("(")
        fail.append("only_bad_regex: an unbalanced regex was accepted")
    except SystemExit as e:
        check("only_bad_regex_code", e.code, 2)
    filtered = plan_rows(planned_counts("p1-cursor", GATE_MIN_SAMPLES, PLAN_FX, only))
    code, gtext = run_gate(filtered, "p1-cursor", 20, pattern=only)
    check("gate_only_pass", (code, "GATE PASS journeys=2" in gtext), (0, True))
    code, gtext = run_gate(filtered, "p1-cursor", 20)
    check("gate_unfiltered_missing", (code, "GATE MISSING nav.list.down list local n=0" in gtext), (2, True))
    slow = [dict(r, first_ms=30.0) if r["journey"] == "nav.dual.wheel" else r for r in filtered]
    code, gtext = run_gate(slow, "p1-cursor", 20, pattern=only)
    check("gate_only_breach", (code, "GATE BREACH nav.dual.wheel dual local p95=30.0 ceiling=16.7" in gtext), (1, True))
    short = [r for r in filtered if r["journey"] != "nav.dual.fling"] + [r for r in filtered if r["journey"] == "nav.dual.fling"][:-1]
    code, gtext = run_gate(short, "p1-cursor", 20, pattern=only)
    check("gate_only_short", (code, "GATE MISSING nav.dual.fling dual local n=19 needs 20" in gtext), (2, True))
    erred = filtered + [{"journey": "nav.dual.enter", "view": "dual", "fixture": "local", "error": "an external window opened"}]
    code, gtext = run_gate(erred, "p1-cursor", 20, pattern=only)
    check("gate_only_error_row", (code, "error an external window opened" in gtext), (2, True))
    # The regex after --only is not taken for a census path.
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl") as tf:
        err = io.StringIO()
        try:
            with redirect_stderr(err), redirect_stdout(io.StringIO()):
                main(["gate", tf.name, "--only", r"nav\.dual\.wheel"])
        except SystemExit:
            pass
        check("gate_only_not_a_path", "need an existing" in err.getvalue(), False)
    # The wheel and the fling aim with scroll_point: stop each series right after its pointer warp.
    import types
    aimed, real = [], {n: globals()[n] for n in ("pointer_to", "paced", "touchpad")}

    class Stop(Exception):
        pass

    def stop(*a):
        raise Stop
    globals().update(pointer_to=lambda keys, x, y: aimed.append((x, y)), paced=stop,
                     touchpad=lambda: (types.SimpleNamespace(create=stop), None))
    try:
        for view in ("dual", "list"):
            aimed.clear()
            ses = object.__new__(Session)
            ses.view, ses.region, ses.keys, ses.rect, ses.settle = view, region, None, lambda kind: left, lambda *a: None
            ses.taken = []
            for series in (series_wheel, series_fling):
                try:
                    series(ses, {}, 1)
                except Stop:
                    pass
            check("series_aim_%s" % view, aimed, [scroll_point(view, region, left)] * 2)
    finally:
        globals().update(real)
    # An outsider is any mapped client but Flea's own: class S.FLEA_CLASS under a pid below the launched process.
    mine = {"address": "0x1", "class": S.FLEA_CLASS, "pid": 11, "mapped": True}
    editor = {"address": "0x2", "class": "org.gnome.TextEditor", "pid": 99, "mapped": True}
    editor_child = dict(editor, address="0x3", pid=12)
    other_flea = dict(mine, address="0x4", pid=77)
    hidden = dict(editor, address="0x5", mapped=False)
    check("external_clients", [c["address"] for c in external_clients([mine, editor, editor_child, other_flea, hidden], {10, 11, 12})],
          ["0x2", "0x3", "0x4"])
    # A timed key taps Shift before the pre-key settle, so the cursor hide never commits in place of the key's frame.
    sent = []
    sess = object.__new__(Session)
    sess.keys, sess.taken = types.SimpleNamespace(chord=lambda *k: sent.append(k) or 7), []
    sess.rec = types.SimpleNamespace(wait_quiet=lambda *a: sent.append("quiet") or True)
    sess.columns = lambda *a: {"journey": a[0], "first_ms": 245.5}
    sess.known, now = {"0x9"}, {"0x9": editor}
    sess.outsiders = lambda: dict(now)
    check("sample_hides_first", (sess.sample("j", ("key", ("l",)), 1), sent),
          ({"journey": "j", "first_ms": 245.5}, [("shift",), "quiet", ("l",), "quiet"]))
    # The editor an Enter on a text file opens maps during the sample: an error row, not a number or a miss.
    now["0x2"] = editor
    try:
        sess.sample("nav.dual.enter", ("key", ("enter",)), 3)
        fail.append("sample_external: an external window did not end the sample")
    except ExternalWindow as e:
        check("sample_external_row", (e.row["error"], e.row["journey"], e.row["external"]),
              ("an external window opened", "nav.dual.enter", [{"class": "org.gnome.TextEditor", "title": None, "pid": 99}]))
        check("external_window_escapes_runtime_handlers", isinstance(e, RuntimeError), False)
    sess.known.add("0x2")
    check("sample_known_window_is_quiet", sess.sample("j", ("key", ("l",)), 1)["journey"], "j")
    # An outside window on the 20th sample of a cycle keeps the 19 timings before it and adds one error row.
    taken, sess.ctx, sess.known = [], {}, {"0x9"}
    sess.columns = lambda *a: taken.append(a[0]) or {"journey": a[0], "first_ms": 1.0}
    sess.outsiders = lambda: {"0x9": editor, **({"0x7": editor} if len(taken) >= 20 else {})}
    rows = run_series(sess, {"type": "cycle", "steps": [{"journey": "nav.dual.enter", "in": ("key", ("enter",))}]}, 20)
    check("series_keeps_samples_before_external", (len(rows), sum("error" not in r for r in rows), rows[-1].get("error")),
          (20, 19, "an external window opened"))
    # An ordinary error on the 4th sample keeps the 3 timings before it and adds one error row.
    seen = []

    def column_or_die(*a):
        if len(seen) == 3:
            raise OSError("uinput gone")
        seen.append(a[0])
        return {"journey": a[0], "first_ms": 1.0}
    sess.view, sess.columns, sess.outsiders = "dual", column_or_die, lambda: {"0x9": editor}
    rows = run_series(sess, {"type": "cycle", "steps": [{"journey": "nav.dual.enter", "in": ("key", ("enter",))}]}, 20)
    check("series_keeps_samples_before_oserror", (len(rows), sum("error" not in r for r in rows), rows[-1].get("error")),
          (4, 3, "OSError: uinput gone"))
    # A qs ipc call that outlives its timeout is an error row after the samples before it, in a series and in a session start.
    ipc_timeout = subprocess.TimeoutExpired(["qs", "ipc"], 10)
    seen.clear()

    def column_or_time_out(*a):
        if len(seen) == 3:
            raise ipc_timeout
        seen.append(a[0])
        return {"journey": a[0], "first_ms": 1.0}
    sess.columns = column_or_time_out
    rows = run_series(sess, {"type": "cycle", "steps": [{"journey": "nav.dual.enter", "in": ("key", ("enter",))}]}, 20)
    check("series_keeps_samples_before_timeout", (len(rows), sum("error" not in r for r in rows), rows[-1].get("error", "")[:16]),
          (4, 3, "TimeoutExpired: "))
    # A window gone after it mapped is a named RuntimeError, which run_session records as an error row, never a StopIteration or IndexError.
    from unittest import mock
    with mock.patch.object(S, "clients", return_value=[{"address": "0x1", "size": [8, 8]}]):
        check("window_at_found", window_at("0x1")["size"], [8, 8])
        try:
            window_at("0x2")
            fail.append("window_at_gone: no refusal")
        except RuntimeError as e:
            check("window_at_gone_names", str(e), "the window 0x2 closed before its geometry was read")
    # A wheel series builds 2 rows per sample, none through sample(): an error on its 5th of 6 inputs keeps the 4 rows built.
    sent_wheel = []

    def wheel_or_die(keys, notches):
        if len(sent_wheel) == 4:
            raise OSError("uinput gone")
        sent_wheel.append(notches)
        return time.monotonic_ns()
    sess.region, sess.rect = (0, 0, 200, 100), lambda kind: (0, 0, 100, 100)
    sess.rec.snapshot = lambda: []
    with mock.patch.dict(globals(), pointer_to=lambda *a: None, wheel=wheel_or_die, PACE_MS=1), \
            mock.patch.object(S, "search_columns", lambda frames, times, *a: {"per_key_first_ms": [2.0] * len(times)}), \
            mock.patch.object(S, "changes_of", lambda frames: []), \
            mock.patch.object(S, "first_change_after", lambda changes, t: None), \
            mock.patch.object(S, "park_cursor", lambda: None):
        rows = run_series(sess, {"type": "wheel"}, 3)
    check("series_keeps_wheel_rows_before_error",
          (len(rows), {(r.get("journey"), r.get("first_ms")) for r in rows[:-1]}, rows[-1].get("error")),
          (5, {("nav.dual.wheel", 2.0)}, "OSError: uinput gone"))

    def session_times_out(*a, **kw):
        raise ipc_timeout
    session_rows = io.StringIO()
    with mock.patch.dict(globals(), Session=session_times_out):
        run_session({"fixture": "/none", "view": "dual", "region": "r", "series": [], "n": 1}, {"out_dir": "/none", "tag": "t"}, session_rows)
    check("session_start_timeout_is_an_error_row", json.loads(session_rows.getvalue())["error"][:16], "TimeoutExpired: ")
    # An op series keeps its own error rows too: an outside window on the 2nd sample follows the 1st sample's error row.
    calls = []

    def sample_then_die(journey, *a):
        calls.append(journey)
        if len(calls) == 1:
            raise RuntimeError("no centre")
        raise ExternalWindow({"journey": journey, "error": "an external window opened"})
    idle_bar = {"activities": [], "errors": 0, "notice": ""}
    sess.ipc = lambda name, *a: json.dumps(idle_bar)
    sess.ctx, sess.sample = {"roots": [], "op_base": "/none", "root": "/none", "tag": "t"}, sample_then_die
    with mock.patch.dict(globals(), empty_dir=lambda *a: None, reflink=lambda *a: None):
        rows = run_series(sess, {"type": "op", "op": "op.trash"}, 3)
    check("series_op_keeps_error_rows_before_external", [r["error"] for r in rows],
          ["RuntimeError: no centre", "an external window opened"])
    # Op isolation (CB02): a scripted statusActivityState drives the Esc count, the row fields and the error row.
    def bar(running=None, errors=0, notice=""):
        acts = [] if running is None else [{"id": 1, "text": "Copying", "running": True, "cancelling": running == "cancelling",
                                            "ownerPath": "/x", "ownerFocused": True}]
        return {"activities": acts, "errors": errors, "notice": notice}

    def drive(states, cap_s=60):
        seq, pressed, clock = list(states), [], [0.0]

        def sleep(secs):
            clock[0] += secs
        got = isolate(lambda: seq.pop(0) if len(seq) > 1 else seq[0], lambda: pressed.append(clock[0]),
                      cap_s, now=lambda: clock[0], sleep=sleep)
        return got[0], len(pressed), got[2], round(clock[0], 1)
    # Running, then cancelling, then two queued errors, then clear: one cancel Esc and one Esc for each error.
    check("isolate_running_cancelled_errors", drive([bar("running"), bar("cancelling"), bar(errors=2), bar(errors=1), bar()]),
          (3, 3, None, 1.2))
    check("isolate_idle_presses_nothing", drive([bar()]), (0, 0, None, 0.0))
    # An error ahead of a running transfer takes the first Esc (it dismisses the error), the transfer the second.
    check("isolate_error_then_transfer", drive([bar("running", errors=1), bar("running"), bar("cancelling"), bar()])[:3],
          (2, 2, None))
    # A cancel that left the transfer running is pressed again only after the gap, never every poll.
    check("isolate_cancel_gap", drive([bar("running")] * 18 + [bar()])[:3], (2, 2, None))
    check("isolate_cancel_once", drive([bar("running")] * 3 + [bar("cancelling")] * 3 + [bar()])[:3], (1, 1, None))
    # A failed IPC read is waited out like a busy bar.
    check("isolate_none_waited_out", drive([None, None, bar()])[:3], (0, 0, None))
    check("isolate_none_names_it", drive([None], 1)[2], "the status bar is not idle: statusActivityState gave no reading")
    stuck = drive([bar("cancelling")], 60)
    check("isolate_stuck_transfer", (stuck[0], stuck[1], stuck[2], stuck[3] >= 60),
          (0, 0, "the status bar is not idle: 1 activities after 0 Esc, the first 'Copying'", True))
    swallowed = drive([bar(errors=2)], 20)
    check("isolate_swallowed_esc", (swallowed[0] >= 3, swallowed[0] == swallowed[1],
                                    swallowed[2].startswith("the status bar is not idle: 2 queued errors after ")),
          (True, True, True))
    # The row fields: a paste that left a queued error is refused, one that left none is not, and no reading is unknown.
    shown = op_state_fields(bar(), bar("running", errors=1, notice="Copied 0 items"))
    check("op_fields_refused", (shown["errors_before"], shown["errors_after"], shown["refused"], shown["notice_after"],
                                shown["activity_after"]),
          (0, 1, True, "Copied 0 items", [{"text": "Copying", "running": True, "cancelling": False}]))
    check("op_fields_clean", (op_state_fields(bar(), bar())["refused"], op_state_fields(bar(), bar())["activity_after"]),
          (False, []))
    check("op_fields_unread", op_state_fields(bar(), None),
          {"errors_before": 0, "errors_after": None, "activity_after": None, "notice_before": "", "notice_after": None,
           "refused": None})
    # Whole samples of an op: statusActivityState follows the script in call order, the rest is read off a pane that follows the keys.
    with tempfile.TemporaryDirectory() as td:
        os.makedirs(os.path.join(td, "ops"))
        with open(os.path.join(td, "ops", "large.bin"), "wb") as f:
            f.write(b"x" * 8)

        class FakePane:
            """The pane of the sample's folders: a symlink to a folder is listed among the files, so after b-src."""

            def __init__(self, base, cwd, link_enters=True, stale=False, late=0, ghost=None, resolves=False):
                self.base, self.cwd, self.cursor, self.sel, self.clip = base, cwd, 0, [], {"paths": [], "moving": False}
                self.link_enters, self.stale, self.pastes = link_enters, stale, []
                self.busy, self.late, self.ghost, self.resolves = 0, late, ghost, resolves
                if ghost:
                    self.clip = {"paths": [ghost], "moving": False}

            def rows(self):
                names = sorted(os.listdir(self.cwd))
                folders = [n for n in names if os.path.isdir(os.path.join(self.cwd, n)) and not os.path.islink(os.path.join(self.cwd, n))]
                return folders + [n for n in names if n not in folders]

            def key(self, chord):
                rows = self.rows()
                if self.busy:
                    return
                if chord == ("home",):
                    self.cursor = 0
                elif chord == ("end",):
                    self.cursor = max(len(rows) - 1, 0)
                elif chord == ("down",):
                    self.cursor = min(self.cursor + 1, max(len(rows) - 1, 0))
                elif chord == ("backspace",):
                    self.cwd, self.cursor = os.path.dirname(os.path.realpath(self.cwd) if self.resolves else self.cwd), 0
                elif chord == ("enter",) and rows:
                    row = os.path.join(self.cwd, rows[self.cursor])
                    if os.path.isdir(row) and (self.link_enters or not os.path.islink(row)):
                        self.cwd, self.cursor = row, 0
                elif chord == ("ctrl", "a"):
                    self.sel = [] if self.stale else [os.path.join(self.cwd, n) for n in rows]
                elif chord in (("ctrl", "c"), ("ctrl", "x")) and not self.ghost:
                    self.clip = {"paths": list(self.sel), "moving": chord == ("ctrl", "x")}
                elif chord == ("ctrl", "v"):
                    self.pastes.append((os.path.realpath(self.cwd), len(self.clip["paths"])))

            def ipc(self, name, *a):
                rows = self.rows()
                if name == "listInFlight" and self.busy:
                    self.busy -= 1
                    return "true"
                return {"path": lambda: self.cwd, "listInFlight": lambda: "false", "total": lambda: str(len(rows)),
                        "cursor": lambda: str(self.cursor), "fileClipboard": lambda: json.dumps(self.clip),
                        "visibleRowName": lambda: rows[int(a[0])] if int(a[0]) < len(rows) else ""}[name]()

        def op_session(script, **pane_kw):
            """A session whose statusActivityState follows script and whose pane follows the keys; a-dst is a symlink."""
            seq, sent = list(script), []
            base = os.path.join(td, "base")
            usb = os.path.join(td, "usb", "dst")
            for d in (os.path.join(base, "b-src"), usb):
                os.makedirs(d, exist_ok=True)
            if not os.path.lexists(os.path.join(base, "a-dst")):
                os.symlink(usb, os.path.join(base, "a-dst"))
            pane = FakePane(base, os.path.join(base, "b-src"), **pane_kw)
            ses = object.__new__(Session)
            ses.view, ses.taken, ses.known = "list", [], set()
            ses.ctx = {"roots": [], "op_base": base, "root": td, "tag": "t"}
            ses.settle = lambda *a: None
            ses.rec = types.SimpleNamespace(wait_quiet=lambda *a: True, armed_ns=None)

            def chord(*c):
                sent.append(c)
                pane.key(c)
                return len(sent)
            ses.keys = types.SimpleNamespace(chord=chord)
            ses.ipc = lambda name, *a: (json.dumps(seq.pop(0) if len(seq) > 1 else seq[0]) if name == "statusActivityState"
                                      else pane.ipc(name, *a))
            ses.columns = lambda name, t, row, evidence=None, extra=None: dict({"journey": name, "first_ms": 7.0},
                                                                               **(extra or {}))
            ses.pane = pane
            return ses, sent

        def fill(dst, count=1):
            """The reflink of a sample: count files into the source folder."""
            for j in range(count):
                with open(os.path.join(dst, "f%d" % j), "w") as f:
                    f.write("x")

        def wipe(path, roots):
            for old in os.listdir(path):
                os.unlink(os.path.join(path, old))

        def run_op(script, finish, caps=None, op="op.copy.large", n=1, **pane_kw):
            ses, sent = op_session(script, **pane_kw)
            count = OPS[op]["count"]
            patches = dict(empty_dir=lambda *a: (sent.append(("empty_dir",)), wipe(*a)),
                           reflink=lambda s, d: (fill(d, count), setattr(ses.pane, "busy", ses.pane.late)),
                           wait_until=lambda pred, cap: finish)
            patches.update(caps or {})
            with mock.patch.dict(globals(), patches), mock.patch.object(time, "sleep", lambda secs: None):
                rows = run_series(ses, {"type": "op", "op": op}, n)
            run_op.last = ses
            return rows, sent
        # A paste on an idle bar that finishes: timed, no Esc, nothing refused.
        rows, sent = run_op([bar()], 0.5)
        check("series_op_idle_paste", (len(rows), rows[0]["first_ms"], rows[0]["errors_before"], rows[0]["errors_after"],
                                       rows[0]["refused"], rows[0]["esc_before"], ("esc",) in sent, "cancel_escs" in rows[0]),
              (1, 7.0, 0, 0, False, 0, False, False))
        # Two errors queued from an earlier sample are pressed away before the folders are emptied or the paste is sent.
        rows, sent = run_op([bar(errors=2), bar(errors=1), bar()], 0.5)
        escs_at = [i for i, c in enumerate(sent) if c == ("esc",)]
        check("series_op_dismisses_queued_errors", (rows[0]["esc_before"], rows[0]["errors_before"], len(escs_at),
                                                     escs_at[-1] < sent.index(("empty_dir",)), escs_at[-1] < sent.index(("ctrl", "v"))),
              (2, 0, 2, True, True))
        # A paste refused and never finished: timed and marked refused, then cancelled; Esc dismisses the error first.
        script = [bar(), bar(), bar("running", errors=1, notice="an operation is already running"),
                  bar("running", errors=1), bar("running"), bar("cancelling"), bar()]
        rows, sent = run_op(script, None)
        after_paste = sent[sent.index(("ctrl", "v")) + 1:]
        check("series_op_refused_and_cancelled", (rows[0]["refused"], rows[0]["errors_after"], rows[0]["op_s"], rows[0]["cancel_escs"],
                                                  [c for c in after_paste if c == ("esc",)], "cancel_failed" in rows[0],
                                                  rows[0]["activity_after"][0]["running"]),
              (True, 1, None, 2, [("esc",), ("esc",)], False, True))
        # A bar that does not clear before the sample is an error row and the folders are never touched.
        touched = []
        rows, sent = run_op([bar("cancelling")], 0.5, {"ISOLATE_PASTE_CAP_S": 0,
                                                        "empty_dir": lambda *a: touched.append(a)})
        check("series_op_unclear_bar_is_an_error_row", (len(rows), rows[0].get("first_ms"), touched, ("ctrl", "v") in sent,
                                                        rows[0]["error"]),
              (1, None, [], False, "RuntimeError: the status bar is not idle: 1 activities after 0 Esc, the first 'Copying'"))
        # A cancel that does not take is named on the row.
        rows, sent = run_op([bar(), bar(), bar("running")], None, {"ISOLATE_CANCEL_CAP_S": 0})
        check("series_op_cancel_that_stays", ("cancel_failed" in rows[0], rows[0]["cancel_escs"]), (True, 1))
        # Every sample of a cross-fs move starts in the source and pastes in the destination, though a-dst is listed after b-src.
        real_dst, real_src = os.path.realpath(os.path.join(td, "base", "a-dst")), os.path.realpath(os.path.join(td, "base", "b-src"))
        with mock.patch.dict(OPS["op.move.cross-fs"], count=3):
            rows, sent = run_op([bar()], 0.5, op="op.move.cross-fs", n=3)
        check("series_op_move_pastes_in_the_destination_every_sample",
              (len(rows), [r.get("refused") for r in rows], [r.get("error") for r in rows], run_op.last.pane.pastes),
              (3, [False] * 3, [None] * 3, [(real_dst, 3)] * 3))
        # The old closing keys, replayed on the same pane, put the second sample in the destination with nothing to take.
        legacy = FakePane(os.path.join(td, "base"), real_src)
        for c in (("ctrl", "a"), ("ctrl", "x"), ("backspace",), ("home",), ("enter",), ("ctrl", "v")):
            legacy.key(c)
        check("legacy_blind_keys_paste_into_the_source", legacy.pastes, [(real_src, 3)])
        for c in (("backspace",), ("end",), ("enter",), ("ctrl", "a"), ("ctrl", "x")):
            legacy.key(c)
        check("legacy_blind_keys_take_nothing_from_the_destination", (legacy.cwd, legacy.clip), (os.path.join(td, "base", "a-dst"),
                                                                                              {"paths": [], "moving": True}))
        # A pane that starts a sample one level up or inside the destination is walked back before anything is taken.
        ses, sent = op_session([bar()])
        ses.pane.cwd = os.path.join(td, "base", "a-dst")
        patches = dict(empty_dir=wipe, reflink=lambda s_, d: fill(d, 1), wait_until=lambda pred, cap: 0.5)
        with mock.patch.dict(globals(), patches), mock.patch.object(time, "sleep", lambda secs: None):
            rows = run_series(ses, {"type": "op", "op": "op.copy.large"}, 1)
        check("series_op_walks_back_from_the_destination", (rows[0].get("error"), ses.pane.pastes), (None, [(real_dst, 1)]))
        # A destination the pane cannot enter, or a take of nothing, is a harness fault row: no paste, no refusal, no timing.
        rows, sent = run_op([bar()], 0.5, link_enters=False)
        check("series_op_unreachable_destination_is_a_harness_fault",
              (len(rows), rows[0].get("harness_fault"), "first_ms" in rows[0], "refused" in rows[0], ("ctrl", "v") in sent,
               rows[0].get("error", "").startswith("HarnessFault: ")),
              (1, True, False, False, False, True))
        rows, sent = run_op([bar()], 0.5, stale=True)
        check("series_op_empty_take_is_a_harness_fault",
              (rows[0].get("harness_fault"), "refused" in rows[0], ("ctrl", "v") in sent,
               rows[0].get("error", "").startswith("HarnessFault: the clipboard holds 0 paths (moving False), not the 1")),
              (True, False, False, True))
        # The series stops after HARNESS_FAULTS_STOP consecutive faults.
        rows, sent = run_op([bar()], 0.5, stale=True, n=5, caps={"TAKE_CAP_S": 0})
        check("series_op_stops_after_repeated_harness_faults", (len(rows), all(r.get("harness_fault") for r in rows)),
              (HARNESS_FAULTS_STOP, True))
        # The take keys wait for a listing still landing after the fill: sent blind, Ctrl+A is swallowed and the take is lost.
        rows, sent = run_op([bar()], 0.5, late=2)
        check("series_op_take_waits_for_a_late_listing", (rows[0].get("error"), run_op.last.pane.pastes), (None, [(real_dst, 1)]))
        # A clipboard of the right size that holds other paths is a harness fault, never a timed paste.
        rows, sent = run_op([bar()], 0.5, ghost=os.path.join(td, "elsewhere", "f0"))
        check("series_op_foreign_clipboard_is_a_harness_fault",
              (rows[0].get("harness_fault"), "first_ms" in rows[0], ("ctrl", "v") in sent,
               rows[0].get("error", "").startswith("HarnessFault: the clipboard holds 1 paths (moving False), not the 1")),
              (True, False, False, True))
        # A pane that climbs out of a-dst by its resolved path lands outside base: a named fault, not a silent skew of every sample.
        rows, sent = run_op([bar()], 0.5, n=2, resolves=True)
        check("series_op_resolved_climb_is_a_named_fault",
              ([r.get("harness_fault") for r in rows], rows[1].get("error", "").startswith("HarnessFault: the pane is at "),
               "first_ms" in rows[1]), ([None, True], True, False))
        # A listing that never lands is a harness fault, not a key sent into a loading pane.
        ses, sent = op_session([bar()])
        ses.ipc = lambda name, *a: "true" if name == "listInFlight" else ""
        try:
            listing_landed(ses, 0)
            landed = "no fault"
        except HarnessFault as e:
            landed = str(e)
        check("listing_that_never_lands_is_a_harness_fault", (landed, sent), ("the listing did not land in 0 s", []))
        # op_s counts from the paste chord, not from the wait that follows the 11 s evidence frame of a long copy.
        ses, sent = op_session([bar()])
        real_chord = ses.keys.chord
        ses.keys = types.SimpleNamespace(chord=lambda *c: (real_chord(*c), time.monotonic_ns() - 5 * NS if c == ("ctrl", "v") else 0)[1])  # the paste began 5 s ago
        with mock.patch.dict(globals(), dict(empty_dir=wipe, reflink=lambda s_, d: fill(d, 1), wait_until=lambda pred, cap: 0.5)), mock.patch.object(time, "sleep", lambda secs: None):
            rows = run_series(ses, {"type": "op", "op": "op.copy.large"}, 1)
        check("op_s_from_paste", 5.0 <= rows[0]["op_s"] < 60.0, True)
    # op.size.recursive has an instrument: its session shows Properties and hides nothing else differently from the shipped default.
    with open(os.path.join(REPO, "src", "uischema.rs")) as f:
        # Sample input: '  "menu": { "hidden": ["delete", "openTerminal", "placeMenu", "runScript",' then more names to '] }'.
        shipped = re.search(r'"menu": \{ "hidden": \[([^\]]*)\]', f.read()).group(1)
    shipped = re.findall(r'"([^"]+)"', shipped)
    check("menu_hidden_is_the_shipped_default", MENU_HIDDEN, shipped)
    check("size_ui_shows_only_properties", ([a for a in MENU_HIDDEN if a not in SIZE_UI["menu"]["hidden"]],
                                            len(SIZE_UI["menu"]["hidden"])), (["properties"], len(shipped) - 1))
    with open(os.path.join(REPO, "ui", "js", "Nav.js")) as f:
        # Sample input: 'var here = pane.path', 'var cut = here.lastIndexOf("/")', 'pane.open(cut <= 0 ? "/" : here.substring(0, cut))'.
        nav = f.read()
    check("flea_backspace_climbs_the_pane_path_lexically",
          'var here = pane.path\n    var cut = here.lastIndexOf("/")\n    pane.pendingSelect = here\n'
          '    pane.open(cut <= 0 ? "/" : here.substring(0, cut))' in nav, True)
    ops_plan = {sp["op"]: sp.get("ui") for sp in parts(PLAN_FX, 20)["p6-ops"]}
    check("size_session_carries_the_ui_patch", (ops_plan["op.size.recursive"], ops_plan["op.copy.many"], ops_plan["op.archive"]),
          (SIZE_UI, None, None))
    check("size_patch_reaches_the_session", dict(ui_patch("list", "/x"), **SIZE_UI)["menu"], SIZE_UI["menu"])
    # The report names a harness fault as one, beside the other error rows, and counts it in no verdict.
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        tf.write(json.dumps({"journey": "nav.list.down", "row": 1, "view": "list", "first_ms": 5.0, "fixture": "local"}) + "\n")
        tf.write(json.dumps({"journey": "op.move.cross-fs", "row": 6, "view": "list", "fixture": "local",
                             "error": "HarnessFault: the clipboard holds 0 paths", "harness_fault": True}) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        report([tf.name])
    os.unlink(tf.name)
    check("rep_harness_fault_named", ("- harness fault op.move.cross-fs list local: HarnessFault: the clipboard holds 0 paths"
                                      in buf.getvalue(), "Verdicts: TARGET 1\n" in buf.getvalue()), (True, True))
    # The calibration probes keep their work directory only while they run.
    works = []

    def fake_window(work, delay):
        works.append(work)
        logp = os.path.join(work, "client.log")
        with open(logp, "w") as f:
            f.write("CALIB key a\n")
        return None, None, [], logp

    with mock.patch.dict(globals(), dict(calib_window=fake_window, end_client=lambda client: None)), \
            mock.patch.object(S, "build_recorder", lambda path: path), \
            mock.patch.object(S, "Recorder", lambda *a: types.SimpleNamespace(snapshot=lambda: [], stop=lambda: None)), \
            mock.patch.object(time, "sleep", lambda secs: None):
        repeat_probe(types.SimpleNamespace(_key=lambda *a: None), {"rate": 40, "delay_ms": 250}, 1)
        negative_control(1, 0.1)
    check("probe_dirs_removed", (len(works), [os.path.exists(w) for w in works]), (2, [False, False]))
    # A qs ipc call that outlives its timeout is an error row for its series, never an abort of the part.
    class HungIpc(Session):
        def __init__(self):
            self.qs_pid, self.view, self.closed, self.load, self.taken = 1, "list", False, None, []

        def close(self):
            self.closed = True

    def hang(argv, **kw):
        raise subprocess.TimeoutExpired(argv, kw["timeout"])

    hung, hung_out = HungIpc(), io.StringIO()
    with mock.patch.object(subprocess, "run", hang), mock.patch.dict(SERIES, {"pairs": lambda s, sp, n: [{"first_ms": s.ipc_int("cursor")}]}), \
            mock.patch.dict(globals(), {"Session": lambda *a, **kw: hung}):
        try:
            run_session({"fixture": "f", "view": "list", "region": "content", "series": [{"type": "pairs"}], "n": 1},
                        {"out_dir": os.devnull, "tag": "t"}, hung_out)
        except subprocess.SubprocessError as e:
            fail.append("ipc_timeout: the part aborted on %s" % type(e).__name__)
    check("ipc_timeout_row", ("TimeoutExpired" in hung_out.getvalue(), hung.closed), (True, True))
    # The fling floor: medians and ranges from fixed arrays, a failed run counted out of n.
    floor = fling_floor([17.0, 18.0, 16.5, None, 17.5], [20.1, 21.0, 19.8, 22.4, 20.5])
    check("fling_floor_wheel", floor["first_wheel_ms"], {"n": 4, "median": 17.2, "min": 16.5, "max": 18.0})
    check("fling_floor_frame", floor["first_frame_ms"], {"n": 5, "median": 20.5, "min": 19.8, "max": 22.4})
    check("fling_floor_runs", (floor["runs"], floor["stroke"]), (5, {"dy_mm": -40, "ms": 120}))
    check("fling_floor_none_read", fling_floor([None, None], [None, None])["first_frame_ms"],
          {"n": 0, "median": None, "min": None, "max": None})
    # The client's first pixel-carrying wheel event on the harness clock: the zero-delta skip and the swap line do not count.
    off = 1_760_000_000_000_000_000 - 7_000_000_000
    log = ["qml: CALIB wheel-skip angle=0", "qml: CALIB swap key=0 changed=1760000000011 swapped=1760000000030",
           "qml: CALIB wheel at=1760000000017 px=-3", "qml: CALIB wheel at=1760000000040 px=-9"]
    check("wheel_ms_first_pixel_event", wheel_ms_of(log, 7_000_000_000, off), 17.0)
    check("wheel_ms_none_logged", wheel_ms_of(log[:2], 7_000_000_000, off), None)
    check("clock_offset", abs(clock_offset_ns() - (time.time_ns() - time.monotonic_ns())) < 5_000_000, True)
    # fling_probe with the display, the uinput touchpad and the recorder replaced: three runs, one touchpad, each run's client ended.
    with tempfile.TemporaryDirectory() as td:
        logp = os.path.join(td, "client.log")
        with open(logp, "w") as f:
            f.write("qml: CALIB wheel-skip angle=0\nqml: CALIB wheel at=1760000000017 px=-3\n")
        made, dropped, ended, pointed, strokes = [], [], [], [], []

        class FakeRec:
            def __init__(self, *a):
                pass

            wait_quiet = lambda self, *a: True
            snapshot = lambda self: []
            stop = lambda self: None
        fake_T = types.SimpleNamespace(create=lambda: made.append(7) or 7, CREATE_SETTLE_S=0)
        fakes = dict(touchpad=lambda: (fake_T, None), calib_window=lambda work, delay: (
            "client", (100, 148, 900, 652), [(100, 100, 900, 48)], logp),
            pointer_to=lambda keys, x, y: pointed.append((x, y)), clock_offset_ns=lambda: off,
            stroke=lambda T, fd, mm, ms: strokes.append((fd, mm, ms)) or 7_000_000_000,
            drop_touchpad=lambda T, fd: dropped.append(fd), end_client=ended.append)
        with mock.patch.dict(globals(), fakes), mock.patch.object(time, "sleep", lambda secs: None), \
                mock.patch.object(S, "build_recorder", lambda d: "rec"), mock.patch.object(S, "Recorder", FakeRec), \
                mock.patch.object(S, "search_columns", lambda frames, times, *a: {"first_ms": 20.0}), \
                mock.patch.object(S, "park_cursor", lambda: None):
            probed = fling_probe(3, keys=object())
        shutil.rmtree(probed.pop("workdir"))
        check("fling_probe_floor", (probed["runs"], probed["first_wheel_ms"]["median"], probed["first_frame_ms"]["median"]),
              (3, 17.0, 20.0))
        check("fling_probe_one_touchpad", (made, dropped, len(strokes), strokes[0], ended), ([7], [7], 3, (7, -40, 120), ["client"] * 3))
        check("fling_probe_pointer", pointed, [(550, 474)] * 3)
        with mock.patch.dict(globals(), touchpad=lambda: (None, "no uinput")):
            check("fling_probe_skipped", fling_probe(3), {"skipped": "no uinput"})
    # The report: refused samples beside the misses, and the floor beside the fling rows only.
    fling_rows = [{"journey": "nav.list.fling", "row": 1, "view": "list", "first_ms": 22.0, "fixture": "local"}
                  for _ in range(20)]
    op_rows = [{"journey": "op.copy.large", "row": 6, "view": "list", "first_ms": None if i < 3 else 6.0, "fixture": "local",
                "refused": i < 2} for i in range(20)]
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as tf:
        for r in fling_rows + op_rows:
            tf.write(json.dumps(r) + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        report([tf.name], floor=floor)
    os.unlink(tf.name)
    lines = {ln.split("|")[2].strip(): ln for ln in buf.getvalue().splitlines() if ln.startswith("| 1 ") or ln.startswith("| 6 ")}
    check("rep_fling_floor", ("idle client floor, stroke to first wheel event p50 17.2 (16.5 to 18.0), to first frame p50 20.5 "
                              "(19.8 to 22.4) over 5 runs, -3.8 left under the 16.7 ms ceiling, not subtracted"
                              in lines["nav.list.fling"]), True)
    check("rep_op_refused_beside_miss", "3 miss; 2 refused" in lines["op.copy.large"], True)
    check("rep_op_has_no_floor", "floor" in lines["op.copy.large"], False)
    # The floor is read from a calibrate output of concatenated objects, and not read from one that has none.
    with tempfile.TemporaryDirectory() as td2:
        cal = os.path.join(td2, "calibrate.json")
        with open(cal, "w") as f:
            f.write(json.dumps({"negative_control": {"clean": True}}, indent=1) + "\n" + json.dumps({"frame_ms": 8.33}, indent=1)
                    + "\n" + json.dumps({"fling": floor}, indent=1) + "\n")
        check("fling_floor_of_file", fling_floor_of(cal), (floor, None))
        with open(cal, "a") as f:
            f.write("PASS\n")
        check("fling_floor_of_kept_past_garbage", fling_floor_of(cal), (floor, None))
        with open(cal, "w") as f:
            f.write(json.dumps({"negative_control": {"clean": True}}))
        check("fling_floor_of_none", fling_floor_of(cal), (None, cal + ": no fling object"))
        with open(cal, "a") as f:
            f.write("\nPASS\n")
        check("fling_floor_of_parse_error", fling_floor_of(cal), (None, cal + ": parse error at offset %d" % (
            len(json.dumps({"negative_control": {"clean": True}})) + 1)))
        gone = os.path.join(td2, "nope.json")
        check("fling_floor_of_missing", fling_floor_of(gone), (None, gone + ": No such file or directory"))
        with open(cal, "wb") as f:
            f.write(b"\xff\xfe")
        check("fling_floor_of_binary", fling_floor_of(cal), (None, cal + ": parse error at offset 0"))
        # A named calibrate file that gives no floor says why beside the fling rows; the file found beside the run stays quiet.
        run = os.path.join(td2, "run.jsonl")
        with open(run, "w") as f:
            for r in fling_rows:
                f.write(json.dumps(r) + "\n")

        def fling_line(*argv):
            out = io.StringIO()
            with redirect_stdout(out):
                main(["report", run, *argv])
            return next(ln for ln in out.getvalue().splitlines() if "nav.list.fling" in ln)
        check("rep_named_missing_says_why", "fling floor not read (%s: No such file or directory)" % gone in fling_line("--calibrate", gone), True)
        check("rep_named_bad_says_why", "fling floor not read (%s: parse error at offset 0)" % cal in fling_line("--calibrate", cal), True)
        with open(cal, "w") as f:
            f.write(json.dumps({"negative_control": {"clean": True}}))
        check("rep_beside_without_floor_is_quiet", "floor" in fling_line(), False)
    check("fling_note_skipped", fling_floor_note({"skipped": "/dev/uinput is not writable"}),
          "fling floor not read (/dev/uinput is not writable)")
    check("fling_note_absent", fling_floor_note(None), None)
    # calibrate: a fling probe that raises anything is printed as an error object and the gate keeps its own exit code.

    def calibrate_with(probe):
        out = io.StringIO()
        fakes = dict(negative_control=lambda runs: {"clean": True}, repeat_probe=lambda *a: {}, repeat_settings=lambda plan: {},
                     fling_probe=probe, calibrate_gate=lambda neg, rep: 7)
        with mock.patch.dict(globals(), fakes), mock.patch.object(S, "session_env", lambda: {}), mock.patch.object(S, "ENV", None), \
                mock.patch.object(S, "calibrate", lambda delays, runs: []), mock.patch.object(N, "NavKeys", object), \
                redirect_stdout(out):
            try:
                code = calibrate(1)
            except Exception as e:
                code = "raised " + type(e).__name__
        return code, out.getvalue()

    def raises(exc):
        def probe(runs, keys=None):
            raise exc
        return probe
    code, text = calibrate_with(raises(KeyError("floor")))
    check("calibrate_fling_keyerror", (code, '"error": "KeyError: \'floor\'"' in text), (7, True))
    code, text = calibrate_with(raises(subprocess.CalledProcessError(2, ["rec"])))
    check("calibrate_fling_calledprocesserror", (code, '"error": "CalledProcessError: ' in text), (7, True))
    code, text = calibrate_with(lambda runs, keys=None: {"skipped": "no uinput"})
    check("calibrate_fling_skipped_keeps_gate", (code, '"skipped": "no uinput"' in text), (7, True))
    # Loads: the probe is read per sample, so a load that stops after sample 2 reads inactive from sample 3.
    check("load_active_moved", (load_active(3, 4), load_active(4, 4), load_active(None, 4), load_active(4, None)),
          (True, False, False, False))
    check("load_active_busy", (load_active((7, True), (7, True)), load_active((7, False), (7, False)),
                               load_active((7, False), (8, False)), load_active((7, False), (7, True))),
          (True, False, True, True))
    check("load_moved_ignores_busy_before", (load_moved((7, True), (7, False)), load_moved((7, False), (7, True))),
          (False, True))

    def load_session(probe, renew=None, chord_log=None, bump=None):
        """A session with no display, a stub load, and a chord that bumps the load as a sample's input is sent."""
        ss = object.__new__(Session)
        ss.view, ss.taken, ss.known, ss.ctx = "dual", [], set(), {}

        def chord(*k):
            if k != ("shift",) and k != ("home",):
                if chord_log is not None:
                    chord_log.append(k)
                if bump:
                    bump()
            return 7
        ss.keys = types.SimpleNamespace(chord=chord)
        ss.rec = types.SimpleNamespace(wait_quiet=lambda *a: True, snapshot=lambda: [])
        ss.columns = lambda journey, *a: {"journey": journey, "first_ms": 1.0}
        ss.outsiders, ss.require_window = lambda: {}, lambda: None
        ss.load = Load("b", "stub", probe, renew)
        return ss

    cycle = {"type": "cycle", "steps": [{"journey": "nav.dual.enter", "in": ("key", ("enter",))}]}
    count = {"n": 0, "keys": 0}

    def stops_after_two():
        count["keys"] += 1
        if count["keys"] <= 2:
            count["n"] += 1
    rows = run_series(load_session(lambda: count["n"], bump=stops_after_two), cycle, 6)
    check("load_per_sample_stops", [r["load_active"] for r in rows], [True, True, False, False, False, False])
    check("load_per_sample_probes", [r["load_probe"] for r in rows[:3]], [[0, 1], [1, 2], [2, 2]])
    check("load_rows_name_the_load", {(r["load"], r["load_note"]) for r in rows}, {("b", "stub")})
    count.update(n=0, keys=0)
    with mock.patch.dict(globals(), PACE_MS=1), \
            mock.patch.object(S, "search_columns", lambda frames, times, *a: {"per_key_first_ms": [2.0] * len(times)}), \
            mock.patch.object(S, "changes_of", lambda frames: []), \
            mock.patch.object(S, "first_change_after", lambda changes, t: None):
        rows = run_series(load_session(lambda: count["n"], bump=stops_after_two),
                          {"type": "pairs", "pairs": [(("down",), ("up",))]}, 3)
    check("load_paced_per_key_stops", [r["load_active"] for r in rows], [True, True, False, False, False, False])
    # A load that ended before sample 1 marks every row inactive, so the gate names it.
    rows = run_series(load_session(lambda: 5), cycle, 4)
    check("load_never_active", [r["load_active"] for r in rows], [False] * 4)

    # load_a_waits_for_start: the renewal comes first and the input only after the probe moved since it.
    log, st = [], {"count": 0, "armed": False, "polls": 0}

    def probe_a():
        if st["armed"]:
            st["polls"] += 1
            if st["polls"] == 3:
                st["count"] += 1
        return st["count"]

    def renew_a():
        st.update(armed=True, polls=0)
        log.append(("renew", st["count"]))
    ss = load_session(probe_a, renew_a)
    ss.keys = types.SimpleNamespace(chord=lambda *k: (log.append(("chord", st["count"])) if k == ("enter",) else None) or 7)
    with mock.patch.dict(globals(), LOAD_POLL_S=0.0):
        rows = run_series(ss, cycle, 2)
    check("load_a_waits_for_start", log, [("renew", 0), ("chord", 1), ("renew", 1), ("chord", 2)])
    log.clear()
    ss = load_session(lambda: 0, lambda: log.append("renew"), log)
    with mock.patch.dict(globals(), LOAD_START_S=0.05, LOAD_POLL_S=0.005):
        rows = run_series(ss, cycle, 3)
    check("load_a_never_starts", (log, [r.get("error") for r in rows]), (["renew"], ["RuntimeError: the load did not start"]))

    def link_failing_at(n):
        """An os.link that raises on its nth call and links as usual on every other."""
        real, calls = os.link, [0]

        def link(a, b):
            calls[0] += 1
            if calls[0] == n:
                raise OSError("link failed")
            return real(a, b)
        return link
    # Load a's renewal links fresh names that sort just after the first name in dst, whatever src calls its files, onto the same inodes.
    with tempfile.TemporaryDirectory() as td:
        src, dst = os.path.join(td, "src"), os.path.join(td, "dst")
        os.makedirs(src)
        os.makedirs(dst)
        for i in range(1, 6):
            open(os.path.join(src, "c_%05d.jpg" % i), "w").close()
            os.link(os.path.join(src, "c_%05d.jpg" % i), os.path.join(dst, "p_%05d.jpg" % i))
        names, seq, base = sorted(os.listdir(src)), [COLD_SEQ], sorted(os.listdir(dst))
        first = [os.path.basename(x) for x in link_cold(src, names, dst, seq, 3)]
        second = [os.path.basename(x) for x in link_cold(src, names, dst, seq, 3)]
        listing = sorted(os.listdir(dst))
        check("cold_links_names", (first, second), (["p_00001.k%06d.jpg" % x for x in (999999, 999998, 999997)],
                                                    ["p_00001.k%06d.jpg" % x for x in (999996, 999995, 999994)]))
        check("cold_links_sort_after_the_first_row", (listing[0] == base[0], listing[1:4] == sorted(second),
                                                      listing[4:7] == sorted(first), listing[7:] == base[1:]),
              (True, True, True, True))
        inodes = {os.stat(os.path.join(src, x)).st_ino for x in names}
        check("cold_links_share_inodes", all(os.stat(os.path.join(dst, x)).st_ino in inodes for x in first + second), True)
        check("cold_links_distinct", (len(first), len(second), len(listing)), (3, 3, 11))
        part = []
        with mock.patch.object(os, "link", link_failing_at(3)):
            try:
                link_cold(src, names, dst, seq, 5, part)
                fail.append("cold_links_failure_propagates: a failing os.link raised nothing")
            except OSError:
                pass
        check("cold_links_name_only_what_exists", (len(part), all(os.path.lexists(x) for x in part)), (2, True))
    # Load a's renewal swaps the previous batch for a new one, so the right folder keeps its size through the run, and removes only inside a marked root.
    with tempfile.TemporaryDirectory() as td:
        open(os.path.join(td, MARKER), "w").close()
        cold, right = os.path.join(td, "load", "cold-src"), os.path.join(td, "load", "cold-x")
        os.makedirs(cold)
        os.makedirs(right)
        for i in range(1, 6):
            open(os.path.join(cold, "c_%05d.jpg" % i), "w").close()
            os.link(os.path.join(cold, "c_%05d.jpg" % i), os.path.join(right, "c_%05d.jpg" % i))
        renew_a = load_renew(types.SimpleNamespace(right=right), "a", {"root": td, "roots": [td]})
        sizes, seen = [], set()
        for _ in range(4):
            renew_a()
            sizes.append(len(os.listdir(right)))
            seen |= set(os.listdir(right))
        check("load_a_renewals_keep_the_folder_size", (sizes, len(os.listdir(cold))), ([5 + COLD_BATCH] * 4, 5))
        check("load_a_renewals_never_reuse_a_name", len(seen), 5 + 4 * COLD_BATCH)
        # A link that fails on the tenth of a batch leaves the folder at its five pictures, and the next renewal links a whole batch.
        base = ["c_%05d.jpg" % i for i in range(1, 6)]
        with mock.patch.object(os, "link", link_failing_at(10)):
            try:
                renew_a()
                fail.append("load_a_failed_renewal_propagates: a failing os.link raised nothing")
            except OSError:
                pass
        check("load_a_failed_renewal_leaves_no_partial_batch", sorted(os.listdir(right)), base)
        renew_a()
        check("load_a_renewal_recovers_after_a_failed_one", len(os.listdir(right)), 5 + COLD_BATCH)
    # With no marker on the root the renewal refuses before it removes anything, so the batch it linked is all still there.
    with tempfile.TemporaryDirectory() as bare:
        cold, other = os.path.join(bare, "load", "cold-src"), os.path.join(bare, "load", "cold-y")
        os.makedirs(cold)
        os.makedirs(other)
        open(os.path.join(cold, "c_00001.jpg"), "w").close()
        os.link(os.path.join(cold, "c_00001.jpg"), os.path.join(other, "c_00001.jpg"))
        unmarked = load_renew(types.SimpleNamespace(right=other), "a", {"root": bare, "roots": [bare]})
        unmarked()
        linked = sorted(os.listdir(other))
        try:
            unmarked()
            fail.append("load_a_renewal_refuses_outside_a_marked_root: a batch was removed with no marked root")
        except SystemExit:
            pass
        check("load_a_refused_renewal_removes_nothing", (len(linked), sorted(os.listdir(other)) == linked), (1 + COLD_BATCH, True))
    # A forced renewal waits out the load's gap since the last one, so load c's folder mtime has moved by the time the probe is read.
    stamps = []
    gap_load = Load("c", "stub", lambda: len(stamps), lambda: stamps.append(time.monotonic()), 0.2)
    gap_session = load_session(None)
    gap_session.load = gap_load
    check("load_renew_inside_the_gap_is_skipped", (gap_load.renew(), gap_load.renew(), len(stamps)), (True, False, 1))
    gap_session.load_ready()
    check("load_forced_renewal_waits_out_the_gap", (len(stamps), stamps[1] - stamps[0] >= 0.2 - 0.01), (2, True))
    # Load c's probe is the right pane's request count and listing flag, never a process tick count; the sample is dualState's shape (ui/Ipc.qml:126).
    ipc = {"out": '{"active": true, "focused": 0, "panes": [{"path": "/a", "total": 3, "cursor": 0, "loading": false, "selected": [], "focused": true, "listRequests": 4}, {"path": "/nas/stream", "total": 5000, "cursor": 0, "loading": true, "selected": [], "focused": false, "listRequests": 7}]}'}
    stub = types.SimpleNamespace(ipc=lambda name, *a: ipc["out"])
    probe_c, note_c = load_start(stub, "c", {"roots": []})
    check("load_c_probe", (probe_c(), note_c), ((7, True), "NAS listing streaming in the right pane"))
    # The probe reads the fields the producer writes: a pane built from the keys of dualState's own object literal.
    with open(os.path.join(REPO, "ui", "Ipc.qml")) as f:
        qml = f.read()
    literal = qml[qml.index("function dualState()"):]
    literal = literal[:literal.index("} : null")]
    keys = re.findall(r"(\w+):", literal.split("return pane ? {", 1)[1])
    ipc["out"] = json.dumps({"panes": [None, {k: 7 for k in keys}]})
    check("load_c_probe_reads_the_producers_fields", (probe_c(), "listRequests" in keys, "loading" in keys),
          ((7, True), True, True))
    ipc["out"] = "{"
    check("load_c_probe_unreadable", probe_c(), (None, False))
    ipc["out"] = '{"panes": [{"listRequests": 4, "loading": false}]}'
    check("load_c_probe_one_pane", probe_c(), (None, False))
    ipc["out"] = '{"active": true, "focused": 0, "panes": [{"listRequests": 4, "loading": false}, null]}'
    check("load_c_probe_null_pane", probe_c(), (None, False))
    # The gate: a load group passes only when every sample's load_active is true.
    idle = [dict(r) for r in full]
    next(r for r in idle if r["journey"] == "nav.dual.down" and where(r) == "local+load-a")["load_active"] = False
    code, gtext = run_gate(idle, "p7-load", 20)
    check("gate_load_inactive_code", code, 2)
    check("gate_load_inactive_line",
          "GATE MISSING nav.dual.down dual local+load-a load inactive in 1 of 20 samples" in gtext, True)
    check("gate_load_inactive_only_that_group", gtext.count("load inactive"), 1)
    code, gtext = run_gate([{k: v for k, v in r.items() if k != "load_active"} for r in full], "p7-load", 20)
    check("gate_load_unproved_is_inactive", (code, "load inactive in 20 of 20 samples" in gtext), (2, True))
    code, gtext = run_gate(full, "p7-load", 20)
    check("gate_load_all_active", (code, "load inactive" in gtext), (0, False))
    # Rows come only from a cycle that leaves the fixture as found; a rename or an exception fails the series and is put back; a dirty start refuses.
    with tempfile.TemporaryDirectory() as td:
        def fresh(name):
            for old in os.listdir(td):
                os.remove(os.path.join(td, old))
            open(os.path.join(td, name), "w").close()
        fresh("f_0001.txt")

        def rename_away(*_):
            os.rename(os.path.join(td, "f_0001.txt"), os.path.join(td, "f_0001ab.txt"))
            return {}

        def rename_then_raise(*_):
            rename_away()
            raise OSError("boom")
        sampled = []
        fake = types.SimpleNamespace(fixture=td, ctx={}, taken=[], send=lambda inp: None, settle=lambda *a: True,
                                     sample=lambda *a: sampled.append(a) or {}, require_window=lambda: None)
        spec = {"steps": [{"journey": "nav.list.pre-rename-j", "in": k("j")}], "restores": ["f_0001.txt"],
                "typed": "ab"}
        check("restores_kept", len(series_cycle(fake, spec, 1)), 1)
        fake.sample = rename_away
        try:
            series_cycle(fake, spec, 1)
            fail.append("restores_changed: a renamed file was not refused")
        except RuntimeError as e:
            check("restores_changed", "f_0001.txt, f_0001ab.txt" in str(e), True)
        check("restores_put_back", sorted(os.listdir(td)), ["f_0001.txt"])
        fresh("f_0001.txt")
        fake.sample = rename_then_raise
        try:
            series_cycle(fake, spec, 1)
            fail.append("restores_raise: the exception did not reach the caller")
        except OSError as e:
            check("restores_raise", str(e), "boom")
        check("restores_raise_put_back", sorted(os.listdir(td)), ["f_0001.txt"])
        fresh("f_0001.txt")

        def rename_partial(*_):
            os.rename(os.path.join(td, "f_0001.txt"), os.path.join(td, "f_0001a.txt"))
            return {}
        fake.sample = rename_partial
        try:
            series_cycle(fake, spec, 1)
            fail.append("restores_partial: a partial-suffix rename was not refused")
        except RuntimeError as e:
            check("restores_partial", "f_0001.txt, f_0001a.txt" in str(e), True)
        check("restores_partial_put_back", sorted(os.listdir(td)), ["f_0001.txt"])
        fresh("f_0001ab.txt")
        fake.sample = lambda *a: sampled.append(a) or {}
        del sampled[:]
        try:
            series_cycle(fake, spec, 1)
            fail.append("restores_drifted: a fixture off its names was not refused")
        except RuntimeError as e:
            check("restores_drifted", ("starts off its expected names" in str(e), sampled), (True, []))
        check("restores_drifted_untouched", sorted(os.listdir(td)), ["f_0001ab.txt"])
    if fail:
        print("\n".join("FAIL " + f for f in fail))
        return 1
    print("plan cases: full planned set passes, one row short is MISSING, a miss p95 is a BREACH")
    print("expected-set cases: p1 without p4 is MISSING part p4-open, no paths is usage error")
    print("samples cases: samples=2 is MISSING plan samples 2, defined 20, floor holds except .first")
    print("rows cases: wheel 2 and first 1 share SERIES_ROWS_PER_SAMPLE with planned_counts")
    print("clean-outside cases: own parent and outside run under every parent, skip is failure")
    print("isolation cases: Esc count over running, cancelling and queued errors, refused pastes marked, an unclear bar is an error row")
    print("op navigation cases: every sample of a move or copy pastes in a-dst whatever the symlink sorts as, the old blind keys paste in b-src and take nothing from a-dst, "
          "a pane left in a-dst is walked back, an unreachable a-dst, an empty take and a listing that never lands are harness fault rows, "
          "a series stops after repeated faults, a take waits for a late listing, a foreign clipboard of the right size is a fault, a resolved climb is named, Backspace climbs the pane path as Nav.js cuts it, op.size.recursive shows the Properties row, the report names a harness fault")
    print("fling floor cases: medians and ranges from fixed arrays, the client wheel line on the harness clock, the floor beside the fling rows, "
          "a floor kept past trailing text, a named file that gives none says why, calibrate survives any probe error")
    print("load cases: the probe is read per sample and per paced key, a load that stops reads inactive from there, a renewed load is awaited before the input and waits out its gap, a renewal swaps its batch inside a marked root, the c probe reads the fields ui/Ipc.qml writes, the gate names an inactive load group")
    print("target cases: dual pointer in the left pane, Enter row from the listing, --only plans, an error or an external window keeps the rows built before it, folders first pinned")
    print("SELFTEST ok (%d journeys across %d parts)" % (len(cover), len(PARTS)))
    return 0


def main(argv):
    if not argv or argv[0] in ("-h", "--help"):
        print(__doc__)
        return 0
    cmd, rest = argv[0], argv[1:]

    def opt(name, default=None):
        return rest[rest.index(name) + 1] if name in rest else default

    def jsonl_paths(args):
        """Existing .jsonl files past --out FILE and --part PART; anything else fails loud."""
        paths, skip = [], False
        for a in args:
            if skip:
                skip = False
                continue
            if a in ("--out", "--part", "--only", "--calibrate"):
                skip = True
                continue
            if not a.endswith(".jsonl") or not os.path.isfile(a):
                print("need an existing .jsonl file, not %r" % a, file=sys.stderr)
                raise SystemExit(2)
            paths.append(a)
        return paths

    if cmd == "selftest":
        return selftest()
    if cmd == "budget":
        return budget_json()
    if cmd == "report":
        paths = jsonl_paths(rest)
        calib = opt("--calibrate")
        named = calib is not None
        if calib is None and paths:  # the census script writes calibrate.json into the same out dir
            calib = os.path.join(os.path.dirname(os.path.abspath(paths[0])), "calibrate.json")
        floor, why = fling_floor_of(calib) if calib else (None, None)
        if floor is None and named:  # a file the operator named says why it gave no floor
            floor = {"error": why}
        return report(paths, opt("--out"), floor)
    if cmd == "gate":
        only = []
        for i, a in enumerate(rest):
            if a == "--part":
                raw = rest[i + 1] if i + 1 < len(rest) else None
                hit = resolve_part(raw) if raw else None
                if hit is None:
                    print("unknown part %r, want one of %s" % (raw, ", ".join(PARTS)), file=sys.stderr)
                    raise SystemExit(2)
                only.append(hit)
        return gate(jsonl_paths(rest), opt("--out"), None if not only else only, opt("--only"))
    if cmd == "fixtures":
        print(json.dumps(build_fixtures(rest[0], opt("--usb"), opt("--nas")), indent=1))
        return 0
    if cmd == "clean":
        return clean(rest[0])
    if cmd == "calibrate":
        return calibrate(int(opt("--runs", 10)))
    if cmd == "run":
        return run_part(rest[0], opt("--root"), opt("--flea"), opt("--out"), int(opt("--samples", 20)),
                        "--dry" in rest, "--no-evidence" not in rest, opt("--only"))
    if cmd == "measure":
        with open(rest[0]) as f:
            plan = json.load(f)
        return run_part(plan["part"], plan["root"], plan["flea"], plan["out"], plan.get("samples", 20),
                        plan.get("dry", False), plan.get("evidence", True), plan.get("only"))
    print("unknown command " + cmd, file=sys.stderr)
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))
