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.

7 files changed, 1751 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..f517d84
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,16 @@
+# python runtime cruft
+__pycache__/
+*.pyc
+.venv/
+venv/
+
+# gdb core dumps from target crashes
+core
+core.*
+
+# livestream / session scratch — not part of the tool
+.auto/
+crack-roulette.md
+solve-puzzle.md
+*_solution.md
+lain_is_here.png
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..8df2560
--- /dev/null
+++ b/README.md
@@ -0,0 +1,134 @@
+# gdb-driver
+
+Puppeteer a **live** gdb (plain CLI + [GEF](https://github.com/hugsy/gef), **not**
+`-tui`) programmatically over a unix socket, while it renders normally in a tmux
+pane. An embedded Python plugin runs an RPC server on gdb's own thread, so
+debugger state is read and written directly — and every driven command is echoed
+at the `gef>` prompt exactly as if a human typed it.
+
+It was built for **agent-driven dynamic analysis on a livestream**: the pane is a
+real, watchable gdb session (GEF's colored context is what a viewer sees), but a
+script is doing the driving.
+
+```
+  your script ──JSONL──▶ unix socket ──▶ driver.py (in gdb's Python)
+                                              │  marshals onto gdb's thread
+   tmux pane  ◀── gef> echo + GEF context ────┘  reads/writes real state
+```
+
+## Why
+
+gdb automation usually means GDB/MI or brittle `expect` scraping. Neither gives
+you *both* a clean structured API **and** a genuine, human-looking terminal
+session. gdb-driver does:
+
+- **Three tiers of control** — raw tmux keys, semantic verbs (run-control /
+  breakpoints / eval), and structured introspection (regs / memory / backtrace
+  as data).
+- **Real stop-event settling.** Mutating verbs return only after gdb's actual
+  `stop`/`exited` event fires — no sleeps, no guessing. The returned `state` is
+  final.
+- **Visible by construction.** Everything is echoed at the prompt and rendered by
+  GEF, so the pane is legible to a human (or a stream) the whole time.
+- **The escape hatch is first-class.** `drive cmd "<any gdb/GEF command>"` runs
+  anything — `disassemble`, `vmmap`, `checksec`, even in-gdb `python …` — so you
+  are never blocked by a missing verb.
+
+All verbs go through gdb's **CLI** path (GEF-compatible). GDB/MI is not used.
+
+## Requirements
+
+- Linux, running **inside tmux** (the driver spawns/handles panes).
+- **gdb 17+** with **GEF** auto-loaded (`~/.gdbinit` → `~/.gef.py`).
+- `python3` (stdlib only) for the `drive`/`pane` client CLIs. The `driver.py`
+  plugin runs inside gdb's own embedded Python.
+
+No pip install, no dependencies — it's three files and the stdlib.
+
+## Quickstart
+
+```bash
+cd gdb-driver   # so ./drive resolves (it's a plain executable)
+
+# 1. spawn a pane with gdb on a target (+ program args). Blocks until ready,
+#    then prints:  ready pane=%N sock=/run/.../gdb-XXXX.sock
+./drive spawn /abs/path/to/binary [prog-args...]
+#    forward environment to the inferior (shim libs, LD_PRELOAD, …):
+#    ./drive spawn --env LD_LIBRARY_PATH=/path/to/libs ./target arg1
+
+# 2. drive it. ./drive auto-finds the socket when one pane is live (no --sock).
+./drive where                 # main @ 0x555… at file:line (or func+0xoff) [stopped]
+./drive break main            # b *main+42 | b <func> | tbreak …
+./drive run                   # settles at the next real stop (breakpoint/exit)
+./drive bt / regs / mem '$sp' 64
+./drive print '(char*)$rdi'; ./drive cont; ./drive finish
+./drive cmd "disassemble main"    # ANY gdb/GEF command (the escape hatch)
+./drive set '$r14d = 5'           # patch a register live
+./drive screen [color]            # tmux capture of the pane (GEF context)
+
+# 3. tear down (graceful quit + kill pane + rm socket). Never leak panes.
+./drive stop
+./drive list [--prune]            # inventory; --prune drops dead panes
+```
+
+**Socket resolution** for `./drive`: `--sock <path>` (anywhere on the line), else
+`$GDB_DRIVER_SOCK`, else the single live pane (with several, it lists them). Calls
+are **sequential** — single-driver, one connection at a time; a second concurrent
+request gets `busy`.
+
+## Command surface
+
+- **Lifecycle / meta:** `ping` · `methods` (self-documenting verb table) · `quit`
+  · `spawn` · `list [--prune]` · `stop`
+- **Read (never settle):** `state`/`where` · `regs` · `bt [n]` · `mem <addr-expr>
+  [len]` · `breakpoints`/`bp` · `screen [color]`
+- **Run-control (settle on the real `stop`/`exited` event):** `run [args]` ·
+  `start` · `cont`/`c` · `next [n]`/`n` · `step [n]`/`s` · `stepi`/`si` ·
+  `nexti`/`ni` · `finish` · `until [loc]`
+- **Breakpoints:** `break <loc>`/`b` · `tbreak <loc>` · `delete [n]` ·
+  `watch <expr>`
+- **Eval / examine / patch:** `print <expr>`/`p` · `x <fmt> <addr>` ·
+  `set <expr>` (vars and registers: `set '$rax=1'`) · `frame <n>` · `up`/`down`
+- **Escape hatch:** `cmd <any gdb/GEF command…>`
+- **Raw keys:** `keys <tmux-key…>` (e.g. `keys C-c` to interrupt a running
+  inferior) — client-side tmux `send-keys`, so it works even while a `cont` is
+  blocking.
+
+Mutating verbs settle, then return a fresh `{cmd, state[, output, note]}`. Reads
+return immediately; a program-dependent read before the inferior is stopped
+returns a clean `not ready` error rather than garbage.
+
+## Layout
+
+| file | what |
+|------|------|
+| `driver.py` | the embedded gdb-Python plugin: unix-socket JSONL server on a background thread, marshalling to gdb's thread via `post_event`, all verbs, run-control stop-event settling, and `gef>`-prompt echoing. |
+| `pane.py` | tmux spawn/stop/list, registry under `$XDG_RUNTIME_DIR/gdb-driver/`, readiness poll, socket resolution, `send_keys`. |
+| `drive` | the ergonomic CLI (terse text, auto socket). Use this all day. |
+| `tests/smoke.py` | end-to-end: compiles a fixture, spawns, drives the whole surface, ~28 asserts. A protocol-regression lock. |
+| `TUI_DRIVING_BLUEPRINT.md` | the design / generalization reference (this is a *Level-2 embedded* adapter). |
+
+## Testing
+
+```bash
+python3 tests/smoke.py     # compiles a fixture, exercises the full surface
+```
+
+## Gotchas worth knowing
+
+A few that will bite (the blueprint and inline comments have the rest):
+
+- **Break at the `call`, not the `mov`/`lea` before it** — argument registers
+  (`$rdi`/`$rsi`/…) are only set *at* the call.
+- **PIE raw-offset breakpoints fail before load** (`break *0x1275` → *Cannot
+  access memory*). Use symbol-relative: `break *main+<off>` so gdb relocates it.
+- **Plugin changes need a respawn** — editing `driver.py` won't affect a running
+  gdb; `./drive stop && ./drive spawn …` reloads it.
+- **fork + ptrace anti-debug targets** (a parent↔child `PTRACE_SEIZE` coroutine)
+  do **not** reproduce native behavior under gdb — a process can have only one
+  tracer. Observe those with eBPF/bpftrace (orthogonal to ptrace) instead of
+  trusting gdb-extracted intermediate state.
+
+## License
+
+No license yet — all rights reserved by the authors pending one.
diff --git a/TUI_DRIVING_BLUEPRINT.md b/TUI_DRIVING_BLUEPRINT.md
new file mode 100644
index 0000000..094f87a
--- /dev/null
+++ b/TUI_DRIVING_BLUEPRINT.md
@@ -0,0 +1,355 @@
+# Blueprint: a generic TUI-driving layer
+
+How to bring the "spawn a pane, drive the real TUI over a socket, read structured
+state back" experience (built for **idatui**, see `docs/RPC.md`) to *other* TUI
+software — most notably **gdb**, but the pattern generalizes to any terminal app
+(vim, k9s, lazygit, htop, a REPL, …).
+
+This is a design reference, not code. It captures the invariants worth keeping,
+the parts that are reusable as-is, and the one hard problem you re-solve per
+target (knowing when the UI has *settled*).
+
+---
+
+## 1. The reference implementation (what we're generalizing)
+
+idatui is a Textual app we **own**, so we embedded an RPC server directly in its
+asyncio event loop. The stack that emerged:
+
+```
+ tmux pane A (viewers see this)          tmux pane B (the agent)
+ ┌────────────────────────────┐  unix    ┌────────────────────────┐
+ │ the TUI, rendering normally │◀socket──▶│ ergonomic client (drive)│
+ │ + in-process RPC server     │  JSONL   │  + pane manager         │
+ └────────────────────────────┘          └────────────────────────┘
+```
+
+Pieces (all reusable ideas): a **unix-socket JSONL server**, **three method
+tiers** (raw keys / semantic verbs / structured introspection), a **settle**
+primitive (block until the UI is quiescent before returning state), a
+**single-driver gate**, a **tmux pane manager** (spawn/stop/list + backend
+auto-start + readiness polling), and an **ergonomic client** (auto socket
+discovery, terse text, composite gestures).
+
+---
+
+## 2. Invariants worth preserving (why it felt good)
+
+1. **Drive the *real* UI, not a hidden API.** Every action goes through the same
+   input path a keyboard would, so the tmux pane shows it happening (livestream /
+   debuggability). Bypassing the renderer to mutate state directly is faster but
+   defeats the purpose.
+2. **Three tiers, always.** `raw` (keystrokes) for fidelity/coverage; `semantic`
+   verbs (typed, with an optional per-char delay for the hand-typed look) for
+   ergonomics; `introspection` (structured reads) so the agent reasons over data,
+   not screen-scraping.
+3. **Every mutating call returns *after* the UI settled**, and returns fresh
+   state. The driver never acts on a half-rendered frame.
+4. **Single-driver by default.** One client at a time; a second concurrent
+   connection is refused. No multi-driver coordination to reason about.
+5. **Self-documenting surface.** A `methods` call returns the verb table; a
+   `state`/`screen` call returns what's happening. The agent can discover the API.
+6. **Lifecycle is first-class.** Spawn a pane, wait until *ready*, drive, tear it
+   down; never leak panes/processes. Auto-start the backing daemon if it's down.
+7. **An ergonomic layer on top.** The raw JSON transport is correct but verbose;
+   a thin CLI that auto-resolves the socket and prints terse text is what you
+   actually use all day.
+
+---
+
+## 3. Generic architecture
+
+Split into a **reusable core** and a per-target **Adapter**. Only the adapter
+changes between targets.
+
+```
+        ┌──────────────── reusable core ────────────────┐
+client ─┤ transport (unix sock, JSONL, single-driver)   │
+ (drive)│ dispatch (method table, tiers, error framing) │
+        │ pane manager (tmux spawn/stop/list, readiness) │
+        └───────────────────────┬───────────────────────┘
+                                 │  Adapter interface
+                    ┌────────────┴────────────┐
+                    │  Target Adapter          │  ← the only per-target code
+                    │  inject / settle /       │
+                    │  snapshot / screen /     │
+                    │  semantic verbs          │
+                    └──────────────────────────┘
+```
+
+### The Adapter interface (the contract)
+
+Everything a target must provide. Keep it small:
+
+| method | purpose |
+|--------|---------|
+| `inject_keys(keys[])` | push keystrokes into the app's real input path |
+| `inject_text(str, delay_ms)` | type a literal string (visible typing) |
+| `settle(pred?, timeout) -> bool` | block until quiescent, then until `pred` (if given) |
+| `snapshot() -> dict` | structured "where am I / what's on screen" state |
+| `screen(fmt) -> {w,h,text}` | full render (plain / colored) of the pane |
+| `verbs` | target-specific semantic ops (each = compose inject + settle + a predicate) |
+| `ready() -> bool` | is the target drivable yet |
+| `quit()` | graceful shutdown |
+
+The core turns these into the wire method table and handles transport, framing,
+the single-driver gate, and the pane lifecycle.
+
+---
+
+## 4. Target taxonomy — how much the app cooperates
+
+The adapter's implementation depends entirely on how much access you get. Three
+levels, best to worst:
+
+### Level 1 — **Embedded** (you own or can patch the app)
+Run an RPC server *inside* the app's event loop (idatui). Injection uses the
+app's own key API; `settle` uses its message pump + task/worker state; `snapshot`
+reads app state directly. **Highest fidelity, cheapest settle** (you have real
+signals). Use when you can add ~200 lines to the target.
+
+### Level 2 — **Control-channel** (app has a machine interface)
+Many serious TUIs expose a second, structured channel alongside the visual one:
+gdb (**GDB/MI** + a **Python API**), tmux (**control mode**, `-CC`), vim
+(**channels / `--remote-expr`**), lldb, many REPLs. Strategy: the **visual** pane
+is driven with keystrokes for fidelity; **semantic** verbs and **introspection**
+go through the control channel, which is request/response (so quiescence is
+*given* — a command is done when the channel replies). This is the sweet spot for
+gdb (see §8).
+
+### Level 3 — **Black-box PTY** (you can't modify or query it)
+The universal fallback: the app is just bytes in / a screen out (vim without
+channels, htop, an arbitrary curses app). You inject keystrokes into its PTY and
+read state by scraping the rendered screen. **Settle is the hard part** (no
+completion signal — you debounce on screen changes). Everything is a keystroke
+macro + screen assertion.
+
+> **Most targets are Level 2 or 3.** Design the core so an adapter can *mix*
+> levels: e.g. gdb = Level 2 for `break/step/bt/read-mem`, Level 3 (screen
+> scrape) for whatever the control channel doesn't expose.
+
+---
+
+## 5. The three hard problems, generically
+
+### 5a. Input injection — "how do keystrokes get in"
+- **Embedded:** call the app's key-press API (idatui used Textual's `_press_keys`,
+  the same path its own tests use). Supports a `wait:<ms>` token → free typed delay.
+- **Control-channel:** structured actions bypass keys entirely (gdb `-exec-*`);
+  but *also* keep a keystroke path for TUI navigation.
+- **Black-box:** **`tmux send-keys -t <pane> …`** is the universal injector — no
+  PTY plumbing. It even names special keys (`Enter`, `C-c`, `Escape`). For a
+  literal string, `send-keys -l 'text'`. For the typed aesthetic, send char-by-char
+  with a sleep between.
+
+### 5b. Settle / quiescence — "when did the UI finish reacting" *(the crux)*
+Pick the strongest signal the target offers. A taxonomy, best → worst:
+
+1. **Completion signal** *(embedded / control-channel).* The app tells you it's
+   done: a worker/task queue drains (idatui: message pump + `WorkerManager`), or a
+   request/response channel returns (`^done` in GDB/MI). Deterministic; prefer this.
+2. **Predicate wait.** Poll a cheap boolean that means "the thing I asked for
+   happened" (idatui: `dec.loaded_ea == target`). Always pair a semantic verb with
+   its own predicate where one exists.
+3. **Prompt / marker detection** *(black-box, structured-ish).* The screen returns
+   to a known prompt regex, or the app emits a sentinel you injected (e.g. drive it
+   to `echo <nonce>` and wait for `<nonce>` on screen). Robust when a prompt exists.
+4. **Output debounce** *(black-box fallback).* Snapshot the screen every ~20 ms;
+   consider it settled once it's unchanged for N consecutive samples (e.g. 3× =
+   ~60 ms) or a hard timeout. Flakiest; tune N per app; combine with (3).
+
+Factor settle into a shared helper that takes the *yield/poll* strategy and an
+optional predicate (idatui's `_sync.settle(app, pred)` is exactly this — reused by
+both the live driver and the test suite). That sharing is high-value: your tests
+and your driver then agree on "settled."
+
+### 5c. Introspection — "what is on screen / what is the state"
+Same best→worst gradient:
+1. **Native state** (embedded): read the app's model directly → richest snapshot.
+2. **Control-channel queries** (gdb: frames, registers, `-data-read-memory`,
+   breakpoints as JSON) → structured, no scraping.
+3. **Screen scrape** (black-box): render the pane to a text grid and parse it.
+   `tmux capture-pane -p -t <pane>` (add `-e` to keep colors) is the universal
+   screen read. For a machine-parseable grid without a real terminal, feed the PTY
+   stream through a headless emulator (e.g. **pyte**) and read its buffer.
+
+Always expose a raw `screen()` too (plain + colored). It's the agent's "look with
+your eyes" fallback and a great debugging aid (idatui added `format=text|html|svg`;
+`html` is nice to pipe to an out-of-band web viewer).
+
+---
+
+## 6. tmux as the universal substrate
+
+For **Level 2/3** targets, don't own the PTY yourself — let **tmux** own it and
+drive through tmux. This collapses three problems into three commands and reuses
+the pane manager we already built:
+
+| need | tmux primitive |
+|------|----------------|
+| a visible pane running the target | `tmux split-window -P -F '#{pane_id}' '<cmd>'` |
+| inject keystrokes | `tmux send-keys -t <pane> …` (named keys, `-l` literal) |
+| read the screen | `tmux capture-pane -p [-e] -t <pane>` |
+| is the pane alive | `tmux list-panes -a -F '#{pane_id}'` |
+| tear down | `tmux kill-pane -t <pane>` |
+
+A black-box adapter can be built *entirely* on these four. The pane manager
+(`spawn/stop/list`, a small registry in `$XDG_RUNTIME_DIR`, backend auto-start,
+readiness polling) carries over unchanged — only "what counts as ready" and "what
+command to spawn" are per-target.
+
+Caveat: `capture-pane` gives you the *rendered* grid but not semantic structure;
+and `send-keys` is fire-and-forget (no completion signal) → you lean on §5b (3)/(4)
+for settle. That's the price of black-box.
+
+---
+
+## 7. Reusable vs. per-target
+
+**Reuse verbatim** (target-independent):
+- Transport: unix socket, newline-delimited JSON, `{id,method,params}` →
+  `{id,result|error}`; single-driver gate; per-request dispatch + error framing.
+- Pane manager: `spawn/stop/list`, registry, backend `_ensure_server`, readiness
+  poll, graceful `quit` (answer then exit).
+- Client + ergonomic CLI: socket auto-resolution (single live pane), terse text
+  output, composite verbs, `raw <method> k=v` passthrough, self-documenting
+  `methods`.
+- The settle *shape* (`wait_for(pred, tick)` + `settle(pred?)`), even though the
+  concrete signals differ.
+
+**Write per-target** (the adapter):
+- Input injection binding (app key API / control channel / `tmux send-keys`).
+- Settle signal (which of §5b applies).
+- `snapshot()` and `screen()` sources.
+- The semantic verb set (the app's real vocabulary).
+- `ready()` and the spawn command.
+
+Rule of thumb: **~80% reuse, ~20% adapter.** Keep the adapter interface narrow so
+that stays true.
+
+---
+
+## 8. Worked example: **gdb**
+
+gdb is a *Level 2* target with a great control channel, so aim for a hybrid.
+
+**Layout.** A tmux pane runs gdb in TUI mode (`gdb -q -tui` / `layout src`,
+`layout asm`, `layout regs`) — that's what viewers see. Alongside it, a **gdb
+Python plugin** (loaded with `-x driver.py`, running inside gdb's own process)
+opens the unix socket and *is* the adapter. gdb's Python runs on gdb's thread, so
+handlers touch gdb state directly — the embedded pattern, for free, inside a
+program you didn't write.
+
+**Injection.**
+- *Semantic* verbs call the API directly (visible in the TUI because gdb echoes
+  and repaints): `gdb.execute("break main", to_string=True)`, `-exec-run`,
+  `-exec-next`, `-exec-continue`, `-exec-finish`. Or GDB/MI via a second channel
+  if you prefer strict JSON.
+- *Raw* verbs (for TUI-only navigation: `C-x o` to switch windows, PgUp/PgDn in
+  the source window, `C-x 2` layouts) go through **`tmux send-keys`** to the pane.
+
+**Settle.** Mostly *given*: `gdb.execute(..., to_string=True)` and MI commands are
+synchronous — they return when the command completed, so a semantic verb is
+settled the moment the call returns. For *async* execution (`-exec-continue` while
+the inferior runs), settle = wait for the next **stop event** (`gdb.events.stop`)
+or the MI `*stopped` async record. For raw `send-keys` TUI moves, fall back to
+`capture-pane` debounce (§5b-4).
+
+**Introspection** (all structured, no scraping):
+- `state()` → `{running|stopped, pc, function, file:line, thread, selected_frame}`
+  from `gdb.selected_frame()`, `gdb.selected_thread()`.
+- `backtrace()` → walk `gdb.newest_frame()` → `[{level,pc,func,file,line,args}]`.
+- `regs()` → `frame.read_register(...)` for the ABI set.
+- `mem(addr,len)` → `gdb.selected_inferior().read_memory(...)` (hex/ascii).
+- `locals()`, `breakpoints()` (`gdb.breakpoints()` → JSON), `disas(addr?)`.
+- `screen()` → `tmux capture-pane -ep` of the TUI pane (the "as a viewer sees it"
+  read), *plus* the structured reads above for reasoning.
+
+**Semantic verb set** (the gdb vocabulary): `run/start`, `cont`, `next`, `step`,
+`finish`, `until`, `break <loc>`, `tbreak`, `delete <n>`, `watch <expr>`,
+`bt [n]`, `frame <n>`, `up/down`, `print <expr>`, `x/<fmt> <addr>`, `set var`,
+`layout <src|asm|regs|split>`, `focus <win>`, `raw keys …`.
+
+**Spawn / readiness.** `spawn --bin ./prog [--args …]` → tmux pane runs
+`gdb -q -tui -x driver.py --args ./prog …`; **ready** when the plugin's socket is
+up *and* gdb reached its prompt (the plugin can signal readiness once loaded).
+`stop` = graceful `quit` verb (plugin calls `gdb.execute("quit")`) then kill-pane.
+
+This gives the same feel as idatui: `drive where` → `#3 main at foo.c:42`,
+`drive bt`, `drive break foo`, `drive cont`, `drive x/16xb $sp`, `drive screen` —
+terse, socket auto-resolved, every action visible in the gdb TUI pane.
+
+---
+
+## 9. Method-surface conventions (keep these consistent across targets)
+
+- **Tiers, named the same everywhere:** `keys`/`text` (raw); `state`/`view`/
+  `screen`/`<structured reads>` (introspection); `<semantic verbs>` (per target);
+  `ping`/`methods`/`quit` (lifecycle).
+- **`ping`** returns `{ok, proto, ready, …}`; **`methods`** returns the verb table
+  (self-documentation); **`quit`** answers *then* exits (so the reply flushes).
+- Mutating verbs **settle then return fresh `state`**. Read verbs never settle.
+- Program-dependent verbs return a clean **`error: not ready`** before init.
+- Typed verbs accept **`delay_ms`** (visible typing; `0` = fast). Movement verbs
+  are fast (no delay, light settle).
+- Errors are data (`{id,error:{message}}`), never drop the connection.
+
+## 10. Client ergonomics (the part you use all day)
+
+- **Auto-resolve the socket**: if exactly one live pane, use it; else `--sock` /
+  env; with several, list and ask. Kills the "copy the socket path everywhere"
+  tax — the single biggest quality-of-life win.
+- **Terse text out**, not raw JSON (`where` → `main @ 0x… [src] L42`); keep a
+  `raw <method> k=v` passthrough for the long tail.
+- **Composite gestures** for the common multi-step flows (idatui's
+  `rename old new` = goto+cursor+type; gdb's `break-and-run`, `stepn N`).
+- Fire calls **sequentially** (single-driver); each invocation opens/closes its
+  own connection.
+
+## 11. Security & operations
+
+- **Unix socket, mode 0600, local only.** No auth by design (whoever can r/w the
+  socket drives it). Add a first-line shared token only if you bind TCP.
+- **Single-driver gate** prevents two clients interleaving mutations.
+- **Never leak panes/daemons.** Registry + `list --prune`; auto-start shared
+  backends idempotently (probe before spawn; a port/socket already-in-use guards
+  duplicates); don't kill shared backends on a per-pane `stop`.
+- Note the **stale-pane** failure mode: a pane can outlive its backend/session;
+  `list` should check liveness, not just tmux presence.
+
+## 12. Onboarding a new target — checklist
+
+1. **Classify** it (§4): can you embed? does it have a control channel? else
+   black-box.
+2. Pick the **injection** binding (§5a) and the **settle** signal (§5b) — decide
+   this first; everything else is easy.
+3. Implement the **Adapter** (§3): `inject_keys/text`, `settle`, `snapshot`,
+   `screen`, `ready`, `quit`, and the **semantic verbs** (the app's real
+   vocabulary — don't invent; mirror what a power user types).
+4. Wire the **spawn command** + **readiness** into the pane manager.
+5. Reuse transport, client, and ergonomic CLI unchanged.
+6. Add a **smoke test** that spawns the target on a socket and drives the whole
+   surface end-to-end (idatui's `rpc_smoke.py` is the template — it also locks the
+   protocol against regressions).
+
+## 13. Pitfalls & lessons (paid for once already)
+
+- **Settle is where the bugs live.** A verb that "passes" by timing out then
+  reading stale state is the classic false-green (idatui hit this: a 25 s hang that
+  a check passed *trivially*). Prefer a real signal/predicate; print per-op timing
+  so a suddenly-slow verb (a hidden timeout) is visible.
+- **Newlines / control bytes** matter when injecting via a line-based reader —
+  pick an escaping convention (idatui: literal `\n` → real newline at apply time).
+- **Black-box screen scraping is lossy** — no semantic structure, colors optional,
+  non-deterministic chrome (a live clock makes frames differ). Crop chrome; prefer
+  structured reads when a control channel exists.
+- **Match on what's shown, resolve by canonical id.** Names can render differently
+  than they're stored (idatui: `.init_proc` shows as `init_proc`).
+- **Keep the app visibly driven.** If you ever bypass the UI for speed, gate it
+  behind an explicit `fast:true` — the default should always render, because
+  "shown as if a user did it" is the whole point.
+
+---
+
+*Reference implementation: `idatui/{rpc,rpcclient,drive,pane,_sync}.py`,
+`docs/RPC.md`, `tests/rpc_smoke.py` in this repo.*
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:])
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()
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))
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())