#!/usr/bin/env python3
"""qemu_session.py -- one typed QEMU session against a kernel boot, and its verdict.

    scripts/qemu_session.py MODE ARCH [--label L] [--make "VAR=val ..."] [--no-zuse]

Builds and boots `make ARCH=<arch> clean qemu` (one QEMU, in the foreground of
this script). Its disks are the Makefile's defaults -- on v4, a fresh data stick
and nothing else, so every boot is a first boot (plan 2026-10-29-stick-5,
Task 3) -- except for the modes that ask for the minted Zuse pair
(disk/zuse-scratch/) by name: copies of it, made under
build/qemu-session/<label>/ so the committed images are never booted, laid out
as a stick (zusestick.img), or her raw identity image for v3.
It reads the serial socket continuously, drives the boot over the serial and
QMP sockets the `qemu` goal opens, prints PASS:/FAIL: lines and
RESULT: PASS|FAIL, and quits QEMU over QMP. The serial log is the usual
logs/<TS>/<arch>/qemu-<arch>-<TS>.log.

Modes (the identity gate's, plan 2026-10-26-engine-step4-identity-gate.md, and
the stick's, plan 2026-10-26-stick-1-screen-and-keyboard.md):
  headless   no Zuse drive: POST: PASSED, then no `ok> ` for 60 s
  zuse       the default disks, a fresh data stick: the genesis line, [zuse@Hera] ok>,
             `3 4 + .` -> 7  ok (plan 2026-10-29-stick-5, Task 3: it was the minted pair)
  hotplug    no drive; Zuse added over QMP -> prompt; pulled -> session ended,
             headless; added again -> prompt
  midpull    pulled while a line is in hand: what is typed after does not run
  waitpull   pulled while a line WAITS: the wait ends at logout; re-attach -> prompt
  typeahead  serial bytes typed while headless do not run once Zuse attaches
  emergency  an EMERGENCY_CONSOLE_ENABLED=1 build: ok> with no drive
  v3_keys    system=v3: `3 4 + .` typed on QEMU's keyboard (send-key) -> 7  ok
  screen_dark_until_login     no drive: 30 s after POST the screen is all black
  screen_console_after_login  Zuse added: the screen shows the console; keys typed
                              on QEMU's keyboard answer 7  ok and change the screen
  screen_dark_after_logout    ... and when her drive is pulled, the screen is black again
  kbd_typeahead               keys pressed while headless do not run once Zuse attaches
  usb_kbd                     the only keyboard a USB one (--make "QEMU_EXTRA='-machine
                              i8042=off -device usb-kbd,bus=xhci0.0,port=3'"): it types
  no_framebuffer              no display (-vga none): serial alone, 7  ok
  alttab_after_logout         after logout, Alt+TAB twice: the screen stays black
  alttab_before_pull          Alt+TAB, then the drive pulled: black; re-attach draws the console
  emergency_pull              EMERGENCY_CONSOLE_ENABLED=1: Zuse added and pulled, the screen
                              still shows the running console, and it types
  gpt_zuse     a GPT drive, Zuse in its StarshipOS identity partition: recognised; the partition
               is attached (run with --make KERNEL_ARGS=--log-level=info)
  gpt_hotplug  ... pulled: session ended; added again: prompt
  gpt_foreign  a GPT drive with only a Linux partition: no identity, not attached (info level)
  gpt_bad_crc  the gpt_zuse image with its header CRC broken: not used, said so, not attached
  gpt_bad_entries  ... its entry array changed (the header intact): the same
  gpt_no_header    ... its "EFI PART" signature gone, the protective MBR kept: the same
  gpt_read_error   the raw Zuse drive with reads of its sector 0 failing (blkdebug): not used,
                   said so, not attached
  virtio_kbd                  aarch64/riscv64: QEMU's virtio keyboard types 7  ok
  stick_boot   --target qemu-stick (amd64): the PC stick booted as a USB stick through OVMF; its
               blank identity partition attached (500 MiB, not the 2 GiB disk); the proxy Zuse
               minted onto it (genesis), [zuse@Hera] ok> (plan 2026-10-29-stick-5, Task 1)
  genesis      a fresh data stick (DATASTICK=1, no Zuse drive), booted twice, one QEMU at a time:
               first the genesis line, [zuse@Hera] ok>, 7  ok; then the same copy again:
               identity confirmed, no genesis, the prompt (plan 2026-10-29-stick-5, Task 1)
  stick_genesis  --target qemu-stick, the PC stick as built (no STICK_PAIR): its first boot mints
               the proxy Zuse and reaches [zuse@Hera] ok> with nothing set up (plug and play)
  stick_grow   --target qemu-stick --stick-gib 16: the first boot grows Artemis to fill the stick
               (over 12000000 user blocks), mints, the prompt, its time POST -> prompt printed; the
               second boot of the same copy is not grown again and recognises Zuse; sfdisk --verify
               on the copy is clean (plan 2026-10-29-stick-5, Task 2)
  stick_small  --target qemu-stick --stick-gib 4: under 8 GB -- not grown, said so; Artemis at the
               image's size; the prompt
  stick_artemis  ARTDISK= (no virtio disk): Artemis and Zuse from the Zuse stick alone -- the
               stick's Artemis partition attached first (785938 user blocks), then its identity
               (48724); [zuse@Hera] ok> (plan 2026-10-27-stick-4-one-disk.md, Task 4)
  stick_zuse   --target qemu-stick, the PC stick with Zuse's pair in it: a stick, Artemis attached,
               [zuse@Hera] ok>
  two_sticks   --make "DATASTICK=1 ...": Zuse's stick on port 1, a blank data stick on port 2;
               Zuse logs in; the data stick is "not this system's stick -- not used" (replaces
               datastick)
  artemis_unsigned  the Zuse stick with its Artemis signature zeroed: "signature is bad -- not
               used", nothing attached, headless
  raw_ident    Zuse's raw identity image, no GPT: "not this system's stick -- not used", headless
  screen_boot  KERNEL_ARGS=screen=boot, no drive: "Screen: the console stays on until login
               (screen=boot)"; 30 s after POST the screen is not black; no ok> (plan
               2026-10-27-stick-4-one-disk.md, Task 5)
  stick_v3     --target qemu-stick, system=v3 in the stick's starforth.cfg: "Boot disk: a StarshipOS
               stick", the refusal, then v4's POST; no v3 (plan 2026-10-27-stick-4-one-disk.md, Task 3)
  entropy      after "System: StarForth v4", the entropy backend -- rdrand on amd64 (qemu and
               --target qemu-stick), virtio-rng on aarch64 and riscv64 -- then "entropy: ready";
               on the qemu target, [zuse@Hera] ok> (plan 2026-10-27-stick-4-one-disk.md, Task 2)

Needs only Python 3. One QEMU at a time (.claude/CLAUDE.md).
"""
import argparse, glob, json, os, re, shlex, shutil, socket, subprocess, sys, threading, time

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
PROMPT = r"\[zuse@Hera\][^\n]*ok> "
ANYPROMPT = r"ok> "

ap = argparse.ArgumentParser()
ap.add_argument("mode")
ap.add_argument("arch")
ap.add_argument("--label", default=None)
ap.add_argument("--make", default="")
ap.add_argument("--no-zuse", action="store_true")
ap.add_argument("--target", default="qemu")      # qemu-stick: boot the PC stick (amd64)
ap.add_argument("--stick-gib", type=int, default=None)   # qemu-stick: the copy extended to N GiB
a = ap.parse_args()

NO_ZUSE_MODES = {"gpt_zuse", "gpt_hotplug", "gpt_foreign", "gpt_bad_crc", "gpt_bad_entries", "gpt_no_header",
                 "gpt_read_error", "headless", "hotplug", "midpull", "waitpull", "typeahead", "emergency",
                 "alttab_after_logout", "alttab_before_pull", "emergency_pull",
                 "screen_dark_until_login", "screen_console_after_login", "screen_dark_after_logout",
                 "kbd_typeahead", "stick_boot", "stick_v3", "stick_zuse", "screen_boot", "genesis",
                 "stick_genesis", "stick_grow", "stick_small"}
label = a.label or f"{a.mode}-{a.arch}"
work = os.path.join(ROOT, "build", "qemu-session", label)
shutil.rmtree(work, ignore_errors=True)
os.makedirs(work)
for f in ("artemis.img", "zuse-thumb-ident.img"):
    subprocess.run(["cp", "--sparse=always", os.path.join(ROOT, "disk/zuse-scratch", f), work], check=True)
art = os.path.join(work, "artemis.img")
ident = os.path.join(work, "zuse-thumb-ident.img")     # the raw identity image (GPT-mode payloads)
# Zuse's drive is her pair as a stick (plan 2026-10-27-stick-4-one-disk.md, Task 1): identity and
# Artemis partitions on one disk, built from this session's copies
zimg = os.path.join(work, "zusestick.img")
subprocess.run(["bash", os.path.join(ROOT, "scripts/mkdiskimage.sh"), "--out", zimg, "--table", "gpt",
                "--label", "ZUSESTICK", "--esp-mib", "0", "--starship-from", ident, art],
               check=True, stdout=subprocess.DEVNULL)
if a.mode == "artemis_unsigned":
    # the Zuse stick with the 4 KiB artemis_sig_t zeroed: it is devblock total - 1 - 64
    # (artemis_sig.c, ARTEMIS_SIG_DEVBLOCK_FROM_TOP), i.e. 65 4 KiB devblocks below the top of the
    # Artemis partition, which starts 1 MiB + |ident| in
    unsigned = os.path.join(work, "unsigned.img")
    subprocess.run(["cp", "--sparse=always", zimg, unsigned], check=True)
    sig_at = 1048576 + os.path.getsize(ident) + os.path.getsize(art) - 65 * 4096
    with open(unsigned, "r+b") as f:
        f.seek(sig_at)
        f.write(bytes(4096))
# the modes that ask for the minted pair by name (plan 2026-10-29-stick-5, Task 3); the others
# boot the Makefile's defaults, or no drive at all
PAIR_MODES = {"stick_artemis", "two_sticks", "artemis_unsigned", "raw_ident", "v3_keys"}

# v3 reads identity drives raw, never through a GPT (only the v4 path reads one: sk_v4_identity.c),
# so a system=v3 boot gets Zuse's raw identity image, not her stick (plan 2026-10-27-stick-4, R2)
zdrive = {"v3_keys": ident, "raw_ident": ident}.get(a.mode, zimg)
if a.mode == "artemis_unsigned":
    zdrive = unsigned
make_args = ["make", "-f", "kernel/Makefile", f"ARCH={a.arch}", "clean", a.target]
if a.mode in PAIR_MODES:
    make_args += [f"ZUSEDISK={zdrive}", "DATASTICK="]
elif a.no_zuse or a.mode in NO_ZUSE_MODES:
    make_args += ["ZUSEDISK=", "DATASTICK="]                # no drive at boot
if a.mode == "v3_keys":
    make_args.append(f"ARTDISK={art}")                      # v3's Artemis is the virtio disk
make_args += shlex.split(a.make)
if a.mode in ("v3_keys", "stick_v3"):
    make_args.append("KERNEL_ARGS=system=v3")
if a.mode == "screen_boot":
    make_args.append("KERNEL_ARGS=screen=boot")
if a.mode == "stick_zuse":
    make_args.append(f"STICK_PAIR={work}")          # the PC stick with this session's copy of the pair
if a.stick_gib:
    make_args.append(f"STICK_GIB={a.stick_gib}")
if a.mode in ("stick_grow", "stick_small"):
    make_args.append("KERNEL_ARGS=--log-level=info")   # the blk: disk lines the sizes are read from
if a.mode == "genesis":
    # a fresh data stick, made once into this session's directory and booted twice: the second
    # boot is the same copy, so it finds what the first wrote (plan 2026-10-29-stick-5, Task 1)
    dstick = os.path.join(work, "data.img")
    subprocess.run(["make", "-f", "kernel/Makefile", f"ARCH={a.arch}", "datastick", f"DATASTICK_IMG={dstick}"],
                   cwd=ROOT, check=True, stdout=subprocess.DEVNULL)
    make_args += ["DATASTICK=1", f"DATASTICK_IMG={dstick}", "DATASTICK_DEP="]
def newest(pattern):
    c = [p for p in glob.glob(pattern) if os.path.getmtime(p) >= start]
    return max(c, key=os.path.getmtime) if c else None


def drain():
    while True:
        try:
            if not s.recv(4096):
                return
        except OSError:
            return


def qcmd(cmd, **args):
    msg = {"execute": cmd}
    if args:
        msg["arguments"] = args
    q.send(json.dumps(msg).encode())
    while True:
        r = json.loads(qf.readline())
        if "return" in r or "error" in r:
            if "error" in r:
                print(f"QMP {cmd}: {json.dumps(r)}", flush=True)
            return r


def boot(args):
    """One `make ... qemu` boot: QEMU in the foreground of make, its serial log, serial and QMP
    sockets found and connected (the globals every check reads)."""
    global start, mk, log, s, q, qf
    start = time.time()
    make_log = open(os.path.join(work, "make.txt"), "a")
    mk = subprocess.Popen(args, cwd=ROOT, stdout=make_log, stderr=subprocess.STDOUT)
    deadline = time.time() + 1800
    log = ser = qmp = None
    while time.time() < deadline and not (log and ser and qmp):
        if mk.poll() is not None:
            print(f"MAKE FAILED (exit {mk.returncode}); see {make_log.name}", flush=True)
            sys.exit(1)
        log = log or newest(f"{ROOT}/logs/*/{a.arch}/qemu-{a.arch}-*.log")
        ser = ser or newest(f"{ROOT}/build/{a.arch}/kernel/qemu-serial-{a.arch}-*.sock")
        qmp = qmp or newest(f"{ROOT}/build/{a.arch}/kernel/qemu-qmp-{a.arch}-*.sock")
        time.sleep(1)
    print(f"LOG: {os.path.relpath(log, ROOT)}", flush=True)
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    for _ in range(60):
        try:
            s.connect(ser)
            break
        except OSError:
            time.sleep(1)
    threading.Thread(target=drain, daemon=True).start()
    q = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    q.connect(qmp)
    qf = q.makefile("r")
    qf.readline()
    qcmd("qmp_capabilities")


def shutdown():
    """Quit QEMU over QMP and let make finish: one QEMU at a time (.claude/CLAUDE.md)."""
    q.send(json.dumps({"execute": "quit"}).encode())
    mk.wait()
    time.sleep(2)


ok = True
boot(make_args)


def text():
    with open(log, errors="replace") as f:
        return f.read()


def count(pat):
    return len(re.findall(pat, text()))


def wait_for(pat, secs, after=0):
    t = time.time()
    while time.time() - t < secs:
        if count(pat) > after:
            return True
        if "POST: FAILED" in text() or "PARITY:FAIL" in text():
            return False
        time.sleep(0.5)
    return False


def excerpt(near=None, after=20, tail=40):
    """The serial log where a failed check looked (#136): the lines after the first match of
    `near`, else the log's last lines -- control characters shown as escapes, so a terminal's
    sequences cannot garble the job's output."""
    lines = text().splitlines()
    start = None
    if near is not None:
        start = next((i for i, l in enumerate(lines) if re.search(near, l)), None)
    if start is None:
        start, what = max(0, len(lines) - tail), f"the log's last {tail} lines"
        stop = len(lines)
    else:
        stop, what = min(len(lines), start + 1 + after), f"after /{near}/, line {start + 1}"
    print(f"  --- serial log, {what} ({os.path.relpath(log, ROOT)}):", flush=True)
    for i in range(start, stop):
        shown = "".join(c if c.isprintable() else repr(c)[1:-1] for c in lines[i])
        print(f"  {i + 1:6} | {shown}", flush=True)
    print("  ---", flush=True)


def check(name, cond, near=None):
    """PASS or FAIL; a FAIL also prints the serial log where the check looked: after the first
    line matching `near` (a regex), when it is given, else the log's end."""
    global ok
    print(f"{'PASS' if cond else 'FAIL'}: {name}", flush=True)
    if not cond:
        excerpt(near)
    ok = ok and cond


def type_line(line):
    """Over the serial line, a character at a time."""
    for ch in line:
        s.send(ch.encode())
        time.sleep(0.05)
    s.send(b"\r")


# QEMU qcodes for what the tests type on a keyboard
QCODE = {" ": ["spc"], ".": ["dot"], "+": ["shift", "equal"], "-": ["minus"], "\n": ["ret"]}


def key_line(line):
    """On QEMU's keyboard (send-key), a key at a time."""
    for ch in line + "\n":
        keys = QCODE.get(ch, [ch.lower()])
        qcmd("send-key", keys=[{"type": "qcode", "data": k} for k in keys])
        time.sleep(0.1)


# The two StarshipOS partition type GUIDs (docs/v4.0.0/plans/2026-10-26-stick-2-gpt-reader.md)
GPT_IDENTITY = "23BEBFE1-4817-4CF3-A537-9069BF159788"
GPT_LINUX = "0FC63DAF-8483-4772-8E79-3D69D8477DE4"


def make_gpt_image(path, ptype, payload, mib=72, break_crc=False, break_entries=False, break_sig=False):
    """A GPT disk image: one partition at LBA 2048, 131072 sectors (64 MiB), of type
    `ptype`, holding `payload` from its start; optionally the primary header's CRC broken,
    a byte of the entry array changed (the header left valid), or the header's "EFI PART"
    signature zeroed (the protective MBR left in place)."""
    subprocess.run(["truncate", "-s", f"{mib}M", path], check=True)
    sf = shutil.which("sfdisk") or "/usr/sbin/sfdisk"
    subprocess.run([sf, "--quiet", path], input=f"label: gpt\nstart=2048, size=131072, type={ptype}\n",
                   text=True, check=True)
    subprocess.run(["dd", f"if={payload}", f"of={path}", "bs=512", "seek=2048", "conv=notrunc,sparse",
                    "status=none"], check=True)
    if break_crc:
        with open(path, "r+b") as f:
            f.seek(512 + 16)                         # the primary header's HeaderCRC32
            b = f.read(1)
            f.seek(512 + 16)
            f.write(bytes([b[0] ^ 0xFF]))
    if break_entries:
        with open(path, "r+b") as f:
            f.seek(1024 + 56)                        # entry 0's name, inside the CRC'd entry array
            b = f.read(1)
            f.seek(1024 + 56)
            f.write(bytes([b[0] ^ 0xFF]))
    if break_sig:
        with open(path, "r+b") as f:
            f.seek(512)                              # the primary header's Signature
            f.write(bytes(8))


def blk_lines():
    return count(r"blk: ")


def alt_tab():
    qcmd("send-key", keys=[{"type": "qcode", "data": "alt"}, {"type": "qcode", "data": "tab"}])
    time.sleep(1)


SHOTS = 0


def screen():
    """A screendump of QEMU's display: (width, height, number of non-black pixels, bytes)."""
    global SHOTS
    SHOTS += 1
    path = os.path.join(work, f"screen-{SHOTS}.ppm")
    qcmd("screendump", filename=path)
    for _ in range(20):
        if os.path.exists(path) and os.path.getsize(path) > 0:
            break
        time.sleep(0.5)
    with open(path, "rb") as f:
        data = f.read()
    parts = data.split(b"\n", 3)                      # P6, "w h", maxval, pixels
    w, h = (int(x) for x in parts[1].split())
    px = parts[3]
    lit = sum(1 for i in range(0, len(px) - 2, 3) if px[i] or px[i + 1] or px[i + 2])
    return w, h, lit, px


def add_zuse(n, img=None, fail_sector0=False):
    f = {"driver": "file", "filename": img or zimg}
    if fail_sector0:                                 # every read touching sector 0 fails, EIO
        f = {"driver": "blkdebug", "image": f,
             "inject-error": [{"event": "none", "iotype": "read", "sector": 0, "errno": 5}]}
    qcmd("blockdev-add", driver="raw", **{"node-name": f"zuse{n}"}, file=f)
    qcmd("device_add", driver="usb-storage", bus="xhci0.0", port="1",
         drive=f"zuse{n}", id=f"zusedev{n}")


def del_zuse(n):
    qcmd("device_del", id=f"zusedev{n}")
    time.sleep(3)
    qcmd("blockdev-del", **{"node-name": f"zuse{n}"})


check("POST: PASSED", wait_for(r"POST: PASSED", 900))
m = a.mode

if m == "headless":
    check("no ok> for 60 s after POST", not wait_for(ANYPROMPT, 60))

elif m == "zuse":
    check("Boot disk: not a StarshipOS stick (an ISO or ESP boot)",
          re.search(r"Boot disk: not a StarshipOS stick", text()) is not None, near=r"Boot successful!")
    check("Zuse: genesis minted onto attached thumbdrive (a fresh data stick)",
          wait_for(r"Zuse: genesis minted onto attached thumbdrive", 180), near=r"Artemis: |Zuse: |USB: ")
    check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
    n = count(PROMPT)
    type_line("3 4 + .")
    check("7  ok, and a prompt", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m in ("stick_boot", "two_sticks"):
    def disks():
        # "us", not "user": the kernel cuts log messages at 63 characters (LOG_LINE_MAX 64,
        # v3/include/vm.h), and with seven-digit LBNs the attach line ends "(383542 us"
        return [int(u) for u in re.findall(r"blk: disk .*?\((\d+) us", text())]
    if m == "stick_boot":
        # the PC stick: Artemis is the rest of 2 GiB (well over 1 million user blocks), its blank
        # identity 500 MiB (about 380000)
        check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", text()) is not None, near=r"Boot successful!")
        check("System: StarForth v4", re.search(r"System: StarForth v4", text()) is not None)
        check("Artemis: the stick's partition attached",
              wait_for(r"Artemis: the stick's partition attached", 120))
        check("Zuse: genesis minted onto attached thumbdrive",
              wait_for(r"Zuse: genesis minted onto attached thumbdrive", 120))
        check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
        d = disks()
        check(f"Artemis attached first, then the identity partition (user blocks {d})",
              len(d) >= 2 and d[0] > 1000000 and 300000 < d[1] < 512000)
    else:
        check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
        check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
        n = count(PROMPT)
        type_line("3 4 + .")
        check("7  ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
        check(f"two USB drives found ({count(r'blkio_usb: MSC device ready')})",
              count(r"blkio_usb: MSC device ready") >= 2)
        time.sleep(5)
        nu = count(r"USB: a drive that is not this system's stick -- not used")
        check(f"the data stick: not this system's stick -- not used (said {nu} times)", nu == 1)
        d = disks()
        check(f"nothing of the data stick attached (user blocks {d})", not any(300000 < u < 512000 for u in d))
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m in ("stick_artemis", "stick_zuse"):
    if m == "stick_zuse":
        check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", text()) is not None, near=r"Boot successful!")
    check("Artemis: the stick's partition attached", wait_for(r"Artemis: the stick's partition attached", 120))
    check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
    check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
    if m == "stick_artemis":
        d = [int(u) for u in re.findall(r"blk: disk .*?\((\d+) us", text())]
        check(f"Artemis first (785938 user blocks), then the identity (48724): {d}", d[:2] == [785938, 48724])
    n = count(PROMPT)
    type_line("3 4 + .")
    check("7  ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m in ("artemis_unsigned", "raw_ident"):
    if m == "artemis_unsigned":
        check("Artemis: the stick's signature is bad -- not used",
              wait_for(r"Artemis: the stick's signature is bad -- not used", 120))
    else:
        check("USB: a drive that is not this system's stick -- not used",
              wait_for(r"USB: a drive that is not this system's stick -- not used", 120))
    check("no ok> for 60 s", not wait_for(ANYPROMPT, 60))
    check("no identity", count(r"Zuse: identity confirmed") == 0)
    check("no blk: disk line (nothing attached)", count(r"blk: disk ") == 0)
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m in ("genesis", "stick_genesis"):
    if m == "stick_genesis":
        check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", text()) is not None,
              near=r"Boot successful!")
    check("Zuse: genesis minted onto attached thumbdrive",
          wait_for(r"Zuse: genesis minted onto attached thumbdrive", 180), near=r"Artemis: |Zuse: ")
    check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
    check("no identity confirmed on a first boot", count(r"Zuse: identity confirmed") == 0)
    n = count(PROMPT)
    type_line("3 4 + .")
    check("7  ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
    if m == "genesis":
        shutdown()
        print("-- the second boot, the same copy of the data stick", flush=True)
        boot([x for x in make_args if x != "clean"])
        check("POST: PASSED (second boot)", wait_for(r"POST: PASSED", 900))
        check("Zuse: identity confirmed (second boot)",
              wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120), near=r"Artemis: |Zuse: ")
        check("no genesis on the second boot", count(r"Zuse: genesis minted") == 0)
        check("[zuse@Hera] ok> (second boot)", wait_for(PROMPT, 60))
        n = count(PROMPT)
        type_line("3 4 + .")
        check("7  ok (second boot)", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
        check("no fault (second boot)", count(r"EXCEPTION|PANIC|panic") == 0)

elif m in ("stick_grow", "stick_small"):
    def artemis_blocks():
        u = re.findall(r"blk: disk .*?\((\d+) us", text())
        return int(u[0]) if u else None
    t0 = time.time()
    if m == "stick_grow":
        check("Artemis: grown to fill the stick", wait_for(r"Artemis: grown to fill the stick", 300),
              near=r"Artemis: |GPT: ")
    else:
        check("Artemis: the stick is under 8 GB -- not grown",
              wait_for(r"Artemis: the stick is under 8 GB -- not grown", 300), near=r"Artemis: |GPT: ")
    check("Zuse: genesis minted onto attached thumbdrive",
          wait_for(r"Zuse: genesis minted onto attached thumbdrive", 300), near=r"Artemis: |Zuse: ")
    check("[zuse@Hera] ok>", wait_for(PROMPT, 120))
    print(f"TIME: POST to the prompt, first boot: {time.time() - t0:.0f} s", flush=True)
    b1 = artemis_blocks()
    if m == "stick_grow":
        check(f"Artemis over 12000000 user blocks ({b1})", b1 is not None and b1 > 12000000)
    else:
        check(f"Artemis at the image's size, 1000000..1200000 user blocks ({b1})",
              b1 is not None and 1000000 < b1 < 1200000)
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
    if m == "stick_grow":
        shutdown()
        print("-- the second boot, the same copy of the stick", flush=True)
        boot([x for x in make_args if x != "clean"] + ["STICK_FRESH=0"])
        check("POST: PASSED (second boot)", wait_for(r"POST: PASSED", 900))
        check("Zuse: identity confirmed (second boot)",
              wait_for(r"Zuse: identity confirmed from attached thumbdrive", 300), near=r"Artemis: |Zuse: ")
        check("[zuse@Hera] ok> (second boot)", wait_for(PROMPT, 120))
        check("not grown again", count(r"Artemis: grown") == 0)
        b2 = artemis_blocks()
        check(f"the same Artemis size ({b1} -> {b2})", b2 == b1)
        check("no fault (second boot)", count(r"EXCEPTION|PANIC|panic") == 0)
        shutdown()
        sf = shutil.which("sfdisk") or "/usr/sbin/sfdisk"
        run = os.path.join(ROOT, "build", a.arch, "kernel", "stick-run.img")
        v = subprocess.run([sf, "--verify", run], capture_output=True, text=True)
        print(v.stdout.strip() + v.stderr.strip(), flush=True)
        check("sfdisk --verify on the grown copy: clean",
              v.returncode == 0 and "No errors detected" in v.stdout + v.stderr)
        boot_done = True

elif m == "stick_v3":
    t = text()
    check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", t) is not None)
    check("System: v3 is not on this stick -- StarForth v4",
          re.search(r"System: v3 is not on this stick -- StarForth v4", t) is not None)
    check("v4's POST (PARITY:V4_POST)", re.search(r"PARITY:V4_POST tests=\d+ pass=\d+ fail=0", t) is not None)
    check("no System: StarForth v3", re.search(r"System: StarForth v3", t) is None)
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m == "entropy":
    want = "rdrand" if a.arch == "amd64" else "virtio-rng"
    wait_for(r"entropy: (ready|not available)|Zuse:|ok> ", 120)     # the block chain runs after POST
    t = text()
    v4 = re.search(r"System: StarForth v4", t)
    after = t[v4.end():] if v4 else ""
    check("System: StarForth v4", v4 is not None)
    be = re.search(r"rng: backend = (\S+)", after)
    check(f"after it, rng: backend = {want} (got {be.group(1) if be else None})",
          be is not None and be.group(1) == want)
    check("entropy: ready", re.search(r"entropy: ready", after) is not None)
    if a.target == "qemu":
        check("[zuse@Hera] ok>", wait_for(PROMPT, 120))
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m == "hotplug":
    check("no ok> for 20 s with no drive", not wait_for(ANYPROMPT, 20))
    add_zuse(1)
    check("identity confirmed on attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 60))
    n = count(PROMPT)
    type_line("3 4 + .")
    check("7  ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
    del_zuse(1)
    check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
    n = count(ANYPROMPT)
    check("no new ok> for 30 s after detach", not wait_for(ANYPROMPT, 30, n))
    c = count(r"Zuse: identity confirmed from attached thumbdrive")
    p = count(PROMPT)
    add_zuse(2)
    check("identity confirmed on re-attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120, c))
    check("[zuse@Hera] ok> again", wait_for(PROMPT, 60, p))

elif m == "midpull":
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    qcmd("device_del", id="zusedev1")
    type_line("3 4 + .")
    type_line("5 6 + .")
    check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
    time.sleep(10)
    t = text()
    check("the line typed after the pull did not run (no 11)", re.search(r"\b11\s+ok", t) is None)
    check("no ok> after the session ended", "ok> " not in t[t.rfind("Zuse: session ended"):])
    # the replug: identity and Artemis were both let go; Artemis is attached first again
    qcmd("blockdev-del", **{"node-name": "zuse1"})
    p = count(PROMPT)
    add_zuse(2)
    check("[zuse@Hera] ok> again after the replug", wait_for(PROMPT, 120, p))
    na = count(r"Artemis: the stick's partition attached")
    check(f"Artemis attached again: two Artemis lines in all ({na})", na == 2)

elif m == "waitpull":
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    type_line("5 AWAIT")
    time.sleep(5)
    del_zuse(1)
    check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
    check("the waiting line ended at logout (Interrupted)", wait_for(r"Interrupted", 30))
    p = count(PROMPT)
    add_zuse(2)
    check("identity confirmed on re-attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120, 1))
    check("[zuse@Hera] ok> again", wait_for(PROMPT, 60, p))
    n = count(PROMPT)
    type_line("3 4 + .")
    check("7  ok after re-attach", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m == "typeahead":
    check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
    type_line("3 4 + .")
    time.sleep(5)
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    time.sleep(5)
    check("what was typed while headless did not run (no 7)", re.search(r"\b7\s+ok", text()) is None)
    n = count(PROMPT)
    type_line("1 2 + .")
    check("3  ok typed at the prompt", wait_for(PROMPT, 60, n) and re.search(r"\b3\s+ok", text()) is not None)

elif m == "emergency":
    check("ok> with no drive", wait_for(ANYPROMPT, 60))
    n = count(ANYPROMPT)
    type_line("3 4 + .")
    check("7  ok", wait_for(ANYPROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m == "v3_keys":
    check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
    time.sleep(2)
    n = count(PROMPT)
    key_line("3 4 + .")
    check("7  ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m == "screen_dark_until_login":
    time.sleep(30)
    w, h, lit, _ = screen()
    check(f"the screen ({w}x{h}) is all black before login ({lit} lit)", lit == 0)

elif m == "screen_boot":
    check("Screen: the console stays on until login (screen=boot)",
          wait_for(r"Screen: the console stays on until login \(screen=boot\)", 60))
    time.sleep(30)
    w, h, lit, _ = screen()
    check(f"the screen ({w}x{h}) shows the console before login ({lit} lit)", lit >= 500)
    check("no ok> (no one is logged in)", not wait_for(ANYPROMPT, 5))
    check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)

elif m in ("screen_console_after_login", "screen_dark_after_logout"):
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    time.sleep(3)
    w, h, lit1, px1 = screen()
    check(f"the screen shows the console ({lit1} lit pixels)", lit1 >= 500)
    n = count(PROMPT)
    key_line("3 4 + .")
    check("7  ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
    time.sleep(3)
    _, _, lit2, px2 = screen()
    check("the screen changed with what was typed", px2 != px1)
    if m == "screen_dark_after_logout":
        del_zuse(1)
        check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
        time.sleep(5)
        _, _, lit3, _ = screen()
        check(f"the screen is black after logout ({lit3} lit)", lit3 == 0)

elif m == "kbd_typeahead":
    check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
    key_line("3 4 + .")
    time.sleep(5)
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    time.sleep(5)
    check("keys pressed while headless did not run (no 7)", re.search(r"\b7\s+ok", text()) is None)
    n = count(PROMPT)
    key_line("1 2 + .")
    check("3  ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"\b3\s+ok", text()) is not None)

elif m == "usb_kbd":
    check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
    time.sleep(2)
    n = count(PROMPT)
    key_line("3 4 + .")          # the USB keyboard is the only one (-machine i8042=off)
    check("7  ok typed on the USB keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m in ("no_framebuffer", "virtio_kbd"):
    check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
    time.sleep(2)
    n = count(PROMPT)
    if m == "no_framebuffer":
        type_line("3 4 + .")
    else:
        key_line("3 4 + .")
    check("7  ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)

elif m in ("alttab_after_logout", "alttab_before_pull"):
    add_zuse(1)
    check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
    n = count(PROMPT)
    key_line("3 4 + .")
    check("7  ok typed", wait_for(PROMPT, 60, n))
    if m == "alttab_before_pull":
        alt_tab()                                   # to the graphics view, her session on screen
    del_zuse(1)
    check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
    time.sleep(5)
    if m == "alttab_after_logout":
        alt_tab()
        alt_tab()
        time.sleep(3)
    _, _, lit, _ = screen()
    check(f"the screen is black after logout ({lit} lit)", lit == 0)
    if m == "alttab_before_pull":
        p = count(PROMPT)
        add_zuse(2)
        check("[zuse@Hera] ok> again", wait_for(PROMPT, 120, p))
        time.sleep(3)
        _, _, lit1, px1 = screen()
        check(f"the console is drawn again ({lit1} lit)", lit1 >= 500)
        n = count(PROMPT)
        key_line("1 2 + .")
        check("3  ok typed", wait_for(PROMPT, 60, n))
        time.sleep(3)
        _, _, _, px2 = screen()
        check("the screen changed with what was typed", px2 != px1)

elif m == "emergency_pull":
    check("ok> with no drive", wait_for(ANYPROMPT, 60))
    add_zuse(1)
    check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
    time.sleep(3)
    del_zuse(1)
    check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
    time.sleep(5)
    _, _, lit1, px1 = screen()
    check(f"the screen still shows the console ({lit1} lit)", lit1 >= 500)
    n = count(ANYPROMPT)
    key_line("3 4 + .")
    check("7  ok typed after the pull", wait_for(ANYPROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
    time.sleep(3)
    _, _, _, px2 = screen()
    check("the screen changed with what was typed", px2 != px1)

elif m in ("gpt_zuse", "gpt_hotplug", "gpt_foreign", "gpt_bad_crc", "gpt_bad_entries", "gpt_no_header",
           "gpt_read_error"):
    gimg = os.path.join(work, "gpt.img")
    make_gpt_image(gimg, GPT_LINUX if m == "gpt_foreign" else GPT_IDENTITY, ident,
                   break_crc=(m == "gpt_bad_crc"), break_entries=(m == "gpt_bad_entries"),
                   break_sig=(m == "gpt_no_header"))
    check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
    b0 = blk_lines()
    if m == "gpt_read_error":
        add_zuse(1, ident, fail_sector0=True)       # the raw Zuse drive: no GPT, but block 0 unreadable
    else:
        add_zuse(1, gimg)
    if m in ("gpt_zuse", "gpt_hotplug"):
        # a GPT disk with only an identity partition is not a stick: on v4 it is not used
        # (plan 2026-10-27-stick-4-one-disk.md, R2; Zuse's own case is the stick now)
        check("not this system's stick -- not used",
              wait_for(r"USB: a drive that is not this system's stick -- not used", 60))
        check("no ok> for 30 s", not wait_for(ANYPROMPT, 30))
        check(f"not attached (blk: lines {b0} -> {blk_lines()})", blk_lines() == b0)
    else:
        if m in ("gpt_bad_crc", "gpt_bad_entries", "gpt_no_header"):
            check("the damaged table is said so", wait_for(r"GPT: a USB drive has a damaged partition table -- not used", 60))
        if m == "gpt_read_error":
            check("the unreadable drive is said so", wait_for(r"GPT: a USB drive could not be read -- not used", 60))
        check("no ok> for 30 s", not wait_for(ANYPROMPT, 30))
        check("no identity", count(r"Zuse: identity confirmed") == 0)
        check(f"not attached (blk: lines {b0} -> {blk_lines()})", blk_lines() == b0)

else:
    check(f"known mode {m}", False)

print(f"RESULT: {'PASS' if ok else 'FAIL'}", flush=True)
if not globals().get("boot_done"):
    shutdown()
for f in ("v3/include/version.h", "capsules/BLOCK_MAP.md"):
    subprocess.run(["git", "checkout", "-q", "--", f], cwd=ROOT)
sys.exit(0 if ok else 2)