gdb-driver: puppeteer live gdb (plain CLI + GEF) over an RPC socketHEADmain

Embedded gdb-Python plugin exposing a JSONL unix-socket RPC: three tiers of
control (raw tmux keys / semantic verbs / structured introspection), real
stop-event settling, and gef>-prompt command echoing so the tmux pane stays
human-legible. Includes the drive/pane client CLIs, an end-to-end smoke test,
and the TUI-driving design blueprint.

1 files changed, 240 insertions, 0 deletions

diff --git a/drive b/drive
new file mode 100755
index 0000000..bea9bd9
--- /dev/null
+++ b/drive
@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+"""drive: ergonomic client for gdb-driver (Milestone 1).
+
+Auto-resolves the unix socket (single live pane), speaks JSONL, prints terse
+text.  Usage:
+    drive ping
+    drive methods
+    drive quit
+    drive raw <method> [k=v ...]
+    drive --sock /path/to.sock ping
+"""
+
+import json
+import os
+import socket
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import pane  # noqa: E402
+
+PROTO = 1
+
+
+def resolve_sock(explicit):
+    try:
+        return pane.resolve(explicit)
+    except RuntimeError as e:
+        die(str(e))
+
+
+def die(msg, code=1):
+    sys.stderr.write(f"drive: {msg}\n")
+    sys.exit(code)
+
+
+def fmt_state(r):
+    if not r:
+        return "?"
+    st = r.get("status")
+    if st == "stopped":
+        loc = r.get("loc") or (f"{r['file']}:{r['line']}"
+                               if r.get("file") else "?")
+        return (f"{r.get('function')} @ {r['pc']} at {loc} "
+                f"[stopped] thr#{r.get('thread')}")
+    return st
+
+
+def call(sock, method, params=None, timeout=35):
+    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    s.settimeout(timeout)
+    s.connect(sock)
+    f = s.makefile("rwb")
+    f.write((json.dumps({"id": 1, "method": method,
+                         "params": params or {}}) + "\n").encode())
+    f.flush()
+    line = f.readline()
+    s.close()
+    if not line:
+        die("no response (connection closed)")
+    msg = json.loads(line)
+    if "error" in msg and msg["error"]:
+        die(msg["error"].get("message", "unknown error"))
+    return msg.get("result")
+
+
+def parse_kv(args):
+    params = {}
+    for a in args:
+        if "=" not in a:
+            die(f"expected k=v, got {a!r}")
+        k, v = a.split("=", 1)
+        try:
+            params[k] = json.loads(v)
+        except json.JSONDecodeError:
+            params[k] = v
+    return params
+
+
+SEMANTIC = {"run", "start", "cont", "continue", "c", "next", "n", "step", "s",
+            "stepi", "si", "nexti", "ni", "finish", "until", "break", "b",
+            "tbreak", "delete", "watch", "print", "p", "x", "set", "frame",
+            "up", "down"}
+
+_ALIAS = {"c": "cont", "continue": "cont", "n": "next", "s": "step",
+          "si": "stepi", "ni": "nexti", "b": "break", "p": "print"}
+
+
+def build_semantic(cmd, rest):
+    cmd = _ALIAS.get(cmd, cmd)
+    if cmd == "run":
+        return "run", ({"args": " ".join(rest)} if rest else {})
+    if cmd in ("next", "step", "stepi", "nexti", "up", "down"):
+        return cmd, ({"n": int(rest[0])} if rest else {})
+    if cmd in ("start", "finish", "cont"):
+        return cmd, {}
+    if cmd == "until":
+        return "until", ({"loc": rest[0]} if rest else {})
+    if cmd in ("break", "tbreak"):
+        return cmd, {"loc": " ".join(rest)}
+    if cmd == "watch":
+        return "watch", {"expr": " ".join(rest)}
+    if cmd == "delete":
+        return "delete", ({"n": int(rest[0])} if rest else {})
+    if cmd == "print":
+        return "print", {"expr": " ".join(rest)}
+    if cmd == "x":
+        return "x", {"fmt": rest[0], "addr": " ".join(rest[1:])}
+    if cmd == "set":
+        return "set", {"expr": " ".join(rest)}
+    if cmd == "frame":
+        return "frame", {"n": int(rest[0])}
+    die(f"cannot build semantic verb: {cmd}")
+
+
+def main(argv):
+    # --sock may appear anywhere (global option)
+    sock = None
+    args = []
+    it = iter(argv)
+    for a in it:
+        if a == "--sock":
+            sock = next(it, None)
+        else:
+            args.append(a)
+    if not args:
+        die("usage: drive <cmd> [args]   (try: drive ping)")
+    cmd, rest = args[0], args[1:]
+
+    # --- pane lifecycle (don't need a resolved socket first) ----------
+    if cmd == "spawn":
+        envs, positional = [], []
+        it = iter(rest)
+        for a in it:
+            if a == "--env":
+                envs.append(next(it, ""))
+            elif a.startswith("--env="):
+                envs.append(a[len("--env="):])
+            else:
+                positional.append(a)
+        if not positional:
+            die("usage: drive spawn [--env K=V ...] <binary> [prog-args...]")
+        try:
+            e = pane.spawn(positional[0], positional[1:], env=envs)
+        except Exception as ex:  # noqa: BLE001
+            die(str(ex))
+        print(f"ready  pane={e['pane']}  sock={e['sock']}\n"
+              f"bin={e['bin']} {' '.join(e['args'])}".rstrip())
+        return
+    if cmd == "list":
+        prune = "--prune" in rest
+        entries = pane.list_all(prune=prune)
+        if not entries:
+            print("(no panes)")
+        for e in entries:
+            flag = "ready" if e["ready"] else ("alive" if e["alive"]
+                                               else "dead")
+            print(f"{flag:<5} {e['pane']:<6} {os.path.basename(e['bin'])}"
+                  f"  {e['sock']}")
+        return
+    if cmd == "stop":
+        target = sock or (rest[0] if rest else None)
+        try:
+            target = pane.resolve(target)
+        except RuntimeError as ex:
+            die(str(ex))
+        print("stopped" if pane.stop(target) else "stopped (untracked)")
+        return
+
+    sock = resolve_sock(sock)
+
+    # --- raw keys (client-side tmux; works during a blocking cont) -----
+    if cmd == "keys":
+        if not rest:
+            die("usage: drive keys <tmux-key ...>   e.g. drive keys C-c")
+        e = pane.entry(sock)
+        if not e:
+            die("no registry entry for socket")
+        pane.send_keys(e["pane"], rest)
+        print(f"sent {len(rest)} key(s) to gdb pane {e['pane']}")
+        return
+
+    if cmd == "ping":
+        r = call(sock, "ping")
+        print(f"ok proto={r['proto']} ready={r['ready']} pid={r['pid']}")
+    elif cmd == "methods":
+        for m in call(sock, "methods"):
+            print(f"{m['name']:<12} [{m['tier']}] {m['desc']}")
+    elif cmd == "quit":
+        r = call(sock, "quit")
+        print("bye" if r.get("bye") else json.dumps(r))
+    elif cmd in ("state", "where"):
+        print(fmt_state(call(sock, "state")))
+    elif cmd in ("breakpoints", "bp", "info-break"):
+        for b in call(sock, "breakpoints"):
+            loc = b.get("loc") or b.get("expr") or "?"
+            t = "t" if b.get("temporary") else " "
+            print(f"#{b['num']}{t} {loc}  hits={b['hits']} "
+                  f"{'on' if b['enabled'] else 'off'}")
+    elif cmd == "cmd":
+        if not rest:
+            die("usage: drive cmd <gdb command ...>")
+        r = call(sock, "cmd", {"line": " ".join(rest)}, timeout=130)
+        if r.get("output"):
+            print(r["output"])
+    elif cmd in SEMANTIC:
+        method, params = build_semantic(cmd, rest)
+        r = call(sock, method, params, timeout=130)
+        if r.get("output"):
+            print(r["output"])
+        print(fmt_state(r.get("state")))
+    elif cmd == "regs":
+        for k, v in call(sock, "regs").items():
+            print(f"{k:<8} {v}")
+    elif cmd == "bt":
+        params = {"n": int(rest[0])} if rest else {}
+        for fr in call(sock, "bt", params):
+            loc = fr.get("loc") or (f"{fr['file']}:{fr['line']}"
+                                    if fr.get("file") else "?")
+            print(f"#{fr['level']:<2} {fr['pc']} {fr.get('func')} at {loc}")
+    elif cmd == "mem":
+        if not rest:
+            die("usage: drive mem <addr-expr> [len]")
+        params = {"addr": rest[0]}
+        if len(rest) > 1:
+            params["len"] = int(rest[1])
+        print(call(sock, "mem", params)["dump"])
+    elif cmd == "screen":
+        color = bool(rest and rest[0] in ("-c", "color", "1"))
+        sys.stdout.write(call(sock, "screen", {"color": color})["text"])
+    elif cmd == "raw":
+        if not rest:
+            die("usage: drive raw <method> [k=v ...]")
+        r = call(sock, rest[0], parse_kv(rest[1:]))
+        print(json.dumps(r, indent=2))
+    else:
+        die(f"unknown command: {cmd}")
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])