mcp: delete the ida-pro-mcp transport, supervisor, and mcp-only tests

The idalib worker is the only backend now, so remove the dead HTTP/supervisor
surface entirely (~2200 lines):

* deleted idatui/client.py (the IDAClient HTTP/JSON-RPC transport + session
  manager), idatui/tui.py (the old mcp TUI entry, superseded by launch.py),
  spawn.sh, and systemd/ (the supervisor unit).
* deleted the mcp-only tests (stress_client, smoke_client, test_keepalive,
  stress_paging, rpc_smoke, serverctl.sh, pane_smoke, test_domain) -- the worker
  pilot (tests/test_scenarios.py) supersedes them.
* migrated the tmux RPC harness (idatui/pane.py) to the worker: it spawns
  `idatui.launch <binary> --rpc <sock>` instead of the mcp `idatui.tui`, drops
  the supervisor auto-start/ensure machinery, and reaps our own worker
  (idatui/worker.py) instead of ida_pro_mcp.idalib_server. --db/--url/--no-
  ensure-server are gone; --open is required.
* __init__ / __main__ / domain no longer import client (exceptions come from
  errors.py, the domain client hint is WorkerClient); pyproject points both
  console scripts at idatui.launch; README + ida-tui header describe the
  worker-only flow.

What stays (by design): the ida_pro_mcp *package* (the worker reuses its @tool
functions in-process) and server/patch_server.py (the worker injects its custom
tools on startup). Verified: whole package imports + IdaTui constructs + pilot
lists 31 scenarios. The worker pilot (134 pass / 2 known flakes) is the E2E gate.

8 files changed, 0 insertions, 1305 deletions

diff --git a/tests/pane_smoke.py b/tests/pane_smoke.py
deleted file mode 100644
index bd7def6..0000000
--- a/tests/pane_smoke.py
+++ /dev/null
@@ -1,87 +0,0 @@
-#!/usr/bin/env python3
-"""Smoke test for idatui.pane's supervisor auto-start machinery (no IDA needed).
-
-    python3 tests/pane_smoke.py
-
-Uses IDATUI_SERVER_CMD to launch a dummy port-binder in a tmux pane instead of
-the real ./spawn.sh, so we can exercise _ensure_server end-to-end (start, detect,
-idempotency, remote guard) without a real ida-pro-mcp server. Must run in tmux.
-"""
-import os
-import socket
-import subprocess
-import sys
-import tempfile
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from idatui import pane  # noqa: E402
-
-PASS = FAIL = 0
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ok   {name}")
-    else:
-        FAIL += 1
-        print(f"  FAIL {name}   {detail}")
-
-
-def _free_port() -> int:
-    s = socket.socket()
-    s.bind(("127.0.0.1", 0))
-    p = s.getsockname()[1]
-    s.close()
-    return p
-
-
-def main() -> int:
-    if not os.environ.get("TMUX"):
-        print("error: must run inside tmux", file=sys.stderr)
-        return 2
-
-    # a dummy "supervisor": bind the port and idle, so _server_up sees it.
-    fake = tempfile.NamedTemporaryFile("w", suffix=".py", delete=False)
-    fake.write(
-        "import socket,sys,time\n"
-        "s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)\n"
-        "s.bind(('127.0.0.1',int(sys.argv[1]))); s.listen(); time.sleep(300)\n")
-    fake.close()
-    port = _free_port()
-    os.environ["IDATUI_SERVER_CMD"] = f"{sys.executable} {fake.name} {port}"
-
-    server_pane = None
-    try:
-        check("port starts down", not pane._server_up("127.0.0.1", port))
-
-        srv = pane._ensure_server("127.0.0.1", port, timeout=15.0)
-        server_pane = srv.get("server_pane")
-        check("ensure_server starts the supervisor and it comes up",
-              srv.get("server_started") and srv.get("server_up") and server_pane,
-              str(srv))
-        check("port is now up", pane._server_up("127.0.0.1", port))
-
-        srv2 = pane._ensure_server("127.0.0.1", port, timeout=5.0)
-        check("ensure_server is idempotent when already up",
-              srv2.get("server_started") is False and srv2.get("server_up") is True,
-              str(srv2))
-
-        srv3 = pane._ensure_server("10.255.255.1", 9, timeout=2.0)
-        check("remote+down server is not auto-started",
-              srv3.get("server_started") is False and "local" in (srv3.get("error") or ""),
-              str(srv3))
-    finally:
-        if server_pane:
-            subprocess.run(["tmux", "kill-pane", "-t", server_pane],
-                           capture_output=True)
-        os.unlink(fake.name)
-        os.environ.pop("IDATUI_SERVER_CMD", None)
-
-    print(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
diff --git a/tests/rpc_smoke.py b/tests/rpc_smoke.py
deleted file mode 100644
index a38162e..0000000
--- a/tests/rpc_smoke.py
+++ /dev/null
@@ -1,284 +0,0 @@
-#!/usr/bin/env python3
-"""End-to-end smoke test for the RPC server, in-process over a real unix socket.
-
-    ~/ida-venv/bin/python tests/rpc_smoke.py --db <session_id>
-
-Boots the TUI headless with --rpc on a temp socket, connects a JSONL client over
-that socket (same loop), and exercises the raw + introspection primitives.
-"""
-import asyncio
-import json
-import os
-import sys
-import tempfile
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from idatui.app import IdaTui  # noqa: E402
-from idatui.client import DEFAULT_URL  # noqa: E402
-from idatui._sync import wait_for  # noqa: E402
-from textual.widgets import DataTable  # noqa: E402
-
-PASS = FAIL = 0
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ok   {name}")
-    else:
-        FAIL += 1
-        print(f"  FAIL {name}   {detail}")
-
-
-class Conn:
-    """Tiny JSONL client over a unix socket."""
-    def __init__(self, r, w):
-        self.r, self.w = r, w
-        self._id = 0
-
-    async def call(self, method, **params):
-        self._id += 1
-        self.w.write(json.dumps({"id": self._id, "method": method,
-                                 "params": params}).encode() + b"\n")
-        await self.w.drain()
-        line = await self.r.readline()
-        return json.loads(line.decode())
-
-
-async def run(db):
-    sock = os.path.join(tempfile.gettempdir(), f"idatui-rpc-{os.getpid()}.sock")
-    url = os.environ.get("IDA_MCP_URL", DEFAULT_URL)
-    app = IdaTui(url=url, db=db, keepalive=False, rpc_path=sock)
-    async with app.run_test(size=(140, 44)) as pilot:
-        # boot: functions loaded + socket up
-        await wait_for(lambda: app.query_one("#func-table", DataTable).row_count > 0,
-                       pilot.pause, 60)
-        await wait_for(lambda: os.path.exists(sock), pilot.pause, 20)
-        if app._func_index is not None and not app._func_index.complete:
-            app._func_index.load_all()
-
-        r, w = await asyncio.open_unix_connection(sock)
-        c = Conn(r, w)
-
-        resp = await c.call("ping")
-        pong = resp.get("result", {})
-        check("ping returns proto + readiness",
-              pong.get("proto") == 1 and pong.get("ready") is True and pong.get("functions", 0) > 0,
-              str(pong))
-
-        resp = await c.call("methods")
-        check("methods lists the verb table",
-              isinstance(resp.get("result"), dict) and "pseudocode" in resp["result"],
-              str(resp)[:80])
-
-        resp = await c.call("functions", limit=400)
-        funcs = resp.get("result", [])
-        check("functions lists entries", isinstance(funcs, list) and len(funcs) > 0,
-              str(resp)[:120])
-        # a normally-named function (its name appears verbatim in its own decomp,
-        # unlike e.g. '.init_proc' which IDA renders as 'init_proc')
-        target = next((f for f in funcs if f["name"] == "main"),
-                      next((f for f in funcs if f["name"].startswith("sub_")), funcs[0]))
-
-        resp = await c.call("state")
-        st = resp.get("result", {})
-        check("state has an active view", st.get("active") in ("decomp", "disasm", "hex"),
-              str(st)[:120])
-
-        # drive it like a user: goto a function by name via raw keys
-        keys = ["g"] + list(target["name"]) + ["enter"]
-        resp = await c.call("keys", keys=keys)
-        st = resp.get("result", {})
-        check("keys(goto) navigates to the function",
-              st.get("function", {}).get("name") == target["name"],
-              f"got {st.get('function')}")
-
-        resp = await c.call("view", lines=6)
-        v = resp.get("result", {})
-        check("view returns visible lines with a cursor",
-              isinstance(v.get("lines"), list) and any(l.get("cur") for l in v["lines"]),
-              str(v)[:120])
-
-        resp = await c.call("screen")
-        scr = resp.get("result", {})
-        check("screen returns a text grid",
-              isinstance(scr.get("text"), str) and target["name"] in scr["text"],
-              f"len={len(scr.get('text','')) if scr else 0}")
-
-        # raw text primitive into the goto input, then escape out
-        from textual.widgets import Input
-        await c.call("keys", keys=["g"])
-        await c.call("text", text="sub_", settle=False)
-        val = app.query_one("#goto", Input).value
-        check("text primitive types into the focused input", val == "sub_", f"val={val!r}")
-        await c.call("keys", keys=["escape"])
-
-        # --- semantic verbs ------------------------------------------------ #
-        resp = await c.call("goto", target=target["name"], delay_ms=0)
-        check("semantic goto lands on the function",
-              resp.get("result", {}).get("function", {}).get("name") == target["name"],
-              str(resp.get("result", {}).get("function")))
-
-        before = app._active
-        resp = await c.call("toggle_view")
-        check("toggle_view flips the active pane",
-              resp.get("result", {}).get("active") != before, f"still {before}")
-        await c.call("toggle_view")  # flip back to a known state
-
-        # fast movement: cursor should advance a few lines
-        line0 = (await c.call("state"))["result"]["cursor"].get("line")
-        resp = await c.call("move", dir="down", n=4)
-        line1 = resp["result"]["cursor"].get("line")
-        check("move(down,4) advances the cursor",
-              isinstance(line0, int) and isinstance(line1, int) and line1 > line0,
-              f"{line0} -> {line1}")
-
-        # cursor_on: place the cursor on the function's own name by token
-        resp = await c.call("cursor_on", word=target["name"])
-        cur = resp.get("result", {})
-        check("cursor_on lands on the named token",
-              cur.get("found") is True and cur.get("cursor", {}).get("word") == target["name"],
-              str(cur.get("cursor")))
-
-        # rename using the ergonomic word= (cursor_on + prompt-fill in one call)
-        v = (await c.call("view", lines=1))["result"]
-        line0_text = v["lines"][0]["text"] if v.get("lines") else ""
-        col = line0_text.find(target["name"])
-        if col >= 0:
-            newname = f"rpc_{os.getpid()}"
-            await c.call("rename", name=newname, word=target["name"], delay_ms=0)
-            got = app._func_index.by_addr(target["ea"])
-            check("rename word= updates the function name",
-                  got is not None and got.name == newname,
-                  got.name if got else None)
-            app.program.client.call(
-                "rename", batch={"func": {"addr": hex(target["ea"]), "name": target["name"]}})
-        else:
-            check("found the function name to rename", False, repr(line0_text[:60]))
-
-        # --- structured introspection ------------------------------------- #
-        resp = await c.call("resolve", name=target["name"])
-        check("resolve maps a name to its ea",
-              resp.get("result", {}).get("ea") == target["ea"], str(resp.get("result")))
-
-        resp = await c.call("pseudocode", target=target["name"])
-        pc = resp.get("result", {})
-        check("pseudocode returns the full body",
-              isinstance(pc.get("code"), str) and len(pc["code"]) > 0 and not pc["failed"],
-              f"failed={pc.get('failed')} len={len(pc.get('code') or '')}")
-
-        resp = await c.call("disassembly", target=target["name"], max=50)
-        da = resp.get("result", {})
-        check("disassembly returns lines with addresses",
-              isinstance(da.get("lines"), list) and len(da["lines"]) > 0
-              and all("ea" in ln and "text" in ln for ln in da["lines"]),
-              f"total={da.get('total')} n={len(da.get('lines', []))}")
-
-        # a function that is actually referenced, so xrefs_to is non-empty
-        callee = None
-        for f in funcs:
-            xr = (await c.call("xrefs_to", target=f["name"], limit=5)).get("result", [])
-            if xr:
-                callee = (f, xr)
-                break
-        check("xrefs_to returns structured references",
-              callee is not None and all("frm" in x for x in callee[1]),
-              "no referenced function found" if callee is None else "")
-
-        # xrefs_from a function is whole-body (decomp refs), not just the entry:
-        # find a function that actually calls something.
-        caller = next((f["name"] for f in funcs if f["name"] == "main"), None)
-        xf = []
-        for name in ([caller] if caller else []) + [f["name"] for f in funcs]:
-            xf = (await c.call("xrefs_from", target=name)).get("result", [])
-            if any(x.get("is_func") for x in xf):
-                caller = name
-                break
-        check("xrefs_from a function lists whole-body callees",
-              isinstance(xf, list) and len(xf) > 1
-              and any(x.get("is_func") for x in xf) and all("to" in x for x in xf),
-              f"caller={caller} n={len(xf)}")
-
-        # --- modal select: open xrefs on the callee, pick the first site --- #
-        if callee is not None:
-            f, xr = callee
-            await c.call("goto", target=f["name"], delay_ms=0)
-            # place the cursor on the function name so xrefs targets it
-            v = (await c.call("view", lines=1))["result"]
-            lt = v["lines"][0]["text"] if v.get("lines") else ""
-            col = lt.find(f["name"])
-            if col >= 0:
-                await c.call("cursor", line=0, col=col + 1)
-            resp = await c.call("xrefs")
-            m = resp.get("result", {}).get("modal") or {}
-            check("xrefs opens the picker with items",
-                  m.get("kind") == "XrefsScreen" and len(m.get("items", [])) > 0,
-                  str(m)[:80])
-            resp = await c.call("select", index=0)
-            check("select follows a picked xref (modal closes, we navigate)",
-                  (resp.get("result", {}).get("modal") is None), str(resp.get("result", {}).get("modal")))
-
-        # --- in-view search ----------------------------------------------- #
-        await c.call("goto", target=target["name"], delay_ms=0)
-        resp = await c.call("search", term="return", delay_ms=0)
-        st = resp.get("result", {})
-        check("search runs without error and returns state",
-              st.get("active") in ("decomp", "disasm"), str(st.get("active")))
-        await c.call("keys", keys=["escape"])
-
-        # colored screen export (for an out-of-band web viewer)
-        resp = await c.call("screen", format="html")
-        html = resp.get("result", {})
-        check("screen format=html returns an html document",
-              html.get("format") == "html" and "<" in (html.get("text") or ""),
-              str(html.get("format")))
-
-        # single-driver gate: a 2nd connection (while c is open) is refused
-        r2, w2 = await asyncio.open_unix_connection(sock)
-        c2 = Conn(r2, w2)
-        resp2 = await c2.call("ping")
-        check("second concurrent client is refused (single-driver)",
-              resp2.get("error") and "busy" in resp2["error"].get("message", ""),
-              str(resp2))
-        w2.close()
-
-        try:
-            w.close()
-        except Exception:  # noqa: BLE001
-            pass
-
-        # --- ergonomic driver (idatui.drive) over the same socket ---------- #
-        # (c is closed now, so the single-driver gate lets drive connect.)
-        from idatui import drive
-        loop = asyncio.get_running_loop()
-        for cmd in (["where"], ["pc", target["name"]], ["callees", target["name"]],
-                    ["names", "sub_", "3"]):
-            rc = await loop.run_in_executor(
-                None, lambda a=cmd: drive.main(["--sock", sock, *a]))
-            check(f"drive {cmd[0]} runs against the socket", rc == 0, f"rc={rc}")
-
-        # --- graceful quit (last: it tears the app down) ------------------- #
-        r2, w2 = await asyncio.open_unix_connection(sock)
-        c2 = Conn(r2, w2)
-        resp = await c2.call("quit")
-        check("quit acknowledges before shutting down",
-              resp.get("result", {}).get("quitting") is True, str(resp))
-        w2.close()
-        await wait_for(lambda: not app.is_running, pilot.pause, 10)
-        check("quit actually exits the app", not app.is_running, "still running")
-    print(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-def main(argv):
-    db = None
-    it = iter(argv)
-    for a in it:
-        if a == "--db":
-            db = next(it)
-    return asyncio.run(run(db))
-
-
-if __name__ == "__main__":
-    raise SystemExit(main(sys.argv[1:]))
diff --git a/tests/serverctl.sh b/tests/serverctl.sh
deleted file mode 100755
index a5bc098..0000000
--- a/tests/serverctl.sh
+++ /dev/null
@@ -1,65 +0,0 @@
-#!/usr/bin/env bash
-# Control the ida-pro-mcp server for stress testing. Assumes ~/ida-venv + spawn.sh.
-set -u
-REPO="$(cd "$(dirname "$0")/.." && pwd)"
-PORT=8745
-LOG=/tmp/ida-stress-spawn.log
-
-wait_port() {  # wait_port <up|down> <secs>
-  local want="$1" secs="${2:-40}" i
-  for ((i=0; i<secs*2; i++)); do
-    if ss -ltn 2>/dev/null | grep -q ":$PORT "; then
-      [ "$want" = up ] && return 0
-    else
-      [ "$want" = down ] && return 0
-    fi
-    sleep 0.5
-  done
-  return 1
-}
-
-wait_ready() {  # wait until hexrays_ready via the client
-  local secs="${1:-60}" i
-  for ((i=0; i<secs; i++)); do
-    if python3 -c "
-import sys; sys.path.insert(0,'$REPO')
-from idatui.client import IDAClient
-try:
-    c=IDAClient(); c.connect()
-    ss=[s for s in c.list_sessions() if s.session_id]
-    if ss:
-        c.set_db(ss[0].session_id)
-        if c.health().get('hexrays_ready'): print(ss[0].session_id); sys.exit(0)
-except Exception: pass
-sys.exit(1)
-" 2>/dev/null; then return 0; fi
-    sleep 1
-  done
-  return 1
-}
-
-case "${1:-}" in
-  start)
-    cd "$REPO"
-    source ~/ida-venv/bin/activate 2>/dev/null
-    nohup ./spawn.sh >"$LOG" 2>&1 &
-    wait_port up 40 || { echo "PORT_TIMEOUT"; tail -5 "$LOG"; exit 1; }
-    wait_ready 90 || { echo "READY_TIMEOUT"; tail -5 "$LOG"; exit 1; }
-    ;;
-  stop)
-    # Kill the whole tree: uv wrapper, supervisor, worker.
-    pkill -9 -f 'idalib_server' 2>/dev/null
-    pkill -9 -f 'idalib-mcp'    2>/dev/null
-    pkill -9 -f 'uv run idalib' 2>/dev/null
-    wait_port down 20 || { echo "STOP_TIMEOUT"; exit 1; }
-    ;;
-  kill9)
-    # Hard kill only the worker (simulate a crash of the analysis process).
-    pkill -9 -f 'idalib_server' 2>/dev/null
-    ;;
-  ready)
-    wait_ready "${2:-60}"
-    ;;
-  *)
-    echo "usage: $0 {start|stop|kill9|ready [secs]}"; exit 2;;
-esac
diff --git a/tests/smoke_client.py b/tests/smoke_client.py
deleted file mode 100644
index 5466284..0000000
--- a/tests/smoke_client.py
+++ /dev/null
@@ -1,162 +0,0 @@
-#!/usr/bin/env python3
-"""Live smoke test for idatui.client against a running ida-pro-mcp server.
-
-Run with the venv Python while a server is up (see spawn.sh):
-
-    IDA_MCP_DB=<session_id> python3 tests/smoke_client.py
-    # or:  python3 tests/smoke_client.py --db <session_id>
-
-Exercises: handshake+warm latency, happy-path payloads, the full error taxonomy
-(hard tool errors vs soft per-item errors), session resolution, connection-pool
-reuse, and concurrent calls from threads.
-"""
-import concurrent.futures
-import os
-import sys
-import time
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from idatui.client import (  # noqa: E402
-    IDAClient,
-    IDASessionError,
-    IDAToolError,
-)
-
-PASS, FAIL = 0, 0
-
-
-def _query_data(payload):
-    res = payload.get("result", payload) if isinstance(payload, dict) else payload
-    if isinstance(res, list):
-        res = res[0] if res else {}
-    return res.get("data", []) if isinstance(res, dict) else []
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ok   {name}")
-    else:
-        FAIL += 1
-        print(f"  FAIL {name}  {detail}")
-
-
-def main(argv):
-    url = "http://127.0.0.1:8745/mcp"
-    db = os.environ.get("IDA_MCP_DB")
-    it = iter(argv)
-    for a in it:
-        if a == "--url":
-            url = next(it)
-        elif a == "--db":
-            db = next(it)
-
-    ida = IDAClient(url, db=db)
-    ida.connect()
-
-    print("[sessions]")
-    sessions = ida.list_sessions()
-    check("list_sessions returns >=1", len(sessions) >= 1, str(sessions))
-    if db is None:
-        usable = [s for s in sessions if s.session_id]
-        if len(usable) == 1:
-            db = usable[0].session_id
-        else:
-            print(f"  (multiple/zero sessions; pin one with --db) -> {usable}")
-            print("  Options:", ", ".join(f"{s.session_id}={s.filename}" for s in usable))
-            return 2
-    ida.set_db(db)
-    print(f"  using db={db}")
-
-    print("[handshake / latency]")
-    # Warm calls; measure.
-    for _ in range(3):
-        ida.health()
-    times = []
-    for _ in range(20):
-        t = time.time()
-        ida.call("list_funcs", queries=[{"count": 50}])
-        times.append((time.time() - t) * 1e3)
-    times.sort()
-    med = times[len(times) // 2]
-    print(f"  list_funcs x20 warm: min={times[0]:.1f} med={med:.1f} max={times[-1]:.1f} ms")
-    check("warm call under 50ms median", med < 50, f"med={med:.1f}ms")
-
-    print("[happy path]")
-    h = ida.health()
-    check("health is dict", isinstance(h, dict), type(h).__name__)
-    funcs = ida.call("list_funcs", queries=[{"count": 5}])
-    check("list_funcs shape", isinstance(funcs, (list, dict)), type(funcs).__name__)
-    # Pick a real function (don't assume 'main' exists — libraries have none).
-    some = ida.call("list_funcs", queries=[{"filter": "sub_*", "count": 1}])
-    target = _query_data(some)[0]["addr"]
-    dis = ida.call("disasm", addr=target, max_instructions=10)
-    lines = (dis.get("asm") or {}).get("lines") if isinstance(dis, dict) else None
-    check("disasm main has lines", bool(lines), str(dis)[:120])
-    check("disasm line carries addr", bool(lines and "addr" in lines[0]),
-          str(lines[0]) if lines else "no lines")
-
-    print("[error taxonomy]")
-    # Hard tool error: wrong params -> isError true -> IDAToolError
-    try:
-        ida.call("xrefs_to", targets=["main"])
-        check("bad params raises IDAToolError", False, "no exception")
-    except IDAToolError as e:
-        check("bad params raises IDAToolError", True)
-        check("  ...message surfaced", "addrs" in e.message or "param" in e.message.lower(),
-              e.message)
-    # Hard tool error: unknown tool -> isError true
-    try:
-        ida.call("definitely_not_a_tool")
-        check("unknown tool raises IDAToolError", False, "no exception")
-    except IDAToolError:
-        check("unknown tool raises IDAToolError", True)
-    # Soft/per-item error: bad addr to decompile -> isError false -> DATA, not raise
-    try:
-        payload = ida.call("decompile", addr="zzz_nope_addr")
-        soft = isinstance(payload, dict) and payload.get("error")
-        check("soft error returned as data (not raised)", bool(soft), str(payload)[:160])
-    except IDAToolError as e:
-        check("soft error returned as data (not raised)", False, f"raised: {e}")
-
-    print("[session resolution]")
-    tmp = IDAClient(url, db=None)
-    tmp.connect()
-    all_sessions = tmp.list_sessions()
-    usable = [s for s in all_sessions if s.session_id]
-    if len(usable) > 1:
-        try:
-            tmp.resolve_db()
-            check("multi-session resolve raises", False, "no exception")
-        except IDASessionError:
-            check("multi-session resolve raises", True)
-    else:
-        check("single-session auto-resolves", tmp.resolve_db() == usable[0].session_id)
-    tmp.close()
-
-    print("[connection reuse]")
-    # Fire many calls; pool should keep idle conns bounded and all succeed.
-    ok = 0
-    for _ in range(30):
-        if ida.call("list_funcs", queries=[{"count": 1}]) is not None:
-            ok += 1
-    check("30 sequential calls all succeed (pooled)", ok == 30, f"{ok}/30")
-
-    print("[concurrency]")
-    def worker(i):
-        return ida.call("disasm", addr=target, max_instructions=5)
-
-    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
-        results = list(ex.map(worker, range(40)))
-    good = sum(1 for r in results if isinstance(r, dict) and r.get("asm"))
-    check("40 concurrent calls across 8 threads", good == 40, f"{good}/40")
-
-    ida.close()
-    print(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main(sys.argv[1:]))
diff --git a/tests/stress_client.py b/tests/stress_client.py
deleted file mode 100644
index 955700c..0000000
--- a/tests/stress_client.py
+++ /dev/null
@@ -1,290 +0,0 @@
-#!/usr/bin/env python3
-"""Adversarial stress tests for idatui.client.
-
-Kills/respawns the server, forces timeouts, drops connections, and hammers with
-concurrency — asserting the client always fails *cleanly* (no hangs, no
-deadlocks) and recovers where recovery is possible.
-
-    python3 tests/stress_client.py            # run all scenarios
-    python3 tests/stress_client.py timeout    # run one by name
-
-Requires ~/ida-venv + spawn.sh (via tests/serverctl.sh). Leaves the server UP.
-"""
-import concurrent.futures
-import os
-import subprocess
-import sys
-import threading
-import time
-
-HERE = os.path.dirname(os.path.abspath(__file__))
-REPO = os.path.dirname(HERE)
-sys.path.insert(0, REPO)
-
-from idatui.client import (  # noqa: E402
-    IDAClient,
-    IDAConnectionError,
-    IDASessionError,
-    IDATimeoutError,
-    IDAToolError,
-    IDAError,
-)
-
-URL = "http://127.0.0.1:8745/mcp"
-CTL = os.path.join(HERE, "serverctl.sh")
-PASS = FAIL = 0
-
-
-def log(m):
-    print(m, flush=True)
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        log(f"  ok   {name}")
-    else:
-        FAIL += 1
-        log(f"  FAIL {name}   {detail}")
-
-
-def ctl(*args, timeout=180):
-    r = subprocess.run([CTL, *args], capture_output=True, text=True, timeout=timeout)
-    return r.returncode, (r.stdout + r.stderr).strip()
-
-
-def ensure_up():
-    rc, out = ctl("ready", "5")
-    if rc != 0:
-        log("  (server down; starting...)")
-        rc, out = ctl("start")
-        assert rc == 0, f"could not start server: {out}"
-    rc, sid = ctl("ready", "60")
-    return sid.strip().splitlines()[-1] if sid.strip() else None
-
-
-def deadline(fn, secs):
-    """Run fn() in a thread; return (finished_bool, result_or_exc)."""
-    box = {}
-    def run():
-        try:
-            box["r"] = fn()
-        except BaseException as e:  # noqa: BLE001
-            box["e"] = e
-    t = threading.Thread(target=run, daemon=True)
-    t.start()
-    t.join(secs)
-    if t.is_alive():
-        return False, None
-    return True, box.get("e", box.get("r"))
-
-
-# --------------------------------------------------------------------------- #
-def s_connect_refused():
-    """Connecting to a dead port fails fast and cleanly, never hangs."""
-    c = IDAClient("http://127.0.0.1:59999/mcp", db="x", timeout=3.0)
-    fin, res = deadline(c.connect, 8)
-    check("dead-port connect returns (no hang)", fin)
-    check("dead-port raises IDAConnectionError", isinstance(res, IDAConnectionError),
-          f"{type(res).__name__}: {res}")
-
-
-def s_timeout():
-    """A too-tight timeout on the slow decompile raises IDATimeoutError, and the
-    client stays usable for subsequent calls."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=sid)
-    c.connect()
-    # force_recompile clears the hexrays cache so decompile is genuinely slow.
-    try:
-        c.call("force_recompile", addr="main")
-    except IDAError:
-        pass
-    fin, res = deadline(lambda: c.call("decompile", addr="main", timeout=0.001), 10)
-    check("tight-timeout returns (no hang)", fin)
-    check("tight-timeout raises IDATimeoutError", isinstance(res, IDATimeoutError),
-          f"{type(res).__name__}: {res}")
-    # Client must still work afterwards (pool not poisoned).
-    fin2, res2 = deadline(lambda: c.call("list_funcs", queries=[{"count": 1}]), 10)
-    check("client usable after timeout", fin2 and not isinstance(res2, Exception),
-          f"{type(res2).__name__}: {res2}")
-    c.close()
-
-
-def s_worker_crash():
-    """Hard-kill the analysis worker mid-use: calls must raise cleanly (no hang),
-    and after a full restart the client recovers by re-resolving the session."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=sid)
-    c.connect()
-    check("baseline call ok", isinstance(c.call("list_funcs", queries=[{"count": 1}]), (list, dict)))
-
-    ctl("kill9")  # kill only the worker; supervisor may stay up
-    time.sleep(1.0)
-
-    def probe():
-        try:
-            return c.call("list_funcs", queries=[{"count": 1}])
-        except IDAError as e:
-            return e
-    fin, res = deadline(probe, 15)
-    check("post-crash call returns (no hang)", fin, "call hung after worker kill")
-    check("post-crash raises IDAError (clean)", isinstance(res, IDAError) or res is not None,
-          f"{type(res).__name__}: {res}")
-
-    # Full restart and recovery.
-    ctl("stop")
-    rc, _ = ctl("start")
-    check("server restart ok", rc == 0)
-    newsid = ensure_up()
-    # A fresh client should just work.
-    c2 = IDAClient(URL, db=None)
-    c2.connect()
-    fin3, res3 = deadline(lambda: (c2.set_db(c2.resolve_db()), c2.health())[1], 20)
-    check("fresh client recovers after restart", fin3 and isinstance(res3, dict),
-          f"{type(res3).__name__}: {res3}")
-    check("session id stable across restart" if newsid == sid else "session id changed (expected-ok)",
-          True, f"{sid} -> {newsid}")
-    c.close(); c2.close()
-
-
-def s_full_restart_same_client():
-    """Keep ONE auto-injected client across a full server restart. The FIRST
-    naive call after restart must transparently self-heal the stale db pin (no
-    internal poking, no manual re-resolve)."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=None)  # auto-resolve -> auto_recover eligible
-    c.connect()
-    old = c.resolve_db()
-    check("pre-restart call ok", isinstance(c.health(), dict))
-
-    ctl("stop")
-    # While down: calls must fail cleanly, fast.
-    fin, res = deadline(lambda: c.health(), 8)
-    check("while-down call returns (no hang)", fin)
-    check("while-down raises IDAError", isinstance(res, IDAError), f"{type(res).__name__}: {res}")
-
-    ctl("start")
-    ensure_up()
-    # NAIVE call: no internal poking. Auto-recovery must kick in on the stale
-    # "Session not found" and retry against the fresh session.
-    fin2, res2 = deadline(lambda: c.health(), 25)
-    check("naive call auto-recovers after restart", fin2 and isinstance(res2, dict),
-          f"{type(res2).__name__}: {res2}")
-    check("db pin was re-resolved to new session", c.db is not None and c.db != old,
-          f"old={old} new={c.db}")
-    c.close()
-
-
-def s_no_autoswitch_when_explicit():
-    """With auto_recover disabled (or an explicitly pinned db), a stale session
-    must NOT be silently switched — it raises so the caller stays in control."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=sid, auto_recover_session=False)
-    c.connect()
-    check("pre-restart call ok", isinstance(c.health(), dict))
-    ctl("stop"); ctl("start"); newsid = ensure_up()
-    fin, res = deadline(lambda: c.health(), 15)
-    check("explicit-pin call returns (no hang)", fin)
-    check("explicit-pin raises IDAToolError (no silent switch)",
-          isinstance(res, IDAToolError), f"{type(res).__name__}: {res}")
-    check("db pin unchanged when recovery disabled", c.db == sid, f"{c.db} vs {sid}")
-    c.close()
-
-
-def s_concurrency_high():
-    """Heavy concurrency: 300 calls over 16 threads, all succeed, no deadlock."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=sid, pool_size=8)
-    c.connect()
-    N = 300
-    errors = []
-    def work(i):
-        try:
-            tool = ("list_funcs", "disasm")[i % 2]
-            if tool == "list_funcs":
-                return bool(c.call("list_funcs", queries=[{"count": 3}]))
-            return bool(c.call("disasm", addr="main", max_instructions=5))
-        except IDAError as e:
-            errors.append(e); return False
-    def run_all():
-        with concurrent.futures.ThreadPoolExecutor(max_workers=16) as ex:
-            return sum(ex.map(work, range(N)))
-    fin, ok = deadline(run_all, 60)
-    check("300/16 concurrent finished (no deadlock)", fin, "pool deadlocked")
-    check("300/16 all succeeded", ok == N, f"{ok}/{N}, errs={errors[:3]}")
-    c.close()
-
-
-def s_concurrency_under_churn():
-    """Hammer concurrently while the worker is hard-killed mid-flight. No hang,
-    no deadlock; errors are clean; after respawn calls succeed again."""
-    sid = ensure_up()
-    c = IDAClient(URL, db=sid, pool_size=8)
-    c.connect()
-    stop = threading.Event()
-    stats = {"ok": 0, "err": 0, "weird": 0}
-    lock = threading.Lock()
-    def spinner():
-        while not stop.is_set():
-            try:
-                c.call("list_funcs", queries=[{"count": 1}])
-                with lock: stats["ok"] += 1
-            except IDAError:
-                with lock: stats["err"] += 1
-            except Exception:  # noqa: BLE001  -- any non-IDAError is a bug
-                with lock: stats["weird"] += 1
-            time.sleep(0.01)
-    threads = [threading.Thread(target=spinner, daemon=True) for _ in range(12)]
-    for t in threads: t.start()
-    time.sleep(1.0)
-    ctl("kill9")            # crash the worker under load
-    time.sleep(2.0)
-    stop.set()
-    for t in threads: t.join(10)
-    alive = [t for t in threads if t.is_alive()]
-    check("no spinner thread hung on worker kill", not alive, f"{len(alive)} stuck")
-    check("no non-IDAError leaked during churn", stats["weird"] == 0, str(stats))
-    log(f"    churn stats: {stats}")
-    # Recover.
-    ctl("stop"); ctl("start"); ensure_up()
-    c.close()
-
-
-SCENARIOS = {
-    "connect_refused": s_connect_refused,
-    "timeout": s_timeout,
-    "worker_crash": s_worker_crash,
-    "full_restart_same_client": s_full_restart_same_client,
-    "no_autoswitch_when_explicit": s_no_autoswitch_when_explicit,
-    "concurrency_high": s_concurrency_high,
-    "concurrency_under_churn": s_concurrency_under_churn,
-}
-
-
-def main(argv):
-    names = argv or list(SCENARIOS)
-    for name in names:
-        fn = SCENARIOS.get(name)
-        if not fn:
-            log(f"unknown scenario: {name} (have: {', '.join(SCENARIOS)})")
-            return 2
-        log(f"\n[{name}]")
-        t0 = time.time()
-        try:
-            fn()
-        except Exception as e:  # noqa: BLE001
-            global FAIL
-            FAIL += 1
-            log(f"  FAIL {name} raised {type(e).__name__}: {e}")
-        log(f"  ({time.time() - t0:.1f}s)")
-    log("\n== ensuring server is UP for subsequent work ==")
-    sid = ensure_up()
-    log(f"  server ready, db={sid}")
-    log(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main(sys.argv[1:]))
diff --git a/tests/stress_paging.py b/tests/stress_paging.py
deleted file mode 100644
index b821354..0000000
--- a/tests/stress_paging.py
+++ /dev/null
@@ -1,150 +0,0 @@
-#!/usr/bin/env python3
-"""Stress the 'many lines of text' problem: paging huge listings/functions.
-
-Validates the windowing strategy the TUI will use so we never hand a widget more
-than a viewport-sized slice. Prints concise stats only.
-
-    python3 tests/stress_paging.py --db <session_id>
-"""
-import os
-import statistics
-import sys
-import time
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from idatui.client import IDAClient  # noqa: E402
-
-URL = "http://127.0.0.1:8745/mcp"
-# Verified server cap: list_* honors count up to ~700, then silently collapses
-# to a 10-item default. Clamp with margin.
-SAFE_PAGE = 500
-
-
-def ms(fn):
-    t = time.time()
-    r = fn()
-    return (time.time() - t) * 1e3, r
-
-
-def q0(payload):
-    """Unwrap a *_query/list_* payload: {'result':[{'data':[...],'next_offset':N}]}."""
-    res = payload.get("result", payload)
-    if isinstance(res, list):
-        res = res[0] if res else {}
-    return res.get("data", []), res.get("next_offset")
-
-
-def page_all_funcs(c, page=SAFE_PAGE):
-    """Enumerate the entire function list correctly: advance by len(data), NOT by
-    next_offset (which is just offset+count and skips over the per-call cap)."""
-    funcs, offset, pages, t0 = [], 0, 0, time.time()
-    while True:
-        payload = c.call("list_funcs", queries=[{"offset": offset, "count": page}])
-        data, _ = q0(payload)
-        funcs.extend(data)
-        pages += 1
-        if len(data) < page:  # short page => reached the end
-            break
-        offset += len(data)
-    return funcs, pages, (time.time() - t0) * 1e3
-
-
-def disasm_meta(c, addr):
-    # total_instructions is a TOP-LEVEL field, not under 'asm'.
-    payload = c.call("disasm", addr=addr, max_instructions=1, include_total=True)
-    return payload.get("total_instructions", payload.get("instruction_count"))
-
-
-def main(argv):
-    db = None
-    it = iter(argv)
-    for a in it:
-        if a == "--db":
-            db = next(it)
-    c = IDAClient(URL, db=db)
-    c.connect()
-    if db is None:
-        c.set_db(c.resolve_db())
-    print(f"db={c.db}  ({c.health().get('module')})")
-
-    # ---- 1. Full function list via cursor pagination ------------------- #
-    print("\n[1] page entire function list (cursor)")
-    funcs, pages, total_ms = page_all_funcs(c, page=SAFE_PAGE)
-    n = len(funcs)
-    print(f"  {n} functions in {pages} pages, {total_ms:.0f}ms "
-          f"({total_ms / max(n,1):.3f}ms/func)")
-
-    def sz(f):
-        s = f["size"]
-        return int(s, 16) if isinstance(s, str) else s
-    biggest = sorted(funcs, key=sz, reverse=True)[:5]
-    print("  biggest by bytes:")
-    for f in biggest:
-        print(f"    {f['addr']:>12}  {sz(f):#8x}  {f['name']}")
-
-    # ---- 2. Instruction counts of the biggest -------------------------- #
-    print("\n[2] instruction totals of biggest funcs")
-    fattest = None
-    fattest_n = 0
-    for f in biggest:
-        dt, total = ms(lambda f=f: disasm_meta(c, f["addr"]))
-        print(f"    {f['name']:<28} {str(total):>8} insns  (meta {dt:.1f}ms)")
-        if isinstance(total, int) and total > fattest_n:
-            fattest, fattest_n = f, total
-
-    # ---- 3. Windowed paging INTO the fattest function ------------------ #
-    print(f"\n[3] window-scroll fattest func {fattest['name']} ({fattest_n} insns)")
-    WIN = 60  # a viewport
-    offsets = list(range(0, max(fattest_n - WIN, 1), max((fattest_n // 12), 1)))
-    times = []
-    for off in offsets:
-        dt, payload = ms(lambda off=off: c.call(
-            "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN))
-        lines = payload.get("asm", {}).get("lines", [])
-        times.append(dt)
-        assert len(lines) <= WIN, f"got {len(lines)} > window {WIN}"
-    print(f"  {len(offsets)} windowed reads @win={WIN}: "
-          f"min={min(times):.1f} med={statistics.median(times):.1f} "
-          f"max={max(times):.1f} ms  (payload capped to <= {WIN} lines)")
-
-    # ---- 3b. O(offset) latency curve (the key hazard) ------------------ #
-    print("\n[3b] disasm offset-latency curve (offset paging is O(offset))")
-    for off in [0, 1000, 5000, 20000, min(50000, fattest_n - WIN)]:
-        if off < 0:
-            continue
-        dt, _ = ms(lambda off=off: c.call(
-            "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN))
-        print(f"    offset={off:>6}: {dt:6.1f}ms")
-
-    # ---- 4. Simulated 'hold page-down' throughput ---------------------- #
-    print("\n[4] rapid sequential scroll (hold page-down)")
-    reads, t0, off = 0, time.time(), 0
-    while time.time() - t0 < 2.0:
-        c.call("disasm", addr=fattest["addr"], offset=off, max_instructions=WIN)
-        off = (off + WIN) % max(fattest_n - WIN, 1)
-        reads += 1
-    dur = time.time() - t0
-    print(f"  {reads} windowed reads in {dur:.1f}s = {reads / dur:.0f} reads/s "
-          f"(~{reads / dur * WIN:.0f} lines/s)")
-
-    # ---- 5. Decompile: truncation AND hard-failure on monsters --------- #
-    print("\n[5] decompile on fattest func (may fail on huge funcs)")
-    dt, payload = ms(lambda: c.call("decompile", addr=fattest["addr"]))
-    code = payload.get("code") if isinstance(payload, dict) else None
-    if not code:
-        err = payload.get("error") if isinstance(payload, dict) else payload
-        print(f"  decompile {dt:.0f}ms -> FAILED (soft error, not raised): {err}")
-    else:
-        marker = "chars total]" in code[-40:]
-        print(f"  decompile {dt:.0f}ms, returned {len(code)} chars, "
-              f"server-truncated={marker}")
-        if marker:
-            print(f"    tail: ...{code[-60:]!r}")
-
-    c.close()
-    print("\nOK")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main(sys.argv[1:]))
diff --git a/tests/test_domain.py b/tests/test_domain.py
deleted file mode 100644
index 24cb781..0000000
--- a/tests/test_domain.py
+++ /dev/null
@@ -1,191 +0,0 @@
-#!/usr/bin/env python3
-"""Validate the domain/paging layer against a real, large binary.
-
-Run with a session open on a big binary (e.g. libcrypto.so.3, 10k funcs):
-
-    python3 tests/test_domain.py --db <session_id>
-
-Checks: correct full pagination (advance by len, not next_offset), viewport
-slicing across block boundaries, window caching (revisit is instant), prefetch
-warming, cached instruction totals, decompile success + hard-failure handling,
-and address resolution.
-"""
-import os
-import sys
-import time
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from idatui.client import IDAClient  # noqa: E402
-from idatui.domain import DISASM_BLOCK, LIST_PAGE, Program  # noqa: E402
-
-URL = "http://127.0.0.1:8745/mcp"
-PASS = FAIL = 0
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ok   {name}")
-    else:
-        FAIL += 1
-        print(f"  FAIL {name}   {detail}")
-
-
-def ms(fn):
-    t = time.time()
-    r = fn()
-    return (time.time() - t) * 1e3, r
-
-
-def main(argv):
-    db = None
-    it = iter(argv)
-    for a in it:
-        if a == "--db":
-            db = next(it)
-    c = IDAClient(URL, db=db)
-    c.connect()
-    if db is None:
-        c.set_db(c.resolve_db())
-    prog = Program(c)
-    module = c.health().get("module")
-    total_funcs = c.call("survey_binary").get("statistics", {}).get("total_functions")
-    print(f"db={c.db} module={module} survey_total_functions={total_funcs}")
-
-    # ---- function index: full enumeration correctness ------------------ #
-    print("\n[function index]")
-    idx = prog.functions()
-    dt, _ = ms(lambda: idx.load_all())
-    n = len(idx)
-    check("enumerated all functions (matches survey)", n == total_funcs,
-          f"got {n} vs survey {total_funcs}")
-    print(f"    loaded {n} funcs in {dt:.0f}ms ({dt / max(n,1):.3f}ms/func), "
-          f"page={LIST_PAGE}")
-    # no duplicates, monotonic-ish uniqueness by addr
-    addrs = [idx.get(i).addr for i in range(min(n, 3000))]
-    check("no duplicate addrs in first 3000", len(addrs) == len(set(addrs)))
-    # viewport slice (adapt to binary size)
-    wstart = min(1000, max(n - 50, 0))
-    wlen = min(50, n - wstart)
-    w = idx.window(wstart, wlen)
-    check(f"window({wstart},{wlen}) returns {wlen}", len(w) == wlen, str(len(w)))
-
-    # ---- filtered index uses server-side glob -------------------------- #
-    print("\n[filtered index]")
-    sub = prog.functions(filter="sub_*")
-    sub.ensure(10)
-    check("filter sub_* yields sub_ names", all(f.name.startswith("sub_") for f in sub.window(0, 10)),
-          str([f.name for f in sub.window(0, 5)]))
-
-    # ---- pick the fattest function for disasm stress ------------------- #
-    print("\n[disasm windowing]")
-    fattest = max((idx.get(i) for i in range(n)), key=lambda f: f.size)
-    dm = prog.disasm(fattest.addr, fattest.name)
-    dt, total = ms(dm.total)
-    check("total() returns positive count", total > 0, str(total))
-    dt2, total2 = ms(dm.total)
-    check("total() cached (2nd call ~instant)", dt2 < dt / 2 + 1, f"{dt:.0f}ms -> {dt2:.1f}ms")
-    print(f"    fattest {fattest.name}: {total} insns, total() {dt:.0f}ms then {dt2:.1f}ms")
-
-    # viewport across a block boundary
-    start = DISASM_BLOCK - 5
-    win = dm.lines(start, 60, prefetch=False)
-    check("viewport spans block boundary, right length",
-          len(win) == min(60, max(total - start, 0)), f"got {len(win)}")
-    # addresses strictly increasing and contiguous slice
-    eas = [ln.ea for ln in win]
-    check("viewport addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])),
-          str(eas[:4]))
-
-    # deep window: first slow (O(offset)), revisit instant (cached)
-    if total > DISASM_BLOCK * 4:
-        deep = (total // DISASM_BLOCK - 1) * DISASM_BLOCK
-        dt_cold, a = ms(lambda: dm.lines(deep, 60, prefetch=False))
-        dt_warm, b = ms(lambda: dm.lines(deep, 60, prefetch=False))
-        check("deep window revisit is cached/instant", dt_warm < dt_cold / 2 + 1,
-              f"cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms")
-        check("cached window identical", [l.ea for l in a] == [l.ea for l in b])
-        print(f"    deep@{deep}: cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms")
-
-    # prefetch warms the next block
-    print("\n[prefetch]")
-    dm2 = prog.disasm(fattest.addr + 0)  # same model (cached by ea)
-    fresh = prog.disasm(idx.get(0).addr, idx.get(0).name)
-    fresh.lines(0, 60, prefetch=True)  # should prefetch block 1
-    time.sleep(0.3)
-    check("prefetch warmed a neighbor block", fresh.cached_blocks() >= 2,
-          f"cached_blocks={fresh.cached_blocks()}")
-
-    # ---- decompile: success + hard-failure ----------------------------- #
-    print("\n[decompile]")
-    # a small function likely decompiles
-    small = min((idx.get(i) for i in range(n)), key=lambda f: f.size if f.size > 4 else 1 << 30)
-    d_small = prog.decompile(small.addr)
-    check("small func decompiles or fails cleanly", isinstance(d_small.failed, bool))
-    dt_c, _ = ms(lambda: prog.decompile(small.addr))
-    check("decompile cached (2nd ~instant)", dt_c < 5, f"{dt_c:.1f}ms")
-    # the monster should hard-fail as a soft error (not raise)
-    d_big = prog.decompile(fattest.addr)
-    check("monster decompile handled (failed flag, no raise)",
-          d_big.failed or d_big.code is not None,
-          f"failed={d_big.failed} err={d_big.error}")
-    print(f"    small {small.name}: failed={d_small.failed} "
-          f"trunc={d_small.truncated} chars={d_small.total_chars}")
-    print(f"    monster {fattest.name}: failed={d_big.failed} err={d_big.error}")
-
-    # ---- resolve ------------------------------------------------------- #
-    print("\n[resolve]")
-    check("resolve hex", prog.resolve(hex(fattest.addr)) == fattest.addr)
-    check("resolve int passthrough", prog.resolve(fattest.addr) == fattest.addr)
-    named = next((idx.get(i) for i in range(n) if not idx.get(i).name.startswith("sub_")), None)
-    if named:
-        try:
-            r = prog.resolve(named.name)
-            check("resolve symbol name", r == named.addr, f"{hex(r)} vs {hex(named.addr)} ({named.name})")
-        except KeyError as e:
-            check("resolve symbol name", False, str(e))
-
-    # ---- flat listing (code + data + undefined heads) ------------------ #
-    print("\n[listing]")
-    seg = prog.segment_bounds(fattest.addr)
-    check("segment_bounds finds the .text segment", seg is not None and seg[0] <= fattest.addr < seg[1],
-          str(seg))
-    lm = prog.listing(fattest.addr)
-    check("listing() returns a model for a mapped address", lm is not None)
-    if lm is not None:
-        lm.ensure(20)
-        w = lm.window(0, 20)
-        check("listing window returns heads", len(w) == 20, str(len(w)))
-        eas = [h.ea for h in w]
-        check("listing head addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])),
-              str(eas[:4]))
-        check("listing heads carry a kind", all(h.kind in ("code", "data", "unknown") for h in w),
-              str({h.kind for h in w}))
-        check("listing head sizes positive", all(h.size >= 1 for h in w),
-              str([h.size for h in w[:6]]))
-        # random access to a mid-segment address lands on the containing head
-        mid = w[10].ea
-        li = lm.ensure_ea(mid)
-        check("ensure_ea lands on the exact head for a head address",
-              li >= 0 and lm.get(li).ea == mid, f"idx={li}")
-        # a mid-item byte resolves to its containing head
-        if w[4].size > 1:
-            inside = w[4].ea + 1
-            li2 = lm.ensure_ea(inside)
-            check("ensure_ea resolves a mid-item byte to its head",
-                  li2 >= 0 and lm.get(li2).ea == w[4].ea, f"idx={li2} ea={w[4].ea:#x}")
-        dt_cold, _ = ms(lambda: lm.window(0, 20))
-        check("listing window revisit is cached/instant", dt_cold < 5, f"{dt_cold:.1f}ms")
-        dt_all, _ = ms(lambda: lm.load_all())
-        check("listing load_all completes the segment", lm.complete and len(lm) > 20,
-              f"n={len(lm)} complete={lm.complete}")
-        print(f"    {seg[2]}: {len(lm)} heads walked in {dt_all:.0f}ms")
-
-    prog.close()
-    print(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main(sys.argv[1:]))
diff --git a/tests/test_keepalive.py b/tests/test_keepalive.py
deleted file mode 100644
index 2910e1a..0000000
--- a/tests/test_keepalive.py
+++ /dev/null
@@ -1,76 +0,0 @@
-#!/usr/bin/env python3
-"""Prove that an interactive session can 'just chill' without the worker idling
-out. Uses a deliberately short worker TTL to keep the test fast.
-
-    python3 tests/test_keepalive.py
-
-Needs ~/ida-venv + a running server (tests/serverctl.sh) and a writable target.
-"""
-import os
-import sys
-import time
-
-HERE = os.path.dirname(os.path.abspath(__file__))
-REPO = os.path.dirname(HERE)
-sys.path.insert(0, REPO)
-from idatui.client import IDAClient, IDAToolError  # noqa: E402
-
-PASS = FAIL = 0
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ok   {name}")
-    else:
-        FAIL += 1
-        print(f"  FAIL {name}   {detail}")
-
-
-def alive(c, sid):
-    try:
-        c.set_db(sid)
-        c.health()
-        return True
-    except IDAToolError:
-        return False
-
-
-def main():
-    target = os.path.join(REPO, "targets", "ls_ttl")
-    if not os.path.exists(target):
-        src = os.path.join(REPO, "bin", "ls")
-        os.makedirs(os.path.dirname(target), exist_ok=True)
-        import shutil
-        shutil.copy(src, target)
-
-    c = IDAClient(timeout=300)
-    c.connect()
-    SHORT = 12  # worker self-exits after ~12s idle unless kept alive
-
-    print("[heartbeat keeps a short-TTL worker alive]")
-    sid = c.call("idb_open", input_path=target, idle_ttl_sec=SHORT)["session"]["session_id"]
-    c.set_db(sid)
-    ka = c.keepalive(interval=4.0).start()
-    time.sleep(SHORT * 2 + 2)  # idle well past the TTL, but heartbeat is beating
-    ok = alive(c, sid)
-    check("alive past 2x TTL with heartbeat", ok, f"beats={ka.beats}")
-    check("heartbeat actually beat", ka.beats >= 4, f"beats={ka.beats}")
-    check("heartbeat had no failures", ka.failures == 0, f"failures={ka.failures}")
-    ka.stop()
-
-    print("\n[bump_idle_ttl makes it effectively immortal]")
-    sid = c.call("idb_open", input_path=target, idle_ttl_sec=SHORT)["session"]["session_id"]
-    c.set_db(sid)
-    c.bump_idle_ttl()  # ~1e9 seconds
-    time.sleep(SHORT * 2 + 6)  # zero requests during this window
-    check("alive past 2x TTL after bump, zero requests", alive(c, sid))
-
-    c.close()
-    print(f"\n{PASS} passed, {FAIL} failed")
-    return 1 if FAIL else 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())