path: root/drive

#!/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:])