refactor(kernel): the keyboard bridge, shared
The stick, increment 1, task 1 (plan 2026-10-26-stick-1-screen-and-keyboard). repl.c's keyboard bridge -- the key tables, Shift/Alt/Caps Lock/Num Lock and their lights, the keypad, Alt+TAB, and the per-tick USB keyboard servicing -- moves unchanged to kernel/src/hal/kbd_input.c as sk_input_getc() (and sk_input_drop()), so that the v4 boot's console reads the same keyboards (docs/v4.0.0/STICK.md 5.4). v3's behaviour is unchanged. scripts/qemu_session.py: the typed-QEMU-session harness, kept in the repo from here on (the identity gate's sessions, and v3_keys: system=v3 typed on QEMU's keyboard with send-key). Yardstick, system=v3 typed by keyboard: before, amd64 PS/2 134805 PASS, aarch64 virtio keyboard 135549 PASS; after, 135746 and 135835 PASS. (riscv64 134850: v3 never reached its prompt -- Artemis virtio_blk writes time out; v3's block path, before this change; being investigated.) Accepted: suite EXIT 0; three clean qemu boots to [zuse@Hera] ok>, POST 538/538, dict_hash 0xfec8c47cfc6f437f (145722, 145859, 150231); test-32, 39 modules, 0 failures. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
rajames
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
6f96502fdf
commit
83d3692976
Executable
+273
@@ -0,0 +1,273 @@
|
||||
#!/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.
|
||||
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
|
||||
|
||||
Needs only Python 3. One QEMU at a time (.claude/CLAUDE.md).
|
||||
"""
|
||||
import argparse, glob, json, os, re, 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")
|
||||
a = ap.parse_args()
|
||||
|
||||
NO_ZUSE_MODES = {"headless", "hotplug", "midpull", "waitpull", "typeahead", "emergency"}
|
||||
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")
|
||||
zimg = os.path.join(work, "zuse-thumb-ident.img")
|
||||
attach_zuse = not (a.no_zuse or a.mode in NO_ZUSE_MODES)
|
||||
|
||||
make_args = ["make", f"ARCH={a.arch}", "clean", "qemu", f"ARTDISK={art}",
|
||||
f"ZUSEDISK={zimg if attach_zuse else ''}"] + a.make.split()
|
||||
if a.mode == "v3_keys":
|
||||
make_args.append("KERNEL_ARGS=system=v3")
|
||||
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)
|
||||
|
||||
|
||||
def add_zuse(n):
|
||||
qcmd("blockdev-add", driver="raw", **{"node-name": f"zuse{n}"},
|
||||
file={"driver": "file", "filename": zimg})
|
||||
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("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 == "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"):])
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user