diff options
Diffstat (limited to '')
| -rwxr-xr-x | drive | 240 | ||||
| -rw-r--r-- | driver.py | 564 |
2 files changed, 804 insertions, 0 deletions
@@ -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:]) diff --git a/driver.py b/driver.py new file mode 100644 index 0000000..0492393 --- /dev/null +++ b/driver.py @@ -0,0 +1,564 @@ +"""gdb-driver: embedded RPC plugin (Milestone 1 skeleton). + +Load inside gdb: gdb -q -x driver.py --args ./prog ... + +Architecture (see TUI_DRIVING_BLUEPRINT.md): + - A background thread owns a unix-socket JSONL server. + - gdb's Python API is MAIN-THREAD-ONLY, so any handler that touches gdb + state is marshalled onto gdb's thread via gdb.post_event() and we block + on a threading.Event for the result. (Proven: post_event fires while gdb + sits idle at the prompt.) + - Wire protocol: {"id":N,"method":str,"params":{}}\\n + -> {"id":N,"result":...} | {"id":N,"error":{"message":...}} + +Milestone 1 surface: ping / methods / quit (lifecycle only). +""" + +import gdb +import json +import os +import socket +import stat +import subprocess +import sys +import threading +import traceback + +PROTO = 1 +MAIN_THREAD_TIMEOUT = 30.0 # seconds to wait for a main-thread callback + +# -------------------------------------------------------------------------- +# socket path resolution +# -------------------------------------------------------------------------- + +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 _sock_path(): + p = os.environ.get("GDB_DRIVER_SOCK") + if p: + return p + return os.path.join(_runtime_dir(), f"{os.getpid()}.sock") + + +# -------------------------------------------------------------------------- +# main-thread marshalling +# -------------------------------------------------------------------------- + +def on_main(fn, timeout=MAIN_THREAD_TIMEOUT): + """Run fn() on gdb's main thread; return its value or raise its exception.""" + box = {} + done = threading.Event() + + def cb(): + try: + box["ok"] = fn() + except Exception as e: # noqa: BLE001 - propagate to caller + box["err"] = e + finally: + done.set() + + gdb.post_event(cb) + if not done.wait(timeout): + raise TimeoutError(f"main-thread callback timed out after {timeout}s") + if "err" in box: + raise box["err"] + return box.get("ok") + + +# -------------------------------------------------------------------------- +# adapter (grows per milestone; M1 = lifecycle) +# -------------------------------------------------------------------------- + +# gdb prints exactly ONE prompt at startup (before the interactive loop) and +# does NOT reprint after a programmatic gdb.execute(). So: consume that one +# pending prompt for the first driven command, then render our own prompt for +# every command after -- giving a consistent 'gef> cmd' line every time. +_PROMPT_PENDING = True + + +def _current_prompt(): + """The live (GEF-colored) prompt string, matching what a human sees.""" + try: + hook = getattr(gdb, "prompt_hook", None) + if hook: + p = hook(lambda: "") + if p: + return p + except Exception: # noqa: BLE001 + pass + try: + return gdb.parameter("prompt") or "(gdb) " + except Exception: # noqa: BLE001 + return "(gdb) " + + +def echo_cmd(cmdline): + """Make a programmatically-driven command look hand-typed in the pane.""" + global _PROMPT_PENDING + try: + prompt = "" if _PROMPT_PENDING else _current_prompt() + _PROMPT_PENDING = False + gdb.write(prompt + cmdline + "\n", gdb.STDOUT) + gdb.flush() + except Exception: # noqa: BLE001 - echoing must never break driving + pass + + +def _sal_of(frame): + """(file, line) for a frame, or (None, None).""" + try: + sal = frame.find_sal() + if sal and sal.symtab: + return sal.symtab.filename, sal.line + except gdb.error: + pass + return None, None + + +def _sym_for(pc): + """'func+0xoff' for a pc via the (minimal) symbol table, else None. + Works for no-debug binaries where find_sal() has no file/line.""" + try: + out = gdb.execute(f"info symbol {pc:#x}", to_string=True).strip() + except gdb.error: + return None + if not out or out.startswith("No symbol"): + return None + # 'add + 10 in section .text of /path' -> 'add+0x10' + part = out.split(" in section")[0].strip() + if " + " in part: + name, off = part.split(" + ", 1) + try: + return f"{name}+{int(off):#x}" + except ValueError: + return part.replace(" + ", "+") + return part + + +def _require_frame(): + """Return the selected frame or raise a clean 'not ready' error.""" + try: + return gdb.selected_frame() + except gdb.error: + raise RuntimeError("not ready: no stack frame (inferior not stopped)") + + +class GdbAdapter: + def __init__(self): + self.ready = False + self.pane = os.environ.get("TMUX_PANE") + + # --- method table -------------------------------------------------- + def methods(self): + return [ + {"name": "ping", "tier": "lifecycle", + "desc": "liveness; {ok,proto,ready,pid}"}, + {"name": "methods", "tier": "lifecycle", + "desc": "this verb table (self-documentation)"}, + {"name": "quit", "tier": "lifecycle", + "desc": "reply, then gdb quit"}, + {"name": "state", "tier": "read", + "desc": "{status,pc,function,file,line,thread}"}, + {"name": "regs", "tier": "read", + "desc": "general-purpose registers {name:hex}"}, + {"name": "bt", "tier": "read", + "desc": "backtrace [{level,pc,func,file,line}]; param n"}, + {"name": "mem", "tier": "read", + "desc": "read memory; params addr(expr), len, [fmt]"}, + {"name": "screen", "tier": "read", + "desc": "tmux capture-pane of the gdb pane; param color"}, + {"name": "breakpoints", "tier": "read", + "desc": "list breakpoints [{num,type,loc,enabled,hits}]"}, + {"name": "run/start/cont/next/step/finish/until", + "tier": "semantic", "desc": "run-control; settles at next stop"}, + {"name": "break/tbreak/delete/watch", "tier": "semantic", + "desc": "breakpoint mgmt"}, + {"name": "print/x/set/frame/up/down", "tier": "semantic", + "desc": "eval / examine / assign / frame nav"}, + {"name": "cmd", "tier": "semantic", + "desc": "run ANY gdb/GEF command; param line. escape hatch " + "(use run-control verbs for continue/step -- cmd does " + "not settle on stop)"}, + ] + + # --- lifecycle ----------------------------------------------------- + def ping(self): + return {"ok": True, "proto": PROTO, "ready": self.ready, + "pid": os.getpid()} + + # --- read tier (all run on gdb's main thread) ---------------------- + def state(self): + inf = gdb.selected_inferior() + if inf is None or not inf.is_valid() or inf.pid == 0: + return {"status": "no-inferior"} + try: + thr = gdb.selected_thread() + tnum = thr.num if thr else None + except gdb.error: + tnum = None + try: + fr = gdb.selected_frame() + except gdb.error: + return {"status": "running", "thread": tnum} + pc = int(fr.pc()) + f, ln = _sal_of(fr) + loc = f"{f}:{ln}" if f else (_sym_for(pc) or hex(pc)) + return {"status": "stopped", "pc": hex(pc), + "function": fr.name(), "file": f, "line": ln, "loc": loc, + "thread": tnum} + + def regs(self): + fr = _require_frame() + arch = fr.architecture() + out = {} + for rd in arch.registers("general"): + try: + val = fr.read_register(rd) + # mask to the register's OWN width, not a hardcoded 64 bits -- + # otherwise a 32-bit reg with the high bit set sign-extends to + # a bogus 0xffffffffXXXXXXXX value. + size = val.type.sizeof or 8 + out[rd.name] = hex(int(val) & ((1 << (8 * size)) - 1)) + except (gdb.error, ValueError, gdb.MemoryError): + out[rd.name] = str(val) + return out + + def bt(self, n=None): + _require_frame() + frames = [] + fr = gdb.newest_frame() + lvl = 0 + while fr is not None and (n is None or lvl < n): + pc = int(fr.pc()) + f, ln = _sal_of(fr) + loc = f"{f}:{ln}" if f else (_sym_for(pc) or hex(pc)) + frames.append({"level": lvl, "pc": hex(pc), + "func": fr.name(), "file": f, "line": ln, + "loc": loc}) + try: + fr = fr.older() + except gdb.error: + break + lvl += 1 + return frames + + def mem(self, addr, len=64, fmt="xb"): + inf = gdb.selected_inferior() + if inf is None or inf.pid == 0: + raise RuntimeError("not ready: no inferior") + base = int(gdb.parse_and_eval(str(addr))) + length = int(len) + raw = bytes(inf.read_memory(base, length)) + lines = [] + for off in range(0, length, 16): + chunk = raw[off:off + 16] + hexs = " ".join(f"{b:02x}" for b in chunk) + asc = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) + lines.append(f"{base + off:#018x} {hexs:<47} {asc}") + return {"addr": hex(base), "len": length, + "hex": raw.hex(), "dump": "\n".join(lines)} + + def breakpoints(self): + out = [] + for b in gdb.breakpoints(): + out.append({"num": b.number, "type": b.type, + "loc": b.location, "expr": b.expression, + "enabled": b.enabled, "hits": b.hit_count, + "temporary": b.temporary}) + return out + + # --- mutating tier ------------------------------------------------- + # run-control: execute WITHOUT to_string so GEF's context renders in the + # viewer pane; gdb.execute blocks until the next stop (sync all-stop mode) + # => settled the moment it returns. Return fresh structured state. + def _run(self, cmdline): + echo_cmd(cmdline) + gdb.execute(cmdline, to_string=False) + return {"cmd": cmdline, "state": self.state()} + + # quick command: capture output for the agent AND replay it into the pane + # so the livestream shows both the command and its result. + def _cmd(self, cmdline): + echo_cmd(cmdline) + out = gdb.execute(cmdline, to_string=True) + if out: + gdb.write(out if out.endswith("\n") else out + "\n") + gdb.flush() + return {"cmd": cmdline, "output": out.rstrip("\n"), + "state": self.state()} + + # NB: run-control verbs are NOT here -- they need stop-event settle and + # are orchestrated from the socket thread (see run_control_dispatch). + def brk(self, loc): + return self._cmd(f"break {loc}") + + def tbreak(self, loc): + return self._cmd(f"tbreak {loc}") + + def delete(self, n=None): + return self._cmd("delete" + (f" {n}" if n is not None else "")) + + def watch(self, expr): + return self._cmd(f"watch {expr}") + + def eval(self, expr): + return self._cmd(f"print {expr}") + + def examine(self, fmt, addr): + return self._cmd(f"x/{fmt} {addr}") + + def setvar(self, expr): + return self._cmd(f"set var {expr}") + + def cmd(self, line): + return self._cmd(line) + + def frame(self, n): + return self._run(f"frame {int(n)}") + + def up(self, n=1): + return self._run(f"up {int(n)}") + + def down(self, n=1): + return self._run(f"down {int(n)}") + + # --- screen (does NOT touch gdb state; runs on socket thread) ------- + def screen(self, color=False): + if not self.pane: + raise RuntimeError("no TMUX_PANE (not spawned inside tmux?)") + cmd = ["tmux", "capture-pane", "-p", "-t", self.pane] + if color: + cmd.insert(2, "-e") + text = subprocess.check_output(cmd, text=True) + return {"pane": self.pane, "text": text} + + +ADAPTER = GdbAdapter() + +# read verbs (touch gdb state; quick) +_READ = {"state", "regs", "bt", "mem", "breakpoints"} + +# mutating verbs: wire name -> adapter attribute +_MUT = { + "break": "brk", "tbreak": "tbreak", "delete": "delete", "watch": "watch", + "print": "eval", "x": "examine", "set": "setvar", + "frame": "frame", "up": "up", "down": "down", "cmd": "cmd", +} + +# run-control verbs may execute arbitrary inferior code -> generous timeout +_RUNCONTROL = {"run", "start", "cont", "continue", "next", "step", + "nexti", "stepi", "finish", "until"} +RUNCONTROL_TIMEOUT = 120.0 + + +def _rc_cmdline(method, params): + if method == "run": + a = params.get("args") + return "run" + (f" {a}" if a else "") + if method == "start": + return "start" + if method in ("cont", "continue"): + return "continue" + if method in ("next", "step", "nexti", "stepi"): + return f"{method} {int(params.get('n', 1))}" + if method == "finish": + return "finish" + if method == "until": + loc = params.get("loc") + return "until" + (f" {loc}" if loc else "") + raise ValueError(f"not a run-control verb: {method}") + + +def run_control_dispatch(method, params): + """Orchestrated on the SOCKET thread. gdb.execute() run-control returns + async under post_event, so we settle on the real stop/exited event. + The main thread must stay free to *deliver* that event, hence we only + marshal the execute() + state read onto it, and wait here.""" + cmdline = _rc_cmdline(method, params) + settled = threading.Event() + err = {} + + def on_stop(_evt): + settled.set() + + def on_exit(_evt): + settled.set() + + def arm_and_go(): + # arm listeners then fire the command, all on the main thread so the + # ordering is race-free even if execute() blocks until stop (sync mode) + gdb.events.stop.connect(on_stop) + gdb.events.exited.connect(on_exit) + echo_cmd(cmdline) + try: + gdb.execute(cmdline, to_string=False) + except gdb.error as e: + err["e"] = str(e) + settled.set() + + gdb.post_event(arm_and_go) + ok = settled.wait(RUNCONTROL_TIMEOUT) + # tear down listeners on the main thread + def disarm(): + try: + gdb.events.stop.disconnect(on_stop) + except Exception: # noqa: BLE001 + pass + try: + gdb.events.exited.disconnect(on_exit) + except Exception: # noqa: BLE001 + pass + try: + on_main(disarm, timeout=10) + except Exception: # noqa: BLE001 + pass + if "e" in err: + msg = err["e"] + # a program that exits/finishes during the command is a normal + # outcome, not an error: settle to the (no-inferior) state + note. + low = msg.lower() + if "exited" in low or "not being run" in low: + return {"cmd": cmdline, "state": on_main(ADAPTER.state), + "note": msg} + raise RuntimeError(msg) + if not ok: + raise TimeoutError( + f"run-control {cmdline!r} did not settle in {RUNCONTROL_TIMEOUT}s") + return {"cmd": cmdline, "state": on_main(ADAPTER.state)} + + +def dispatch(method, params): + """Return a result dict/value, or raise. Runs on the socket thread.""" + if method == "ping": + return ADAPTER.ping() + if method == "methods": + return ADAPTER.methods() + if method == "quit": + # scheduled by the server after the reply is flushed + return {"ok": True, "bye": True} + if method == "screen": + return ADAPTER.screen(**params) + if method in _READ: + fn = getattr(ADAPTER, method) + return on_main(lambda: fn(**params)) + if method in _RUNCONTROL: + return run_control_dispatch(method, params) + if method in _MUT: + fn = getattr(ADAPTER, _MUT[method]) + return on_main(lambda: fn(**params)) + raise ValueError(f"unknown method: {method}") + + +# -------------------------------------------------------------------------- +# server (background thread) +# -------------------------------------------------------------------------- + +class Server(threading.Thread): + daemon = True + + def __init__(self, path): + super().__init__(name="gdb-driver-rpc") + self.path = path + self._sock = None + self._busy = threading.Lock() # single-driver gate + + def run(self): + try: + if os.path.exists(self.path): + os.unlink(self.path) + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.bind(self.path) + os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR) # 0600 + s.listen(4) + self._sock = s + ADAPTER.ready = True + # NB: log to a file, not the pane -- writing here would land on + # gdb's startup prompt line and corrupt the hand-typed look. + try: + with open(self.path + ".log", "a") as lf: + lf.write(f"listening on {self.path}\n") + except OSError: + pass + while True: + conn, _ = s.accept() + # handle each connection concurrently so the single-driver + # gate can *refuse* a second client instead of queuing it + threading.Thread(target=self._serve, args=(conn,), + daemon=True).start() + except Exception: # noqa: BLE001 + sys.stderr.write("[gdb-driver] server crashed:\n") + traceback.print_exc() + + def _serve(self, conn): + try: + self._loop(conn) + finally: + conn.close() + + def _loop(self, conn): + f = conn.makefile("rwb") + for raw in f: + raw = raw.strip() + if not raw: + continue + rid = None + try: + msg = json.loads(raw) + rid = msg.get("id") + method = msg["method"] + params = msg.get("params") or {} + except Exception as e: # noqa: BLE001 - malformed frame + f.write(_enc({"id": rid, "error": {"message": str(e)}})) + f.flush() + continue + # single-driver gate is PER-REQUEST: it guards against two clients + # dispatching concurrently, but is not held during the idle + # readline wait (which would race a fast sequential client's + # teardown and spuriously refuse it). + if not self._busy.acquire(blocking=False): + f.write(_enc({"id": rid, "error": { + "message": "busy: another request is in flight"}})) + f.flush() + continue + try: + resp = {"id": rid, "result": dispatch(method, params)} + except Exception as e: # noqa: BLE001 - errors are data + resp = {"id": rid, "error": {"message": str(e), + "type": type(e).__name__}} + finally: + # release BEFORE writing: the write flush lets the client fire + # its next request, which (on its own thread) must not see the + # gate still held (that race caused spurious 'busy'). + self._busy.release() + f.write(_enc(resp)) + f.flush() + if method == "quit": + gdb.post_event(lambda: gdb.execute("quit")) + return + + +def _enc(obj): + return (json.dumps(obj, separators=(",", ":")) + "\n").encode() + + +# -------------------------------------------------------------------------- +# bootstrap +# -------------------------------------------------------------------------- + +def _init(): + # make gdb non-interactive-safe for programmatic driving + for cmd in ("set pagination off", "set confirm off"): + try: + gdb.execute(cmd, to_string=True) + except gdb.error: + pass + path = _sock_path() + Server(path).start() + + +_init() |