LithosAnanke/scripts/qemu_session.py

#!/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), with copies of the minted Zuse pair (disk/zuse-scratch/) made
under build/qemu-session/<label>/ so the committed images are never booted;
Zuse's drive is that pair laid out as a stick (zusestick.img).
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 minted pair: identity confirmed, [zuse@Hera] ok>, `3 4 + .` -> 7  ok
  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); headless
  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
  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)
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"}
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))
attach_zuse = not (a.no_zuse or a.mode in NO_ZUSE_MODES)

# 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, f"ARTDISK={art}",
             f"ZUSEDISK={zdrive if attach_zuse else ''}"] + shlex.split(a.make)
if a.mode in ("v3_keys", "stick_v3"):
    make_args.append("KERNEL_ARGS=system=v3")
if a.mode == "stick_zuse":
    make_args.append(f"STICK_PAIR={work}")          # the PC stick with this session's copy of the pair
start = time.time()
make_log = open(os.path.join(work, "make.txt"), "w")
mk = subprocess.Popen(make_args, cwd=ROOT, stdout=make_log, stderr=subprocess.STDOUT)
deadline = time.time() + 1800
ok = True


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


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)


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


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()


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


qcmd("qmp_capabilities")


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 check(name, cond):
    global ok
    print(f"{'PASS' if cond else 'FAIL'}: {name}", flush=True)
    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)
    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, 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)
        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: no genesis on this disk", wait_for(r"Zuse: no genesis on this disk", 60))
        time.sleep(5)
        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)
        check("no ok> for 30 s (a blank identity: headless)", not wait_for(ANYPROMPT, 30))
    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)
    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 == "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 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)
q.send(json.dumps({"execute": "quit"}).encode())
mk.wait()
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)