control: socket-based debug channel + gbctl CLI driver

Replace the input-only FIFO's limitations with a Unix-domain socket control
channel (--sock) that is a superset of the button vocabulary plus emulator
introspection: read/write bus memory, get/set CPU context, single-step, PC
breakpoints, value-change watchpoints, run/pause. Being a socket it replies to
each command and broadcasts async 'event stop ...' lines; a stored last-stop
record (stopinfo) lets per-request clients recover a missed event.

Add gbctl: a stdlib-python CLI that spawns a tmux pane running the emulator on
a ROM, auto-resolves the socket, and drives it with terse verbs (cpu/read/write/
reg/step/break/watch/continue/pause/press/hold/screen/stop). The FIFO stays as
legacy input-only.

1 files changed, 362 insertions, 0 deletions

diff --git a/gbctl b/gbctl
new file mode 100755
index 0000000..7eb521c
--- /dev/null
+++ b/gbctl
@@ -0,0 +1,362 @@
+#!/usr/bin/env python3
+"""gbctl — drive the gbc emulator's socket control/debug channel from the CLI.
+
+Model mirrors gdb-driver: spawn a tmux pane running the emulator with a control
+socket, then drive it with terse verbs, then stop. Socket auto-resolves to the
+single live instance (else pass --sock / set GBCTL_SOCK).
+
+  gbctl spawn <rom> [emu-args...]     # tmux pane + emulator --sock; waits ready
+  gbctl cpu | state | stopinfo        # introspection (one-line replies)
+  gbctl read 0xC000 16                # hex dump
+  gbctl write 0xC000 de ad be ef      # hex bytes (round-trips with read)
+  gbctl reg pc 0x150                  # set a register/flag
+  gbctl step [n]                      # single-step (synchronous)
+  gbctl break 0x150 | break | delete all
+  gbctl watch 0xFF44 | unwatch all
+  gbctl continue [secs]              # resume; if secs given, block for a stop event
+  gbctl pause
+  gbctl press a b   |  gbctl hold left  |  gbctl release   # inject input
+  gbctl send <raw protocol line...>   # escape hatch: send any line verbatim
+  gbctl monitor [secs]                # stream async events (default until Ctrl-C)
+  gbctl screen [color]                # tmux capture of the emulator pane
+  gbctl list [--prune]                # inventory of live instances
+  gbctl stop                          # quit emulator + kill pane + rm socket
+"""
+import json, os, re, socket, subprocess, sys, time, glob
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+BIN  = os.path.join(HERE, "build", "gbc")
+REGDIR = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "gbctl")
+
+# ---------------------------------------------------------------------------
+# registry
+# ---------------------------------------------------------------------------
+def _ensure_regdir():
+    os.makedirs(REGDIR, exist_ok=True)
+
+def _reg_path(inst_id):
+    return os.path.join(REGDIR, inst_id + ".json")
+
+def _load_all():
+    out = []
+    for p in glob.glob(os.path.join(REGDIR, "*.json")):
+        try:
+            with open(p) as f:
+                d = json.load(f)
+            d["_path"] = p
+            out.append(d)
+        except Exception:
+            pass
+    return out
+
+def _pane_alive(pane):
+    if not pane:
+        return False
+    try:
+        r = subprocess.run(["tmux", "list-panes", "-a", "-F", "#{pane_id}"],
+                           capture_output=True, text=True)
+        return pane in r.stdout.split()
+    except Exception:
+        return False
+
+def _live_instances(prune=False):
+    live = []
+    for d in _load_all():
+        ok = os.path.exists(d.get("sock", "")) and _pane_alive(d.get("pane"))
+        if ok:
+            live.append(d)
+        elif prune:
+            try:
+                os.unlink(d["_path"])
+                if d.get("sock") and os.path.exists(d["sock"]):
+                    os.unlink(d["sock"])
+            except Exception:
+                pass
+    return live
+
+# ---------------------------------------------------------------------------
+# socket resolution + I/O
+# ---------------------------------------------------------------------------
+def _resolve_sock(argv):
+    # explicit --sock anywhere
+    if "--sock" in argv:
+        i = argv.index("--sock")
+        s = argv[i + 1]
+        del argv[i:i + 2]
+        return s
+    if os.environ.get("GBCTL_SOCK"):
+        return os.environ["GBCTL_SOCK"]
+    live = _live_instances()
+    if len(live) == 1:
+        return live[0]["sock"]
+    if not live:
+        die("no live emulator; run: gbctl spawn <rom>")
+    die("multiple instances; pass --sock <path>:\n" +
+        "\n".join("  %s  %s" % (d["sock"], d.get("rom", "")) for d in live))
+
+class Conn:
+    """Line-buffered wrapper around a connected unix socket."""
+    def __init__(self, sock_path, timeout=5.0):
+        self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.s.settimeout(timeout)
+        self.s.connect(sock_path)
+        self.buf = b""
+
+    def read_line(self, timeout=None):
+        if timeout is not None:
+            self.s.settimeout(timeout)
+        while b"\n" not in self.buf:
+            try:
+                d = self.s.recv(65536)
+            except socket.timeout:
+                return None
+            if not d:
+                if self.buf:
+                    line, self.buf = self.buf, b""
+                    return line.decode(errors="replace")
+                return None
+            self.buf += d
+        line, _, self.buf = self.buf.partition(b"\n")
+        return line.decode(errors="replace")
+
+    def send(self, line):
+        self.s.sendall((line + "\n").encode())
+
+    def close(self):
+        try:
+            self.s.close()
+        except Exception:
+            pass
+
+def _connect(sock_path, timeout=5.0):
+    c = Conn(sock_path, timeout)
+    c.read_line()          # consume greeting
+    return c
+
+def send_cmd(sock_path, line, wait_event=0.0):
+    """Send one protocol line; print the reply. If wait_event>0, keep reading
+    for an async 'event ...' line up to that many seconds and print it too."""
+    c = _connect(sock_path)
+    c.send(line)
+    reply = c.read_line(timeout=5.0)
+    if reply is not None:
+        print(reply)
+    if wait_event > 0:
+        deadline = time.time() + wait_event
+        while time.time() < deadline:
+            ev = c.read_line(timeout=max(0.01, deadline - time.time()))
+            if ev is None:
+                break
+            if ev.startswith("event"):
+                print(ev)
+                break
+    c.close()
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+def die(msg, code=1):
+    print(msg, file=sys.stderr)
+    sys.exit(code)
+
+def _wait_ready(sock_path, secs=10.0):
+    deadline = time.time() + secs
+    while time.time() < deadline:
+        if os.path.exists(sock_path):
+            try:
+                c = _connect(sock_path, timeout=1.0)
+                c.send("ping")
+                ok = (c.read_line(1.0) or "").strip() == "pong"
+                c.close()
+                if ok:
+                    return True
+            except Exception:
+                pass
+        time.sleep(0.1)
+    return False
+
+# ---------------------------------------------------------------------------
+# subcommands
+# ---------------------------------------------------------------------------
+def cmd_spawn(argv):
+    if not os.path.exists(BIN):
+        die("emulator not built: run `make` in %s" % HERE)
+    split = "-v"                              # stacked keeps full width for render
+    while argv and argv[0].startswith("--split"):
+        val = argv.pop(0).split("=", 1)
+        mode = (val[1] if len(val) == 2 else argv.pop(0)) if val[0] == "--split" else "v"
+        split = {"h": "-h", "v": "-v", "window": "window"}.get(mode, "-v")
+    if not argv:
+        die("usage: gbctl spawn [--split h|v|window] <rom> [emu-args...]")
+    rom = argv[0]
+    emu_args = argv[1:]
+    if not os.path.exists(rom):
+        die("rom not found: %s" % rom)
+
+    _ensure_regdir()
+    inst_id = "%d-%s" % (os.getpid(), os.urandom(3).hex())
+    sock_path = os.path.join(REGDIR, inst_id + ".sock")
+
+    if "TMUX" not in os.environ:
+        die("not inside tmux; gbctl spawn needs a tmux session")
+
+    cmd = [BIN, rom, "--sock", sock_path] + emu_args
+    # exec via a shell wrapper so the pane stays around & shows a title
+    shell_cmd = "printf '\\033]2;gbc %s\\007'; exec %s" % (
+        os.path.basename(rom), " ".join(_shquote(c) for c in cmd))
+    if split == "window":
+        r = subprocess.run(["tmux", "new-window", "-P", "-F", "#{pane_id}",
+                            shell_cmd], capture_output=True, text=True)
+    else:
+        r = subprocess.run(["tmux", "split-window", split, "-P", "-F",
+                            "#{pane_id}", shell_cmd], capture_output=True, text=True)
+    if r.returncode != 0:
+        die("tmux spawn failed: %s" % r.stderr.strip())
+    pane = r.stdout.strip()
+
+    with open(_reg_path(inst_id), "w") as f:
+        json.dump({"id": inst_id, "sock": sock_path, "pane": pane,
+                   "rom": rom, "created": time.time()}, f)
+
+    if _wait_ready(sock_path):
+        print("ready pane=%s sock=%s rom=%s" % (pane, sock_path, rom))
+    else:
+        die("spawned pane=%s but control socket never became ready\n"
+            "  check the pane for errors (gbctl screen)" % pane)
+
+def _shquote(s):
+    if re.fullmatch(r"[A-Za-z0-9_./=:+-]+", s or ""):
+        return s
+    return "'" + s.replace("'", "'\\''") + "'"
+
+def cmd_list(argv):
+    prune = "--prune" in argv
+    live = _live_instances(prune=prune)
+    if not live:
+        print("(no live instances)")
+        return
+    for d in live:
+        print("pane=%s sock=%s rom=%s" % (d.get("pane"), d["sock"],
+                                          d.get("rom", "")))
+
+def cmd_stop(argv):
+    sock_path = _resolve_sock(argv)
+    # find matching registry entry (for pane id)
+    inst = None
+    for d in _load_all():
+        if d.get("sock") == sock_path:
+            inst = d
+            break
+    # ask the emulator to quit
+    try:
+        c = _connect(sock_path, timeout=2.0)
+        c.send("quit")
+        c.read_line(1.0)
+        c.close()
+    except Exception:
+        pass
+    time.sleep(0.2)
+    if inst:
+        if _pane_alive(inst.get("pane")):
+            subprocess.run(["tmux", "kill-pane", "-t", inst["pane"]],
+                          capture_output=True)
+        try:
+            os.unlink(inst["_path"])
+        except Exception:
+            pass
+    if os.path.exists(sock_path):
+        try:
+            os.unlink(sock_path)
+        except Exception:
+            pass
+    print("stopped %s" % sock_path)
+
+def cmd_screen(argv):
+    color = "color" in argv
+    argv = [a for a in argv if a != "color"]
+    sock_path = _resolve_sock(argv)
+    inst = next((d for d in _load_all() if d.get("sock") == sock_path), None)
+    if not inst or not inst.get("pane"):
+        die("no pane for %s" % sock_path)
+    args = ["tmux", "capture-pane", "-p", "-t", inst["pane"]]
+    if color:
+        args.insert(2, "-e")
+    r = subprocess.run(args, capture_output=True, text=True)
+    sys.stdout.write(r.stdout)
+
+def cmd_monitor(argv):
+    secs = float(argv[0]) if argv and _isnum(argv[0]) else None
+    if secs is not None:
+        argv = argv[1:]
+    sock_path = _resolve_sock(argv)
+    c = _connect(sock_path, timeout=None)
+    print("# monitoring %s (Ctrl-C to stop)" % sock_path, file=sys.stderr)
+    deadline = (time.time() + secs) if secs else None
+    try:
+        while True:
+            if deadline and time.time() >= deadline:
+                break
+            to = None if deadline is None else max(0.01, deadline - time.time())
+            line = c.read_line(timeout=to)
+            if line is None:
+                if deadline:
+                    break
+                continue
+            print(line, flush=True)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        c.close()
+
+def _isnum(x):
+    try:
+        float(x)
+        return True
+    except Exception:
+        return False
+
+# ---------------------------------------------------------------------------
+# main dispatch
+# ---------------------------------------------------------------------------
+LIFECYCLE = {"spawn": cmd_spawn, "list": cmd_list, "stop": cmd_stop,
+             "screen": cmd_screen, "monitor": cmd_monitor}
+
+# convenience aliases -> raw protocol verbs
+ALIASES = {"press": None, "tap": None, "hold": None, "buttons": None}
+
+def main():
+    argv = sys.argv[1:]
+    if not argv or argv[0] in ("-h", "--help", "help") and len(argv) == 1:
+        print(__doc__)
+        return
+    verb = argv[0]
+
+    if verb in LIFECYCLE:
+        return LIFECYCLE[verb](argv[1:])
+
+    # everything else is a protocol line sent to the resolved socket
+    rest = argv[1:]
+    sock_path = _resolve_sock(rest)   # strips --sock if present
+
+    # input convenience: press/tap/hold map to button tokens
+    if verb in ("press", "tap"):
+        line = " ".join(rest)                       # e.g. "a b"
+    elif verb == "hold":
+        line = " ".join("+" + t for t in rest)      # +left +a
+    elif verb == "send":
+        line = " ".join(rest)                       # raw passthrough
+    else:
+        line = " ".join([verb] + rest)              # verb + args
+
+    # continue with an optional wait-for-event timeout (positional seconds)
+    wait = 0.0
+    if verb in ("continue", "cont", "c", "run"):
+        # a trailing numeric arg means "block this long for a stop event"
+        if rest and _isnum(rest[-1]):
+            wait = float(rest[-1])
+            line = " ".join([verb] + rest[:-1])
+
+    send_cmd(sock_path, line, wait_event=wait)
+
+if __name__ == "__main__":
+    main()