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, 232 insertions, 0 deletions

diff --git a/tests/smoke.py b/tests/smoke.py
new file mode 100644
index 0000000..2b9568c
--- /dev/null
+++ b/tests/smoke.py
@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+"""End-to-end smoke test for gdb-driver (Milestone 5).
+
+Spawns a real gdb+GEF pane on a unix socket via the pane manager, drives the
+whole method surface against a known fixture, and asserts on the *structured*
+results.  Doubles as a protocol regression lock (blueprint §12.6).
+
+Run:  python3 tests/smoke.py     (must be inside a tmux session)
+Exit: 0 = all checks passed, 1 = one or more failed.
+"""
+
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+sys.path.insert(0, ROOT)
+import pane  # noqa: E402
+
+FIXTURE_SRC = r"""
+#include <stdio.h>
+int add(int a, int b) {
+    int c = a + b;
+    return c;
+}
+int main(void) {
+    int x = add(2, 3);
+    printf("%d\n", x);
+    return 0;
+}
+"""
+
+# --------------------------------------------------------------------------
+# tiny assert framework
+# --------------------------------------------------------------------------
+
+_FAILS = []
+
+
+def check(name, cond, detail=""):
+    ok = bool(cond)
+    tag = "PASS" if ok else "FAIL"
+    line = f"[{tag}] {name}"
+    if not ok and detail:
+        line += f"  -- {detail}"
+    print(line, flush=True)
+    if not ok:
+        _FAILS.append(name)
+    return ok
+
+
+# --------------------------------------------------------------------------
+# rpc client
+# --------------------------------------------------------------------------
+
+class RpcError(Exception):
+    pass
+
+
+def rpc(sock, method, params=None, timeout=130):
+    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    s.settimeout(timeout)
+    s.connect(sock)
+    try:
+        f = s.makefile("rwb")
+        f.write((json.dumps({"id": 1, "method": method,
+                             "params": params or {}}) + "\n").encode())
+        f.flush()
+        line = f.readline()
+    finally:
+        s.close()
+    if not line:
+        raise RpcError("empty response")
+    msg = json.loads(line)
+    if msg.get("error"):
+        raise RpcError(msg["error"].get("message", "unknown"))
+    return msg.get("result")
+
+
+# --------------------------------------------------------------------------
+# the drive
+# --------------------------------------------------------------------------
+
+def run_checks(sock, fixture):
+    base = os.path.basename(fixture)
+
+    # -- lifecycle ------------------------------------------------------
+    p = rpc(sock, "ping")
+    check("ping ready", p.get("ready") is True and p.get("proto") == 1, p)
+
+    names = {m["name"] for m in rpc(sock, "methods")}
+    for want in ("ping", "state", "regs", "bt", "mem", "screen",
+                 "breakpoints"):
+        check(f"methods advertises {want}", want in names, names)
+
+    # -- pre-run state / not-ready errors -------------------------------
+    st = rpc(sock, "state")
+    check("state=no-inferior before run", st.get("status") == "no-inferior", st)
+
+    try:
+        rpc(sock, "regs")
+        check("regs errors before run", False, "no error raised")
+    except RpcError as e:
+        check("regs errors before run", "not ready" in str(e), str(e))
+
+    # -- breakpoints ----------------------------------------------------
+    r = rpc(sock, "break", {"loc": "add"})
+    check("break add output", "Breakpoint" in (r.get("output") or ""), r)
+
+    bps = rpc(sock, "breakpoints")
+    check("one breakpoint on add", len(bps) == 1 and bps[0]["loc"].endswith("add"),
+          bps)
+
+    # -- run -> settle at breakpoint (the crux) -------------------------
+    r = rpc(sock, "run")
+    st = r["state"]
+    check("run settles STOPPED", st.get("status") == "stopped", st)
+    check("run stopped in add", st.get("function") == "add", st)
+    check("run stopped in fixture file",
+          (st.get("file") or "").endswith(base.replace(base, "smoke_fixture.c")) or
+          (st.get("file") or "").endswith("smoke_fixture.c"), st)
+
+    # -- eval args ------------------------------------------------------
+    a = rpc(sock, "print", {"expr": "a"})
+    b = rpc(sock, "print", {"expr": "b"})
+    check("print a == 2", a["output"].endswith("0x2") or a["output"].endswith("2"),
+          a)
+    check("print b == 3", b["output"].endswith("0x3") or b["output"].endswith("3"),
+          b)
+
+    # -- reads ----------------------------------------------------------
+    regs = rpc(sock, "regs")
+    check("regs has rip", "rip" in regs, list(regs)[:5])
+    bt = rpc(sock, "bt")
+    check("bt: #0 add #1 main",
+          len(bt) >= 2 and bt[0]["func"] == "add" and bt[1]["func"] == "main",
+          bt)
+    m = rpc(sock, "mem", {"addr": "$sp", "len": 16})
+    check("mem returns 16 bytes hex", len(m.get("hex", "")) == 32, m)
+
+    # -- step: line must advance, then c becomes defined ----------------
+    line_before = rpc(sock, "state")["line"]
+    r = rpc(sock, "next")
+    line_after = r["state"]["line"]
+    check("next advances source line", line_after == line_before + 1,
+          f"{line_before}->{line_after}")
+    c = rpc(sock, "print", {"expr": "c"})
+    check("print c == 5 after next",
+          c["output"].endswith("0x5") or c["output"].endswith("5"), c)
+
+    # -- finish -> back in main ----------------------------------------
+    r = rpc(sock, "finish")
+    check("finish returns to main", r["state"].get("function") == "main",
+          r["state"])
+
+    # -- screen (viewer surface) ---------------------------------------
+    scr = rpc(sock, "screen")
+    text = scr.get("text", "")
+    check("screen returns text", len(text) > 0)
+    # capture-pane returns only the visible region; GEF's context fills it, so
+    # match on any GEF context marker a viewer sees (not necessarily the prompt)
+    markers = ("gef", "registers", "$rip", "$rsp", "\u2500", "code:")
+    check("screen shows GEF context",
+          any(m in text.lower() if m == "gef" else m in text for m in markers),
+          text[:80])
+
+    # -- continue -> program exits -------------------------------------
+    r = rpc(sock, "cont")
+    check("cont settles no-inferior (exit)",
+          r["state"].get("status") == "no-inferior", r["state"])
+
+    # -- concurrency: parallel pings never corrupt framing --------------
+    results = []
+
+    def hammer():
+        try:
+            results.append(("ok", rpc(sock, "ping", timeout=5)))
+        except RpcError as e:
+            results.append(("busy", str(e)))
+        except Exception as e:  # noqa: BLE001
+            results.append(("crash", str(e)))
+    ts = [threading.Thread(target=hammer) for _ in range(8)]
+    [t.start() for t in ts]
+    [t.join() for t in ts]
+    check("concurrent pings: no crashes/garbage",
+          all(k in ("ok", "busy") for k, _ in results), results)
+
+
+# --------------------------------------------------------------------------
+# main
+# --------------------------------------------------------------------------
+
+def main():
+    if "TMUX" not in os.environ:
+        print("smoke: must run inside a tmux session", file=sys.stderr)
+        return 2
+
+    tmp = tempfile.mkdtemp(prefix="gdbdrv-smoke-")
+    src = os.path.join(tmp, "smoke_fixture.c")
+    binp = os.path.join(tmp, "smoke_fixture")
+    with open(src, "w") as f:
+        f.write(FIXTURE_SRC)
+    subprocess.check_call(["gcc", "-g", "-O0", "-o", binp, src])
+
+    entry = None
+    try:
+        entry = pane.spawn(binp)
+        print(f"spawned pane={entry['pane']} sock={entry['sock']}\n")
+        run_checks(entry["sock"], binp)
+    finally:
+        if entry:
+            pane.stop(entry["sock"])
+        # verify no leak
+        leaked = any(e["sock"] == (entry or {}).get("sock")
+                     for e in pane.list_all())
+        check("teardown: pane deregistered/killed", not leaked)
+
+    print()
+    if _FAILS:
+        print(f"SMOKE FAILED: {len(_FAILS)} check(s): {', '.join(_FAILS)}")
+        return 1
+    print("SMOKE PASSED")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())