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

diff --git a/pane.py b/pane.py
new file mode 100644
index 0000000..ff8278d
--- /dev/null
+++ b/pane.py
@@ -0,0 +1,210 @@
+"""pane.py: tmux pane manager for gdb-driver (Milestone 4).
+
+Reusable core (target-independent shape): spawn a visible tmux pane running
+the backend, register it, poll until *ready*, and tear it down without leaking
+panes/sockets.  Only the spawn command + readiness check are gdb-specific.
+
+Registry lives at $XDG_RUNTIME_DIR/gdb-driver/panes.json, keyed by socket path.
+"""
+
+import json
+import os
+import shlex
+import socket
+import subprocess
+import time
+import uuid
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+DRIVER = os.path.join(HERE, "driver.py")
+
+
+def runtime_dir():
+    base = os.environ.get("XDG_RUNTIME_DIR") or f"/tmp/gdb-driver-{os.getuid()}"
+    d = os.path.join(base, "gdb-driver")
+    os.makedirs(d, mode=0o700, exist_ok=True)
+    return d
+
+
+def _reg_path():
+    return os.path.join(runtime_dir(), "panes.json")
+
+
+def _load():
+    try:
+        with open(_reg_path()) as f:
+            return json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {}
+
+
+def _save(reg):
+    tmp = _reg_path() + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(reg, f, indent=2)
+    os.replace(tmp, _reg_path())
+
+
+# --------------------------------------------------------------------------
+# liveness
+# --------------------------------------------------------------------------
+
+def _pane_alive(pane_id):
+    try:
+        out = subprocess.check_output(
+            ["tmux", "list-panes", "-a", "-F", "#{pane_id}"], text=True)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return False
+    return pane_id in out.split()
+
+
+def _rpc(sock, method, params=None, timeout=1.0):
+    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    s.settimeout(timeout)
+    try:
+        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()
+        return json.loads(line) if line else None
+    finally:
+        s.close()
+
+
+def sock_ready(sock, timeout=0.5):
+    """True iff the driver answers ping with ready=True."""
+    try:
+        r = _rpc(sock, "ping", timeout=timeout)
+    except OSError:
+        return False
+    return bool(r and r.get("result", {}).get("ready"))
+
+
+# --------------------------------------------------------------------------
+# spawn / stop / list
+# --------------------------------------------------------------------------
+
+def spawn(binary, args=None, split="-h", ready_timeout=15.0, env=None):
+    """Split a tmux pane running gdb+driver on a fresh socket; wait for ready.
+
+    `env` is a list of 'KEY=VALUE' strings exported for gdb AND inherited by the
+    inferior (e.g. LD_LIBRARY_PATH for shim libs, LD_PRELOAD).
+    Returns the registry entry dict {sock,pane,bin,args,env,created}.
+    """
+    if "TMUX" not in os.environ:
+        raise RuntimeError("not inside a tmux session")
+    binary = os.path.abspath(binary)
+    if not os.path.exists(binary):
+        raise FileNotFoundError(binary)
+    args = args or []
+    env = env or []
+    sock = os.path.join(runtime_dir(), f"gdb-{uuid.uuid4().hex[:8]}.sock")
+
+    env_tokens = [f"GDB_DRIVER_SOCK={shlex.quote(sock)}"]
+    env_tokens += [shlex.quote(e) for e in env]
+    inner = "env {envs} gdb -q -x {drv} --args {bin} {args}".format(
+        envs=" ".join(env_tokens),
+        drv=shlex.quote(DRIVER),
+        bin=shlex.quote(binary),
+        args=" ".join(shlex.quote(a) for a in args),
+    ).strip()
+
+    pane = subprocess.check_output(
+        ["tmux", "split-window", split, "-d", "-P", "-F", "#{pane_id}", inner],
+        text=True).strip()
+
+    entry = {"sock": sock, "pane": pane, "bin": binary, "args": args,
+             "env": env, "created": time.time()}
+    reg = _load()
+    reg[sock] = entry
+    _save(reg)
+
+    # readiness poll: socket up AND driver reports ready
+    deadline = time.time() + ready_timeout
+    while time.time() < deadline:
+        if sock_ready(sock):
+            return entry
+        if not _pane_alive(pane):
+            _deregister(sock)
+            raise RuntimeError(f"pane {pane} died before becoming ready")
+        time.sleep(0.15)
+    # timed out
+    stop(sock)
+    raise TimeoutError(f"gdb did not become ready in {ready_timeout}s")
+
+
+def _deregister(sock):
+    reg = _load()
+    reg.pop(sock, None)
+    _save(reg)
+    try:
+        os.unlink(sock)
+    except OSError:
+        pass
+
+
+def stop(sock):
+    """Graceful quit (best-effort) then kill the pane(s); deregister."""
+    reg = _load()
+    entry = reg.get(sock)
+    try:
+        _rpc(sock, "quit", timeout=2.0)
+    except OSError:
+        pass
+    if entry:
+        subprocess.run(["tmux", "kill-pane", "-t", entry["pane"]],
+                       stderr=subprocess.DEVNULL)
+    _deregister(sock)
+    return entry is not None
+
+
+# --------------------------------------------------------------------------
+# raw keys (client-side tmux; bypasses the RPC gate so it works even while a
+# blocking run-control call is in flight -- e.g. C-c to interrupt)
+# --------------------------------------------------------------------------
+
+def entry(sock):
+    return _load().get(sock)
+
+
+def send_keys(pane_id, tokens):
+    """Send named tmux key tokens (Enter, C-c, Escape, ...) to a pane."""
+    subprocess.run(["tmux", "send-keys", "-t", pane_id] + list(tokens),
+                   stderr=subprocess.DEVNULL)
+
+
+def list_all(prune=False):
+    """Return [{sock,pane,bin,alive,ready}]; optionally drop dead entries."""
+    reg = _load()
+    out = []
+    changed = False
+    for sock, e in list(reg.items()):
+        alive = _pane_alive(e["pane"])
+        ready = sock_ready(sock) if alive else False
+        if prune and not alive:
+            _deregister(sock)
+            changed = True
+            continue
+        out.append({"sock": sock, "pane": e["pane"], "bin": e["bin"],
+                    "alive": alive, "ready": ready})
+    if changed:
+        pass
+    return out
+
+
+def resolve(explicit=None):
+    """Pick the socket to drive: explicit > env > the single live one."""
+    if explicit:
+        return explicit
+    env = os.environ.get("GDB_DRIVER_SOCK")
+    if env:
+        return env
+    live = [e for e in list_all() if e["ready"]]
+    if len(live) == 1:
+        return live[0]["sock"]
+    if not live:
+        raise RuntimeError("no live gdb-driver pane (try: drive spawn <bin>)")
+    raise RuntimeError("multiple live panes; pass --sock:\n  " +
+                       "\n  ".join(e["sock"] for e in live))