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.

7 files changed, 31 insertions, 806 deletions

diff --git a/idatui/__init__.py b/idatui/__init__.py
index 92d8b5d..2cbdde8 100644
--- a/idatui/__init__.py
+++ b/idatui/__init__.py
@@ -1,4 +1,5 @@
-"""idatui — a minimal keyboard-first TUI for IDA Pro over ida-pro-mcp (idalib)."""
+"""idatui — a minimal keyboard-first TUI for IDA Pro, driving idalib via a
+private unix-socket worker (idatui.worker / WorkerClient)."""
 
 from .errors import (
     IDAError,
@@ -10,7 +11,6 @@ from .errors import (
     IDASessionError,
     Session,
 )
-from .client import IDAClient, KeepAlive  # deprecated mcp transport
 from .domain import (
     Program,
     FunctionIndex,
@@ -35,7 +35,6 @@ __all__ = [
     "Decompilation",
     "LIST_PAGE",
     "DISASM_BLOCK",
-    "IDAClient",
     "IDAError",
     "IDAConnectionError",
     "IDATimeoutError",
@@ -44,5 +43,4 @@ __all__ = [
     "IDAToolError",
     "IDASessionError",
     "Session",
-    "KeepAlive",
 ]
diff --git a/idatui/__main__.py b/idatui/__main__.py
index 29f5cf4..0090ace 100644
--- a/idatui/__main__.py
+++ b/idatui/__main__.py
@@ -1,6 +1,6 @@
-"""``python -m idatui`` -> client self-check (until the TUI app lands)."""
+"""``python -m idatui`` -> the one-shot launcher (open a binary in the TUI)."""
 import sys
 
-from .client import _main
+from .launch import main
 
-raise SystemExit(_main(sys.argv[1:]))
+raise SystemExit(main(sys.argv[1:]))
diff --git a/idatui/client.py b/idatui/client.py
deleted file mode 100644
index 2f74cbe..0000000
--- a/idatui/client.py
+++ /dev/null
@@ -1,636 +0,0 @@
-"""Persistent, thread-safe client for the ida-pro-mcp (idalib) MCP server.
-
-DEPRECATED. The default backend is now our own idalib worker (idatui.worker +
-idatui.worker_client.WorkerClient), which talks a unix socket instead of HTTP and
-is ~50-100x cheaper per call. This module (and the whole ida-pro-mcp
-supervisor/HTTP path: server/patch_server.py, spawn.sh, launch.py's server
-plumbing) is kept only for `--backend mcp` / --db attach and will be removed once
-the worker is battle-tested. Prefer WorkerClient for new code.
-
-This is the foundation the whole TUI stands on. Unlike the throwaway CLI in the
-ida-mcp skill (which re-does the MCP handshake and spins a fresh socket on every
-call), this client:
-
-  * Performs the MCP handshake exactly once and keeps the session warm
-    (measured: ~7ms/call warm vs ~60ms cold).
-  * Reuses TCP connections via a small keep-alive pool (no socket churn over a
-    long TUI session) while still allowing concurrent calls from worker threads.
-  * Has a precise, *grounded* error taxonomy (see below) so callers can tell
-    apart transport failures, protocol errors, hard tool errors, and soft
-    per-item "not found" results.
-  * Recovers automatically from an expired server session (404 -> re-handshake
-    -> retry once) and from transient transport hiccups (bounded retries).
-  * Auto-injects the mandatory ``database=<session_id>`` argument, resolving a
-    single open session automatically and refusing to guess when several are
-    open.
-
-Error taxonomy (verified against the live server, 2026-07):
-
-  IDAConnectionError  transport could not be established / was lost
-  IDATimeoutError     a request exceeded its deadline
-  IDAProtocolError    malformed / unexpected HTTP or JSON-RPC framing
-  IDARPCError         JSON-RPC ``error`` object in the envelope
-  IDAToolError        tool returned ``result.isError == true`` (hard failure:
-                      bad params, unknown tool, missing database, ...)
-  IDASessionError     no session / multiple sessions and none pinned
-
-Note the deliberate *non*-error: a tool that returns ``isError == false`` with an
-``error`` field inside its payload (e.g. ``decompile`` on an unknown address, or
-per-item failures in a batch tool) is treated as *data*, not an exception. The
-caller inspects the payload. Raising on those would break every batch tool.
-
-The client is stdlib-only (``http.client`` / ``urllib.parse``).
-"""
-
-from __future__ import annotations
-
-import http.client
-import json
-import socket
-import threading
-import time
-from collections import deque
-from dataclasses import dataclass
-from typing import Any
-from urllib.parse import urlsplit
-
-DEFAULT_URL = "http://127.0.0.1:8745/mcp"
-PROTOCOL_VERSION = "2025-06-18"
-
-# Tools that must NOT receive an injected ``database`` argument.
-SESSION_AGNOSTIC_TOOLS = frozenset({"idb_list", "idb_open", "int_convert"})
-
-# Substrings (lowercased) that identify a stale/invalid IDB session in a tool
-# error message. Verified against the live server: after a server restart the
-# old session id is gone and every call fails with "Session not found: <id>".
-_STALE_SESSION_MARKERS = ("session not found", "database is required")
-
-
-# The error hierarchy and Session model now live in errors.py (transport-
-# agnostic, shared with the idalib worker path); re-exported here so the
-# deprecated mcp tooling and the stress tests keep importing them from client.
-from .errors import (  # noqa: E402,F401
-    IDAError,
-    IDAConnectionError,
-    IDATimeoutError,
-    IDAProtocolError,
-    IDARPCError,
-    IDAToolError,
-    IDASessionError,
-    Session,
-)
-
-
-# --------------------------------------------------------------------------- #
-# Transport: a tiny keep-alive connection pool over http.client
-# --------------------------------------------------------------------------- #
-class _Transport:
-    """Keep-alive HTTP/1.1 pool. Thread-safe. One request == one pooled conn.
-
-    Connections are checked out, used for exactly one request/response, then
-    returned to the pool if still healthy. On a dropped/broken connection the
-    request is retried on a fresh connection (bounded by ``max_retries``).
-    """
-
-    def __init__(self, url: str, *, max_retries: int = 2, pool_size: int = 8):
-        parts = urlsplit(url)
-        if parts.scheme not in ("http", "https"):
-            raise IDAConnectionError(f"unsupported scheme: {parts.scheme!r}")
-        self._https = parts.scheme == "https"
-        self._host = parts.hostname or "127.0.0.1"
-        self._port = parts.port or (443 if self._https else 80)
-        self._path = parts.path or "/"
-        self._max_retries = max_retries
-        self._pool_size = pool_size
-        self._idle: deque[http.client.HTTPConnection] = deque()
-        self._lock = threading.Lock()
-
-    def _new_conn(self, timeout: float) -> http.client.HTTPConnection:
-        cls = http.client.HTTPSConnection if self._https else http.client.HTTPConnection
-        return cls(self._host, self._port, timeout=timeout)
-
-    def _checkout(self, timeout: float) -> http.client.HTTPConnection:
-        with self._lock:
-            while self._idle:
-                conn = self._idle.popleft()
-                # Reuse only if the socket still looks alive.
-                if conn.sock is not None:
-                    conn.timeout = timeout
-                    try:
-                        conn.sock.settimeout(timeout)
-                    except OSError:
-                        try:
-                            conn.close()
-                        except OSError:
-                            pass
-                        continue
-                    return conn
-        return self._new_conn(timeout)
-
-    def _checkin(self, conn: http.client.HTTPConnection) -> None:
-        with self._lock:
-            if len(self._idle) < self._pool_size:
-                self._idle.append(conn)
-                return
-        try:
-            conn.close()
-        except OSError:
-            pass
-
-    def request(
-        self, body: bytes, headers: dict[str, str], timeout: float
-    ) -> tuple[int, dict[str, str], bytes]:
-        """POST ``body``; return (status, response_headers, response_body)."""
-        last_exc: Exception | None = None
-        for attempt in range(self._max_retries + 1):
-            conn = self._checkout(timeout)
-            try:
-                conn.request("POST", self._path, body=body, headers=headers)
-                resp = conn.getresponse()
-                status = resp.status
-                resp_headers = {k.lower(): v for k, v in resp.getheaders()}
-                data = resp.read()  # must fully drain before reuse
-            except socket.timeout as e:
-                self._discard(conn)
-                raise IDATimeoutError(f"request timed out after {timeout}s") from e
-            except (
-                http.client.RemoteDisconnected,
-                http.client.BadStatusLine,
-                ConnectionError,
-                OSError,
-            ) as e:
-                # A stale pooled connection, or the server closed on us. Drop it
-                # and retry on a fresh connection.
-                self._discard(conn)
-                last_exc = e
-                continue
-            else:
-                if resp.will_close:
-                    self._discard(conn)
-                else:
-                    self._checkin(conn)
-                return status, resp_headers, data
-        raise IDAConnectionError(
-            f"transport failed after {self._max_retries + 1} attempts: {last_exc}"
-        ) from last_exc
-
-    def _discard(self, conn: http.client.HTTPConnection) -> None:
-        try:
-            conn.close()
-        except OSError:
-            pass
-
-    def close(self) -> None:
-        with self._lock:
-            while self._idle:
-                self._discard(self._idle.popleft())
-
-
-# --------------------------------------------------------------------------- #
-# The client
-# --------------------------------------------------------------------------- #
-class IDAClient:
-    """A warm, thread-safe handle to one MCP server (and one pinned IDB).
-
-    Typical use::
-
-        with IDAClient(db="4f2223f9") as ida:
-            ida.health()
-            funcs = ida.call("list_funcs", queries=[{"count": 50}])
-            code = ida.call("decompile", addr="main")
-
-    Concurrency: safe to call from multiple threads (Textual workers). Requests
-    run on independent pooled connections; only the id counter and handshake
-    state are lock-guarded, so calls do not serialize on each other.
-    """
-
-    def __init__(
-        self,
-        url: str = DEFAULT_URL,
-        db: str | None = None,
-        *,
-        timeout: float = 30.0,
-        max_retries: int = 2,
-        pool_size: int = 8,
-        client_name: str = "idatui",
-        client_version: str = "0.0.1",
-        auto_recover_session: bool = True,
-    ):
-        self.url = url
-        self._db = db
-        self.timeout = timeout
-        self.auto_recover_session = auto_recover_session
-        self._transport = _Transport(url, max_retries=max_retries, pool_size=pool_size)
-        self._client_info = {"name": client_name, "version": client_version}
-
-        self._rid = 0
-        self._sid: str | None = None
-        self._ready = False
-        self._state_lock = threading.Lock()  # guards _rid, _sid, _ready
-        self._handshake_lock = threading.Lock()  # serializes (re)handshake
-
-    # -- lifecycle --------------------------------------------------------- #
-    def __enter__(self) -> "IDAClient":
-        self.connect()
-        return self
-
-    def __exit__(self, *exc) -> None:
-        self.close()
-
-    def connect(self) -> "IDAClient":
-        """Ensure the MCP handshake has completed (idempotent, thread-safe)."""
-        if self._ready:
-            return self
-        self._handshake()
-        return self
-
-    def close(self) -> None:
-        self._transport.close()
-
-    # -- low-level plumbing ------------------------------------------------ #
-    def _next_id(self) -> int:
-        with self._state_lock:
-            self._rid += 1
-            return self._rid
-
-    def _headers(self) -> dict[str, str]:
-        h = {
-            "Content-Type": "application/json",
-            "Accept": "application/json, text/event-stream",
-        }
-        sid = self._sid
-        if sid:
-            h["Mcp-Session-Id"] = sid
-        return h
-
-    def _handshake(self) -> None:
-        with self._handshake_lock:
-            if self._ready:
-                return
-            with self._state_lock:
-                self._sid = None
-            rid = self._next_id()
-            status, headers, body = self._transport.request(
-                self._encode(
-                    {
-                        "jsonrpc": "2.0",
-                        "id": rid,
-                        "method": "initialize",
-                        "params": {
-                            "protocolVersion": PROTOCOL_VERSION,
-                            "capabilities": {},
-                            "clientInfo": self._client_info,
-                        },
-                    }
-                ),
-                {
-                    "Content-Type": "application/json",
-                    "Accept": "application/json, text/event-stream",
-                },
-                self.timeout,
-            )
-            if status // 100 != 2:
-                raise IDAProtocolError(
-                    f"initialize failed: HTTP {status}: {body[:200]!r}"
-                )
-            sid = headers.get("mcp-session-id")
-            envelope = self._parse_body(body, rid)
-            self._raise_on_rpc_error(envelope)
-            with self._state_lock:
-                self._sid = sid
-            # notifications/initialized is a fire-and-forget notification.
-            self._transport.request(
-                self._encode({"jsonrpc": "2.0", "method": "notifications/initialized"}),
-                self._headers(),
-                self.timeout,
-            )
-            with self._state_lock:
-                self._ready = True
-
-    @staticmethod
-    def _encode(obj: dict) -> bytes:
-        return json.dumps(obj).encode()
-
-    @staticmethod
-    def _parse_body(body: bytes, want_id: int) -> dict:
-        """Extract the JSON-RPC envelope for ``want_id`` from a (possibly SSE) body."""
-        text = body.decode("utf-8", "replace")
-        found: dict | None = None
-        for line in text.splitlines():
-            line = line.strip()
-            if line.startswith("data:"):
-                line = line[5:].strip()
-            if not line:
-                continue
-            try:
-                obj = json.loads(line)
-            except json.JSONDecodeError:
-                continue
-            if isinstance(obj, dict) and obj.get("id") == want_id:
-                found = obj
-        if found is None:
-            raise IDAProtocolError(
-                f"no JSON-RPC response with id={want_id} in body: {text[:200]!r}"
-            )
-        return found
-
-    @staticmethod
-    def _raise_on_rpc_error(envelope: dict) -> None:
-        err = envelope.get("error")
-        if err:
-            raise IDARPCError(
-                code=err.get("code", -1),
-                message=err.get("message", "unknown"),
-                data=err.get("data"),
-            )
-
-    def _rpc(self, method: str, params: dict, *, timeout: float | None = None) -> dict:
-        """Send a JSON-RPC request, recovering once from an expired session."""
-        if not self._ready:
-            self._handshake()
-        to = self.timeout if timeout is None else timeout
-        rid = self._next_id()
-        payload = self._encode(
-            {"jsonrpc": "2.0", "id": rid, "method": method, "params": params}
-        )
-        status, headers, body = self._transport.request(payload, self._headers(), to)
-
-        if status == 404 and self._sid is not None:
-            # Server session expired: re-handshake and retry exactly once.
-            with self._state_lock:
-                self._ready = False
-            self._handshake()
-            rid = self._next_id()
-            payload = self._encode(
-                {"jsonrpc": "2.0", "id": rid, "method": method, "params": params}
-            )
-            status, headers, body = self._transport.request(
-                payload, self._headers(), to
-            )
-
-        if status // 100 != 2:
-            raise IDAProtocolError(f"HTTP {status}: {body[:200]!r}")
-
-        envelope = self._parse_body(body, rid)
-        self._raise_on_rpc_error(envelope)
-        return envelope
-
-    # -- tool calls -------------------------------------------------------- #
-    @staticmethod
-    def _extract_payload(tool: str, result: dict) -> Any:
-        """Turn an MCP ``result`` object into a Python payload, or raise.
-
-        Precedence:
-          1. ``isError == true``            -> IDAToolError (hard failure)
-          2. ``structuredContent`` present  -> return it (already parsed)
-          3. ``content[0].text`` is JSON    -> return parsed JSON
-          4. otherwise                      -> return the raw text string
-        """
-        if result.get("isError"):
-            text = _first_text(result) or "(no message)"
-            raise IDAToolError(tool, text)
-        if "structuredContent" in result:
-            return result["structuredContent"]
-        text = _first_text(result)
-        if text is None:
-            return result
-        try:
-            return json.loads(text)
-        except (json.JSONDecodeError, TypeError):
-            return text
-
-    def _prepare_args(self, tool: str, args: dict) -> tuple[dict, bool]:
-        """Return (arguments, db_was_injected). Never mutates the caller's dict."""
-        prepared = dict(args)
-        if tool in SESSION_AGNOSTIC_TOOLS or "database" in prepared:
-            return prepared, False
-        prepared["database"] = self.resolve_db()
-        return prepared, True
-
-    def call_envelope(self, tool: str, *, timeout: float | None = None, **args) -> dict:
-        """Call ``tool`` and return the full JSON-RPC envelope (for debugging)."""
-        prepared, _ = self._prepare_args(tool, args)
-        return self._rpc(
-            "tools/call", {"name": tool, "arguments": prepared}, timeout=timeout
-        )
-
-    def call(self, tool: str, *, timeout: float | None = None, **args) -> Any:
-        """Call ``tool`` and return its payload (parsed JSON when possible).
-
-        Raises IDAToolError on a hard tool failure. Soft/per-item errors (an
-        ``error`` field with ``isError == false``) are returned as data.
-
-        If ``auto_recover_session`` is set and the db was auto-injected, a stale
-        "Session not found" error (e.g. after a server restart) triggers exactly
-        one transparent recovery: drop the stale pin, re-resolve the session,
-        and retry. A db the caller pinned explicitly is never silently switched.
-        """
-        prepared, injected = self._prepare_args(tool, args)
-        envelope = self._rpc(
-            "tools/call", {"name": tool, "arguments": prepared}, timeout=timeout
-        )
-        try:
-            return self._extract_payload(tool, envelope.get("result", {}))
-        except IDAToolError as e:
-            if not (injected and self.auto_recover_session and _is_stale_session(e)):
-                raise
-            # The pinned IDB session vanished (server restart). Re-resolve the
-            # sole session and retry once. resolve_db() raises IDASessionError
-            # if zero/many sessions exist, so we never guess.
-            self.set_db(None)
-            prepared2, _ = self._prepare_args(tool, args)
-            envelope2 = self._rpc(
-                "tools/call", {"name": tool, "arguments": prepared2}, timeout=timeout
-            )
-            return self._extract_payload(tool, envelope2.get("result", {}))
-
-    # -- session management ------------------------------------------------ #
-    def list_sessions(self) -> list[Session]:
-        result = self._rpc("tools/call", {"name": "idb_list", "arguments": {}})
-        payload = self._extract_payload("idb_list", result.get("result", {}))
-        sessions = payload.get("sessions", []) if isinstance(payload, dict) else []
-        return [Session.from_dict(s) for s in sessions]
-
-    def resolve_db(self) -> str:
-        """Return the pinned session id, auto-resolving a lone open session.
-
-        Raises IDASessionError if none is open, or if several are open and none
-        has been pinned via ``db=`` / :meth:`set_db`.
-        """
-        if self._db:
-            return self._db
-        sessions = self.list_sessions()
-        usable = [s for s in sessions if s.session_id]
-        if len(usable) == 1:
-            self._db = usable[0].session_id
-            return self._db
-        if not usable:
-            raise IDASessionError(
-                "no open IDB session with a usable id; open one with idb_open"
-            )
-        opts = ", ".join(f"{s.session_id} ({s.filename})" for s in usable)
-        raise IDASessionError(
-            f"multiple sessions open; pin one with db=... . Options: {opts}"
-        )
-
-    def set_db(self, db: str | None) -> None:
-        self._db = db
-
-    @property
-    def db(self) -> str | None:
-        return self._db
-
-    def _session_input_path(self, db: str) -> str:
-        for s in self.list_sessions():
-            if s.session_id == db:
-                if not s.input_path:
-                    raise IDASessionError(f"session {db} has no input_path to re-open")
-                return s.input_path
-        raise IDASessionError(f"session {db} not found")
-
-    def bump_idle_ttl(self, idle_ttl_sec: int = 1_000_000_000,
-                      path: str | None = None) -> None:
-        """Raise the worker's idle self-exit TTL so an interactive session never
-        gets reaped while the user is just reading.
-
-        Headless idalib workers self-exit after ``idle_ttl_sec`` (default 600s)
-        with no requests. ``idb_open`` on the already-open path is idempotent
-        (returns the same session) and re-applies the TTL, so this simply opens
-        the pinned session's path with a huge TTL. The default (~31 years) is
-        effectively 'never'.
-        """
-        p = path or self._session_input_path(self.resolve_db())
-        self.call("idb_open", input_path=p, idle_ttl_sec=int(idle_ttl_sec))
-
-    def keepalive(self, interval: float = 120.0) -> "KeepAlive":
-        """Return a (not-yet-started) heartbeat that touches the worker's idle
-        watchdog every ``interval`` seconds. Belt-and-suspenders next to
-        :meth:`bump_idle_ttl`; also covers adopted sessions we didn't open.
-        """
-        return KeepAlive(self, interval=interval)
-
-    # -- convenience ------------------------------------------------------- #
-    def health(self) -> dict:
-        return self.call("server_health")
-
-    def list_tools(self) -> list[tuple[str, str]]:
-        envelope = self._rpc("tools/list", {})
-        out = []
-        for t in envelope.get("result", {}).get("tools", []):
-            desc = (t.get("description") or "").splitlines()
-            out.append((t["name"], desc[0] if desc else ""))
-        return out
-
-    def schema(self, tool: str) -> dict:
-        envelope = self._rpc("tools/list", {})
-        for t in envelope.get("result", {}).get("tools", []):
-            if t["name"] == tool:
-                return t.get("inputSchema", {})
-        raise IDAError(f"tool not found: {tool}")
-
-
-class KeepAlive:
-    """Background heartbeat that keeps an idalib worker from idling out.
-
-    Any forwarded request resets the worker's idle timer, so a periodic cheap
-    ``server_health`` is enough. Failures are swallowed (the next real call will
-    auto-recover); the heartbeat just keeps a chilling TUI's worker alive.
-    """
-
-    def __init__(self, client: "IDAClient", interval: float = 120.0):
-        if interval <= 0:
-            raise ValueError("interval must be > 0")
-        self._client = client
-        self.interval = interval
-        self._stop = threading.Event()
-        self._thread: threading.Thread | None = None
-        self.beats = 0
-        self.failures = 0
-
-    def start(self) -> "KeepAlive":
-        if self._thread is not None:
-            return self
-        self._stop.clear()
-        self._thread = threading.Thread(
-            target=self._run, daemon=True, name="idatui-keepalive"
-        )
-        self._thread.start()
-        return self
-
-    def stop(self) -> None:
-        self._stop.set()
-        t = self._thread
-        self._thread = None
-        if t is not None:
-            t.join(timeout=2.0)
-
-    def __enter__(self) -> "KeepAlive":
-        return self.start()
-
-    def __exit__(self, *exc) -> None:
-        self.stop()
-
-    def _run(self) -> None:
-        while not self._stop.wait(self.interval):
-            try:
-                self._client.health()
-                self.beats += 1
-            except IDAError:
-                self.failures += 1
-
-
-def _is_stale_session(e: IDAToolError) -> bool:
-    msg = e.message.lower()
-    return any(marker in msg for marker in _STALE_SESSION_MARKERS)
-
-
-def _first_text(result: dict) -> str | None:
-    content = result.get("content")
-    if isinstance(content, list):
-        for item in content:
-            if isinstance(item, dict) and item.get("type") == "text":
-                return item.get("text")
-    return None
-
-
-# --------------------------------------------------------------------------- #
-# Tiny self-check CLI:  python -m idatui.client [--db ID] [--url URL] [health]
-# --------------------------------------------------------------------------- #
-def _main(argv: list[str]) -> int:
-    import os
-
-    url = DEFAULT_URL
-    db = os.environ.get("IDA_MCP_DB")
-    rest: list[str] = []
-    it = iter(argv)
-    for a in it:
-        if a == "--url":
-            url = next(it)
-        elif a == "--db":
-            db = next(it)
-        else:
-            rest.append(a)
-
-    ida = IDAClient(url, db=db)
-    try:
-        ida.connect()
-        t0 = time.time()
-        sessions = ida.list_sessions()
-        print(f"sessions ({(time.time() - t0) * 1e3:.1f}ms):")
-        for s in sessions:
-            mark = "*" if s.is_active else " "
-            print(f"  {mark} {s.session_id or '<none>':10} {s.filename}")
-        try:
-            h = ida.health()
-            print("health:", json.dumps(h, indent=1)[:400])
-        except IDASessionError as e:
-            print(f"health: skipped ({e})")
-    finally:
-        ida.close()
-    return 0
-
-
-if __name__ == "__main__":
-    import sys
-
-    raise SystemExit(_main(sys.argv[1:]))
diff --git a/idatui/domain.py b/idatui/domain.py
index 4f72f5b..d7ebf3c 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -32,8 +32,8 @@ from typing import Callable, TYPE_CHECKING
 
 from .errors import IDAToolError
 
-if TYPE_CHECKING:  # type hint only; the runtime client is mcp IDAClient or worker
-    from .client import IDAClient  # noqa: F401
+if TYPE_CHECKING:  # type hint only
+    from .worker_client import WorkerClient  # noqa: F401
 
 # Clamps derived from measured caps (list ~700, disasm ~500). Margin included.
 LIST_PAGE = 500
@@ -797,7 +797,7 @@ class HexModel:
 class Program:
     """The bound analysis session: models, caches, and a small prefetch pool."""
 
-    def __init__(self, client: IDAClient, prefetch_workers: int = 2):
+    def __init__(self, client: "WorkerClient", prefetch_workers: int = 2):
         self.client = client
         self._pool = ThreadPoolExecutor(
             max_workers=prefetch_workers, thread_name_prefix="idatui-prefetch"
diff --git a/idatui/pane.py b/idatui/pane.py
index af81982..e34b822 100644
--- a/idatui/pane.py
+++ b/idatui/pane.py
@@ -8,9 +8,6 @@ then close it — all without a human touching the keyboard.
     python -m idatui.pane spawn --open /abs/path/to/bin
     # -> {"sock": "/run/user/1000/idatui-3f2a.sock", "pane": "%7", "ready": true, ...}
 
-    # or attach to an existing session id
-    python -m idatui.pane spawn --db 80d83396
-
     # drive it (see docs/RPC.md / the idatui-rpc skill)
     python -m idatui.rpcclient --sock <sock> pseudocode target=main
 
@@ -18,9 +15,9 @@ then close it — all without a human touching the keyboard.
     python -m idatui.pane list
     python -m idatui.pane stop --sock <sock>        # graceful quit + kill pane
 
-Requires: running inside tmux, and the ida-pro-mcp supervisor already up
-(./spawn.sh). Uses ~/ida-venv/bin/python for the TUI (needs textual) unless
---python / IDATUI_PYTHON says otherwise.
+Requires: running inside tmux. Each pane spawns its own private idalib worker
+(no shared supervisor). Uses ~/ida-venv/bin/python for the TUI (needs textual)
+unless --python / IDATUI_PYTHON says otherwise.
 """
 from __future__ import annotations
 
@@ -29,14 +26,11 @@ import json
 import os
 import secrets
 import signal
-import socket
 import subprocess
 import sys
 import time
 from typing import Any
-from urllib.parse import urlparse
 
-from .client import DEFAULT_URL
 from .rpcclient import RpcClient, RpcError
 
 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -81,24 +75,17 @@ def _tmux(*args: str) -> str:
 # --------------------------------------------------------------------------- #
 # idalib worker reaping
 #
-# ``pane stop`` kills the TUI pane, but the ida-pro-mcp supervisor does not
-# reliably reap the ``ida_pro_mcp.idalib_server`` worker it forked for that
-# session. Leaked workers accumulate against IDA_MCP_MAX_WORKERS until the next
-# ``spawn`` blocks forever waiting for a free slot (the TUI comes up but never
-# becomes ready). A worker is only *safe* to reap when no idatui pane is live
-# (then every worker is orphaned) — that mirrors the hand workaround
-# `pkill -f ida_pro_mcp.idalib_server` and avoids killing an in-use analyser.
+# ``pane stop`` kills the TUI pane, but a hard-killed pane can leave its private
+# idalib worker (idatui/worker.py) running. A worker is only *safe* to reap when
+# no idatui pane is live (then every worker is orphaned), which avoids killing an
+# in-use analyser.
 # --------------------------------------------------------------------------- #
-_WORKER_PATTERN = r"ida_pro_mcp\.idalib_server"
+_WORKER_PATTERN = r"idatui/worker\.py"
 
 
 def _worker_pids() -> list[int]:
-    """PIDs of the supervisor's per-binary idalib worker processes.
-
-    Matches the worker module invocation only (not the ``idalib-mcp``
-    supervisor, whose command line does not contain the module path), and
-    never our own PID.
-    """
+    """PIDs of our private per-pane idalib worker processes (idatui/worker.py),
+    never our own PID."""
     try:
         out = subprocess.run(["pgrep", "-f", _WORKER_PATTERN],
                              capture_output=True, text=True)
@@ -138,73 +125,19 @@ def _reap_orphan_workers(force: bool = False) -> int:
 
 
 # --------------------------------------------------------------------------- #
-# supervisor (ida-pro-mcp server) — auto-start if down
-# --------------------------------------------------------------------------- #
-def _server_addr(url: str) -> tuple[str, int]:
-    u = urlparse(url)
-    return (u.hostname or "127.0.0.1", u.port or 8745)
-
-
-def _server_up(host: str, port: int, timeout: float = 0.75) -> bool:
-    """Is something listening on host:port? (Cheap TCP probe; the readiness poll
-    that follows catches a half-up server.)"""
-    try:
-        with socket.create_connection((host, port), timeout=timeout):
-            return True
-    except OSError:
-        return False
-
-
-def _ensure_server(host: str, port: int, timeout: float,
-                   detached: bool = True) -> dict[str, Any]:
-    """Make sure the supervisor is up; start ./spawn.sh in a tmux pane if not.
-
-    Only auto-starts a *local* server (can't launch a remote one). spawn.sh binds
-    the port, so starting a duplicate is impossible — the probe guards that.
-    ``IDATUI_SERVER_CMD`` overrides the launch command (used by the tests).
-    """
-    if _server_up(host, port):
-        return {"server_started": False, "server_up": True}
-    if host not in ("127.0.0.1", "localhost", "::1"):
-        return {"server_started": False, "server_up": False,
-                "error": f"server at {host}:{port} is down and not local; "
-                         "cannot auto-start"}
-    cmd_str = os.environ.get("IDATUI_SERVER_CMD", "./spawn.sh")
-    cmd = f"cd {_q(REPO)} && exec {cmd_str}"
-    split = ["split-window", "-v", "-P", "-F", "#{pane_id}"]
-    if detached:
-        split += ["-d"]
-    anchor = os.environ.get("TMUX_PANE")
-    if anchor:
-        split += ["-t", anchor]
-    split.append(cmd)
-    pane = _tmux(*split)
-    deadline = time.time() + timeout
-    while time.time() < deadline:
-        if not _pane_alive(pane):
-            return {"server_started": True, "server_up": False, "server_pane": pane,
-                    "error": "supervisor pane exited during startup (check it)"}
-        if _server_up(host, port):
-            return {"server_started": True, "server_up": True, "server_pane": pane}
-        time.sleep(0.5)
-    return {"server_started": True, "server_up": False, "server_pane": pane,
-            "error": "supervisor did not come up in time"}
-
-
-# --------------------------------------------------------------------------- #
 # spawn
 # --------------------------------------------------------------------------- #
 def spawn(args) -> int:
     if not os.environ.get("TMUX"):
         print("error: not inside tmux (spawn creates a tmux pane)", file=sys.stderr)
         return 2
-    if not args.open and not args.db:
-        print("error: pass --open <binary> or --db <session>", file=sys.stderr)
+    if not args.open:
+        print("error: pass --open <binary>", file=sys.stderr)
         return 2
 
     sock = args.sock or os.path.join(_sockdir(), f"idatui-{secrets.token_hex(3)}.sock")
-    target = os.path.abspath(os.path.expanduser(args.open)) if args.open else args.db
-    if args.open and not os.path.exists(target):
+    target = os.path.abspath(os.path.expanduser(args.open))
+    if not os.path.exists(target):
         print(f"error: no such binary: {target}", file=sys.stderr)
         return 2
 
@@ -216,26 +149,9 @@ def spawn(args) -> int:
         print(f"reaped {reaped} orphaned idalib worker(s) before spawn",
               file=sys.stderr)
 
-    # make sure the ida-pro-mcp supervisor is up (auto-start it if not)
-    srv: dict[str, Any] = {"server_started": False, "server_up": True}
-    if not args.no_ensure_server:
-        host, port = _server_addr(args.url or DEFAULT_URL)
-        srv = _ensure_server(host, port, args.server_timeout)
-        if srv.get("server_started"):
-            print(f"supervisor was down — started it ({srv.get('server_pane')})",
-                  file=sys.stderr)
-        if not srv.get("server_up"):
-            print(json.dumps({"ready": False, **srv}), file=sys.stderr)
-            return 3
-
-    # the command the pane runs: become the TUI so kill-pane kills it cleanly
-    inner = [args.python, "-m", "idatui.tui", "--rpc", sock]
-    if args.open:
-        inner += ["--open", target]
-    else:
-        inner += ["--db", target]
-    if args.url:
-        inner += ["--url", args.url]
+    # the command the pane runs: the launcher spawns a private idalib worker for
+    # this binary and becomes the TUI, so kill-pane tears the whole thing down.
+    inner = [args.python, "-m", "idatui.launch", target, "--rpc", sock]
     cmd = f"cd {REPO!r} && exec " + " ".join(_q(a) for a in inner)
 
     split = ["split-window", "-v" if args.vertical else "-h",
@@ -251,10 +167,7 @@ def spawn(args) -> int:
     pane = _tmux(*split)
 
     row = {"sock": sock, "pane": pane, "target": target,
-           "kind": "open" if args.open else "db", "started": time.time(),
-           "server_started": srv.get("server_started", False)}
-    if srv.get("server_pane"):
-        row["server_pane"] = srv["server_pane"]
+           "kind": "open", "started": time.time()}
     reg = [r for r in _load_registry() if r.get("sock") != sock]
     reg.append(row)
     _save_registry(reg)
@@ -298,8 +211,8 @@ def _wait_ready(sock: str, timeout: float, pane: str,
             why = ("RPC socket not created yet" if not os.path.exists(sock)
                    else "TUI up but analysis not ready")
             print(f"still waiting ({int(time.time() - start)}s): {why}. If this "
-                  f"hangs, the idalib worker may be stuck or IDA_MCP_MAX_WORKERS "
-                  f"is full — try `python -m idatui.pane reap`.", file=sys.stderr)
+                  f"hangs, the idalib worker may be stuck — try "
+                  f"`python -m idatui.pane reap`.", file=sys.stderr)
         time.sleep(0.4)
     last = dict(last)
     last["ready"] = False
@@ -388,15 +301,10 @@ def main(argv: list[str]) -> int:
     sub = p.add_subparsers(dest="cmd", required=True)
 
     sp = sub.add_parser("spawn", help="open a TUI pane and wait until ready")
-    sp.add_argument("--open", metavar="PATH", help="binary to open (dir must be writable)")
-    sp.add_argument("--db", metavar="SESSION", help="attach to an existing session id")
+    sp.add_argument("--open", metavar="PATH", required=True,
+                    help="binary to open (its dir must be writable)")
     sp.add_argument("--sock", help="RPC socket path (default: auto in $XDG_RUNTIME_DIR)")
     sp.add_argument("--python", default=DEFAULT_PY, help=f"python for the TUI ({DEFAULT_PY})")
-    sp.add_argument("--url", help="MCP server URL (default: idatui's default)")
-    sp.add_argument("--no-ensure-server", action="store_true",
-                    help="don't auto-start ./spawn.sh if the supervisor is down")
-    sp.add_argument("--server-timeout", type=float, default=90.0,
-                    help="seconds to wait for an auto-started supervisor")
     sp.add_argument("--vertical", action="store_true", help="split vertically (stacked)")
     sp.add_argument("--size", help="new pane size (tmux -l value, e.g. 60%% or 120)")
     sp.add_argument("--detached", action="store_true", help="don't focus the new pane")
diff --git a/idatui/tui.py b/idatui/tui.py
deleted file mode 100644
index b399ffe..0000000
--- a/idatui/tui.py
+++ /dev/null
@@ -1,45 +0,0 @@
-"""Launcher for the idatui TUI.
-
-  # attach to the single open session on a running server
-  python -m idatui.tui
-
-  # attach to a specific session
-  python -m idatui.tui --db 80d83396
-
-  # open (or reopen) an arbitrary binary, then drive it
-  python -m idatui.tui --open /path/to/binary
-
-The server (supervisor) must already be running (see spawn.sh). --open creates a
-session via idb_open; the binary's directory must be writable (idalib writes a
-.i64 next to it).
-"""
-from __future__ import annotations
-
-import argparse
-import os
-
-from .app import IdaTui
-from .client import DEFAULT_URL
-
-
-def main(argv: list[str] | None = None) -> int:
-    p = argparse.ArgumentParser(prog="idatui", description="Minimal TUI for IDA over MCP")
-    p.add_argument("--url", default=os.environ.get("IDA_MCP_URL", DEFAULT_URL),
-                   help=f"MCP server URL (default {DEFAULT_URL})")
-    p.add_argument("--db", default=os.environ.get("IDA_MCP_DB"),
-                   help="attach to an existing session id")
-    p.add_argument("--open", metavar="PATH",
-                   help="open (or reopen) a binary and drive it (dir must be writable)")
-    p.add_argument("--no-keepalive", action="store_true",
-                   help="Do not bump idle-TTL / run the keepalive heartbeat")
-    p.add_argument("--rpc", metavar="PATH",
-                   help="listen for RPC on this unix socket path (puppeteer the TUI)")
-    args = p.parse_args(argv)
-    rpc_path = os.path.abspath(os.path.expanduser(args.rpc)) if args.rpc else None
-    IdaTui(url=args.url, db=args.db, open_path=args.open,
-           keepalive=not args.no_keepalive, rpc_path=rpc_path).run()
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
diff --git a/idatui/worker.py b/idatui/worker.py
index 64cafb3..bfefa4a 100644
--- a/idatui/worker.py
+++ b/idatui/worker.py
@@ -62,7 +62,7 @@ def recv(sock: socket.socket):
 def _ensure_tools_injected() -> None:
     """Inject idatui's custom tools (heads/read_raw/resolve_names/func_types/...)
     into the installed ida_pro_mcp, idempotently, so the worker is self-sufficient
-    without spawn.sh having run server/patch_server.py first. Must run BEFORE
+    (nothing else has to inject these tools first). Must run BEFORE
     ida_pro_mcp.ida_mcp is imported (the injected code lives in api_types.py)."""
     import importlib.util
     repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))