mcp: collapse app + launcher to worker-only

The idalib worker is now the sole backend for opening a binary, so remove the
ida-pro-mcp code paths from the hot path:

* app.py: IdaTui.__init__ drops url/db/ensure_server/backend (now just
  open_path/keepalive/rpc_path/ttl); _connect calls _open_worker_client directly;
  _open_mcp_client deleted; _reconnect respawns the worker only; self.client and
  _after_reconnect typed WorkerClient. No more IDAClient import.
* launch.py: rewritten worker-only -- validate the binary, sweep stale locks,
  spawn the TUI (which starts the private worker behind its overlay). The whole
  supervisor dance (_ensure_server/_start_supervisor/_open_binary/
  _existing_session) is gone; `ida-tui foo.elf` is the one usage.
* tests/test_scenarios.py: pilot is worker-only (run(binary); binary via
  positional/--worker/--binary, defaults to targets/echo).

Verified: app + launch + pilot import and construct; pilot lists 31 scenarios.
The mcp modules (client.py/pane.py/tui.py/spawn.sh) still exist as dead code and
are deleted in the next commit. Worker pilot (134/2-known-flakes) still the gate.

3 files changed, 54 insertions, 330 deletions

diff --git a/idatui/app.py b/idatui/app.py
index 712b4de..d442b0a 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -43,7 +43,7 @@ from textual.widgets.option_list import Option
 from .highlight import highlight_c
 
 from .errors import IDAToolError, IDAConnectionError
-from .client import IDAClient  # deprecated mcp transport (worker path uses WorkerClient)
+from .worker_client import WorkerClient
 from .domain import DisasmModel, Func, Head, ListingModel, Program, Struct
 
 # Styles for the disassembly listing.
@@ -2316,21 +2316,15 @@ class IdaTui(App):
         Binding("escape", "back", "Back"),
     ]
 
-    def __init__(self, url: str, db: str | None, keepalive: bool = True,
-                 open_path: str | None = None, rpc_path: str | None = None,
-                 ttl: int = 1800, ensure_server: bool = False,
-                 backend: str = "mcp") -> None:
+    def __init__(self, open_path: str, keepalive: bool = True,
+                 rpc_path: str | None = None, ttl: int = 1800) -> None:
         super().__init__()
-        self._url = url
-        self._db = db
         self._open_path = open_path
         self._ttl = ttl
-        self._ensure_server = ensure_server
-        self._backend = backend  # "mcp" (ida-pro-mcp) or "worker" (our idalib worker)
         self._do_keepalive = keepalive
         self._rpc_path = rpc_path
         self._rpc = None
-        self.client: IDAClient | None = None
+        self.client: WorkerClient | None = None
         self.program: Program | None = None
         self._loading_screen: LoadingScreen | None = None
         self._ka = None
@@ -2418,11 +2412,7 @@ class IdaTui(App):
             self._start_rpc()
 
     def _loading_title(self) -> str:
-        if self._open_path:
-            return os.path.basename(self._open_path)
-        if self._db:
-            return str(self._db)
-        return "database"
+        return os.path.basename(self._open_path) if self._open_path else "database"
 
     def _dismiss_loading(self) -> None:
         ls = self._loading_screen
@@ -2513,55 +2503,22 @@ class IdaTui(App):
 
     @work(thread=True, exclusive=True, group="reconnect")
     def _reconnect(self) -> None:
+        # The worker died (segfault -> dropped socket). Respawn it: it re-opens
+        # and re-analyzes the binary in a fresh process, then we rebuild.
         try:
-            if self._backend == "worker":
-                from .worker_client import WorkerClient
-                if self._open_path is None:
-                    self.app.call_from_thread(self._reconnect_failed,
-                                              "no binary to reopen")
-                    return
-                client = WorkerClient(self._open_path, ttl=self._ttl)
-                client.connect(progress=lambda m: self.app.call_from_thread(
-                    self._conn_note, m))
-                self.app.call_from_thread(self._after_reconnect, client,
-                                          Program(client))
-                return
-            from .launch import _ensure_server
-            if not _ensure_server(
-                    self._url, self._open_path,
-                    progress=lambda m: self.app.call_from_thread(self._conn_note, m)):
+            if self._open_path is None:
                 self.app.call_from_thread(self._reconnect_failed,
-                                          "analysis server is unreachable")
+                                          "no binary to reopen")
                 return
-            client = IDAClient(self._url, db=None)
-            client.connect()
-            if self._open_path is not None:
-                path = os.path.abspath(os.path.expanduser(self._open_path))
-                self.app.call_from_thread(
-                    self._conn_note, f"re-opening {os.path.basename(path)}\u2026")
-                res = client.call("idb_open", input_path=path,
-                                  idle_ttl_sec=self._ttl, timeout=1800.0)
-                if not (isinstance(res, dict) and res.get("success")):
-                    self.app.call_from_thread(self._reconnect_failed, "re-open failed")
-                    return
-                client.set_db(res["session"]["session_id"])
-            else:
-                client.set_db(client.resolve_db())
-            client.health()
-            if self._ka is not None:
-                try:
-                    self._ka.stop()
-                except Exception:  # noqa: BLE001
-                    pass
-            if self._do_keepalive:
-                self._ka = client.keepalive(interval=120.0).start()
-            program = Program(client)
+            client = WorkerClient(self._open_path, ttl=self._ttl)
+            client.connect(progress=lambda m: self.app.call_from_thread(
+                self._conn_note, m))
         except Exception as e:  # noqa: BLE001
             self.app.call_from_thread(self._reconnect_failed, str(e))
             return
-        self.app.call_from_thread(self._after_reconnect, client, program)
+        self.app.call_from_thread(self._after_reconnect, client, Program(client))
 
-    def _after_reconnect(self, client: IDAClient, program: "Program") -> None:
+    def _after_reconnect(self, client: "WorkerClient", program: "Program") -> None:
         self.client = client
         self.program = program
         self._reconnecting = False
@@ -2581,8 +2538,7 @@ class IdaTui(App):
     @work(thread=True, exclusive=True, group="connect")
     def _connect(self) -> None:
         try:
-            client = (self._open_worker_client() if self._backend == "worker"
-                      else self._open_mcp_client())
+            client = self._open_worker_client()
             if client is None:
                 return  # the opener already reported + dismissed the overlay
             module = client.health().get("module", "?")
@@ -2600,55 +2556,6 @@ class IdaTui(App):
         self.app.call_from_thread(self._status, f"{module} — loading functions…")
         self._load_functions()
 
-    def _open_mcp_client(self):  # type: ignore[no-untyped-def]
-        """ida-pro-mcp path: ensure the server, connect, open/adopt the DB.
-        Returns the client, or None after reporting the failure."""
-        if self._ensure_server:
-            # Start the analysis server here (not in the launcher) so the chrome
-            # + overlay are already up while we wait for it.
-            from .launch import _ensure_server as _ensure
-            if not _ensure(self._url, self._open_path,
-                           progress=lambda m: self.app.call_from_thread(
-                               self._status, m)):
-                self.app.call_from_thread(
-                    self._status, "could not start the analysis server")
-                self.app.call_from_thread(self._dismiss_loading)
-                return None
-        client = IDAClient(self._url, db=self._db)
-        client.connect()
-        if self._open_path is not None:
-            path = os.path.abspath(os.path.expanduser(self._open_path))
-            base = os.path.basename(path)
-            self.app.call_from_thread(self._status, f"opening {base} — analyzing…")
-
-            def _do_open():
-                return client.call("idb_open", input_path=path,
-                                   idle_ttl_sec=self._ttl, timeout=1800.0)
-
-            res = _do_open()
-            if not (isinstance(res, dict) and res.get("success")):
-                # A hard-killed worker can wedge the DB; sweep its stale unpacked
-                # lock files (never the .i64) and retry once.
-                try:
-                    from .launch import _sweep_locks
-                    swept = _sweep_locks(path)
-                except Exception:  # noqa: BLE001
-                    swept = 0
-                if swept:
-                    self.app.call_from_thread(
-                        self._status, f"recovering {base} (removed {swept} stale "
-                        f"lock file(s))…")
-                    res = _do_open()
-            if not (isinstance(res, dict) and res.get("success")):
-                err = res.get("error") if isinstance(res, dict) else res
-                self.app.call_from_thread(self._status, f"open failed: {err}")
-                self.app.call_from_thread(self._dismiss_loading)
-                return None
-            client.set_db(res["session"]["session_id"])
-        elif self._db is None:
-            client.set_db(client.resolve_db())
-        return client
-
     def _open_worker_client(self):  # type: ignore[no-untyped-def]
         """Our idalib-worker path: spawn the worker (it opens + analyzes the
         binary in its own process) and connect. Returns the client, or None."""
diff --git a/idatui/launch.py b/idatui/launch.py
index b2f0eec..e2845e7 100644
--- a/idatui/launch.py
+++ b/idatui/launch.py
@@ -1,38 +1,24 @@
 """One-shot launcher: ``ida-tui foo.elf`` and you're in the TUI.
 
-Does the whole plumbing so you don't have to:
-
-  * ensures the ida-pro-mcp supervisor is up (starts ``spawn.sh`` detached and
-    waits for the port if it isn't) — works inside OR outside tmux;
-  * recovers a binary wedged by a crashed worker (sweeps the stale unpacked
-    ``.id0/.id1/.id2/.nam/.til`` files left when a worker was hard-killed, then
-    retries) — the packed ``.i64`` is never touched;
-  * reuses an already-open session for the same binary, or opens it fresh;
-  * launches the TUI attached to that session with keepalive on.
+Spawns a private idalib worker (``idatui.worker``) that opens + auto-analyzes
+THIS binary in its own process, talking to the TUI over a unix socket. No shared
+supervisor, no HTTP: everything slow (open + analysis) happens behind the TUI's
+loading overlay.
 
 Usage:
 
-    ida-tui /path/to/binary          # open (or adopt) a binary and drive it
-    ida-tui                          # attach to the sole open session
-    ida-tui --db 80d83396            # attach to a specific session
+    ida-tui /path/to/binary          # open a binary and drive it
 
-Everything ``idatui.tui`` accepts (--url, --rpc, --no-keepalive) is accepted
-here too. The heavy lifting for probing/reaping the supervisor lives in
-``pane.py``; this module adds the tmux-free auto-start and lock recovery.
+Extras: --ttl, --no-keepalive, --rpc (all forwarded to the TUI).
 """
 from __future__ import annotations
 
 import argparse
 import os
-import subprocess
 import sys
-import time
-
-from .client import DEFAULT_URL, IDAClient, IDAError
-from .pane import REPO, _server_addr, _server_up
 
 # The unpacked working-copy files IDA writes next to a `.i64` while a database is
-# open. A hard-killed worker leaves them behind and then the `.i64` refuses to
+# open. A hard-killed worker leaves them behind and the `.i64` then refuses to
 # reopen ("Failed to open database"). Safe to delete when nothing holds the DB.
 _LOCK_SUFFIXES = (".id0", ".id1", ".id2", ".nam", ".til")
 
@@ -41,220 +27,55 @@ def _log(msg: str) -> None:
     print(f"ida-tui: {msg}", file=sys.stderr)
 
 
-# --------------------------------------------------------------------------- #
-# supervisor auto-start (tmux-free)
-# --------------------------------------------------------------------------- #
-def _server_log_path() -> str:
-    base = os.environ.get("XDG_RUNTIME_DIR") or "/tmp"
-    return os.path.join(base, "idatui-supervisor.log")
-
-
-def _start_supervisor(target: str | None) -> subprocess.Popen | None:
-    """Start ``spawn.sh`` detached (its own session, so it outlives us) with the
-    given initial target. Returns the Popen (for a failure hint) or None."""
-    cmd = os.environ.get("IDATUI_SERVER_CMD", "./spawn.sh")
-    env = dict(os.environ)
-    if target:
-        env["IDA_MCP_TARGET"] = target
-    logf = open(_server_log_path(), "ab", buffering=0)
-    try:
-        return subprocess.Popen(
-            cmd, cwd=REPO, shell=True, env=env,
-            stdin=subprocess.DEVNULL, stdout=logf, stderr=logf,
-            start_new_session=True,  # detach: survives this process exiting
-        )
-    except OSError as e:
-        _log(f"could not start supervisor ({cmd!r}): {e}")
-        return None
-
-
-def _ensure_server(url: str, target: str | None, timeout: float = 90.0,
-                   progress=None) -> bool:
-    """Make sure the supervisor is listening; start it if not. Returns True when
-    the port is up. Only auto-starts a local server. ``progress`` (if given) gets
-    human-readable status lines instead of stderr — so the TUI can narrate the
-    wait in its loading overlay."""
-    def note(msg: str) -> None:
-        (progress or _log)(msg)
-
-    host, port = _server_addr(url)
-    if _server_up(host, port):
-        return True
-    if host not in ("127.0.0.1", "localhost", "::1"):
-        note(f"server at {host}:{port} is down and not local — cannot auto-start")
-        return False
-    # The supervisor opens + auto-analyzes its seed binary BEFORE it binds the
-    # port (idalib_supervisor.main), so this whole wait is really IDA running
-    # initial auto-analysis — label it as such, not just "starting a server".
-    label = os.path.basename(target) if target else "binary"
-    note(f"starting IDA — initial auto-analysis of {label}…")
-    proc = _start_supervisor(target)
-    deadline = time.time() + timeout
-    t0 = time.time()
-    while time.time() < deadline:
-        if _server_up(host, port):
-            return True
-        if proc is not None and proc.poll() is not None:
-            note("analysis server exited during startup — check the log")
-            return False
-        note(f"auto-analyzing {label}… ({int(time.time() - t0)}s)")
-        time.sleep(0.5)
-    note(f"analysis did not finish within {timeout:.0f}s")
-    return False
-
-
-# --------------------------------------------------------------------------- #
-# session resolution + lock recovery
-# --------------------------------------------------------------------------- #
 def _sweep_locks(binary: str) -> int:
     """Remove stale unpacked DB files next to ``binary``. Returns how many."""
     stem = os.path.splitext(binary)[0]
     n = 0
     for base in (binary, stem):  # IDA may key on the full name or the stem
         for suf in _LOCK_SUFFIXES:
-            p = base + suf
             try:
-                os.remove(p)
+                os.remove(base + suf)
                 n += 1
             except OSError:
                 pass
     return n
 
 
-def _existing_session(client: IDAClient, binary: str) -> str | None:
-    """A live session already open on ``binary`` (by real path), if any."""
-    target = os.path.realpath(binary)
-    for s in client.list_sessions():
-        if s.session_id and s.input_path and os.path.realpath(s.input_path) == target:
-            return s.session_id
-    return None
-
-
-def _open_binary(client: IDAClient, binary: str, ttl: int) -> str:
-    """Open (or adopt) ``binary`` and return its session id. Recovers from a
-    crashed-worker lock by sweeping stale files and retrying once."""
-    existing = _existing_session(client, binary)
-    if existing:
-        _log(f"adopting the open session for {os.path.basename(binary)} ({existing})")
-        # re-open on the same path is idempotent and (re)applies the TTL
-        try:
-            client.call("idb_open", input_path=binary, idle_ttl_sec=ttl)
-        except IDAError:
-            pass
-        return existing
-
-    def _do_open() -> str:
-        r = client.call("idb_open", input_path=binary, idle_ttl_sec=ttl)
-        sess = r.get("session") if isinstance(r, dict) else None
-        sid = (sess or {}).get("session_id") if isinstance(sess, dict) else None
-        if not sid:  # fall back to whatever list_sessions now shows for this path
-            sid = _existing_session(client, binary)
-        if not sid:
-            raise IDAError(f"idb_open returned no session id for {binary}")
-        return sid
-
-    try:
-        return _do_open()
-    except IDAError as e:
-        # A wedged database from a hard-killed worker: sweep THIS binary's
-        # unpacked working-copy files (never the .i64) and retry once. We only
-        # get here when no live session already holds this binary (adopted
-        # above), so the sweep can't race a healthy session.
-        swept = _sweep_locks(binary)
-        if not swept:
-            raise e
-        _log(f"recovered a wedged database (removed {swept} stale lock "
-             f"file(s)); retrying")
-        return _do_open()
-
-
-# --------------------------------------------------------------------------- #
-# entry point
-# --------------------------------------------------------------------------- #
 def main(argv: list[str] | None = None) -> int:
     p = argparse.ArgumentParser(
         prog="ida-tui",
-        description="Open a binary in the IDA TUI, doing all the server plumbing.")
-    p.add_argument("binary", nargs="?",
-                   help="binary to open/adopt (omit to attach to the sole session)")
-    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 (skips open)")
+        description="Open a binary in the IDA TUI (private idalib worker).")
+    p.add_argument("binary", help="binary to open and analyze")
     p.add_argument("--ttl", type=int, default=1800,
-                   help="worker idle-TTL seconds for the opened session (default 1800)")
+                   help="worker idle-TTL seconds (default 1800)")
     p.add_argument("--no-keepalive", action="store_true",
                    help="do not run the keepalive heartbeat")
     p.add_argument("--rpc", metavar="PATH",
                    help="listen for RPC on this unix socket (puppeteer the TUI)")
-    p.add_argument("--no-server", action="store_true",
-                   help="do not auto-start the supervisor if it's down")
-    p.add_argument("--backend", choices=("mcp", "worker"), default=None,
-                   help="'worker' (our own idalib worker over a unix socket, the "
-                        "default when opening a binary) or 'mcp' (deprecated "
-                        "ida-pro-mcp supervisor; used for --db/attach)")
     args = p.parse_args(argv)
 
-    # Backend selection: explicit flag > IDATUI_BACKEND > worker for a fresh
-    # binary open, mcp for the attach modes (--db / bare `ida-tui`) which have no
-    # worker equivalent. The mcp path is deprecated and on its way out.
-    if args.backend is None:
-        args.backend = os.environ.get("IDATUI_BACKEND") or (
-            "worker" if (args.binary and not args.db) else "mcp")
-    if args.backend == "mcp":
-        _log("using the deprecated ida-pro-mcp backend "
-             "(the idalib worker is default for opening a binary)")
-
-    binary = None
-    if args.binary and not args.db:
-        binary = os.path.abspath(os.path.expanduser(args.binary))
-        if not os.path.isfile(binary):
-            _log(f"no such file: {binary}")
-            return 2
-        if not os.access(os.path.dirname(binary), os.W_OK):
-            _log(f"directory not writable (IDA writes a .i64 there): "
-                 f"{os.path.dirname(binary)}")
-            return 2
-
-    if args.backend == "worker":
-        # The worker path owns no shared supervisor: the TUI spawns a private
-        # idalib worker that opens THIS binary. No server/lock plumbing here.
-        if binary is None:
-            _log("--backend worker needs a binary path")
-            return 2
-        swept = _sweep_locks(binary)  # a crashed worker can leave the DB wedged
-        if swept:
-            _log(f"cleared {swept} stale lock file(s) from a crashed worker")
-    else:
-        # Everything slow — starting the supervisor and opening/analyzing the
-        # binary — is deferred to the TUI so its chrome + "loading…" overlay are
-        # on screen the whole time. Here we only do the instant checks.
-        server_down = not _server_up(*_server_addr(args.url))
-        if args.no_server and server_down:
-            _log("supervisor is down and --no-server was given")
-            return 1
-        if binary is not None and server_down:
-            # About to start a fresh supervisor (nothing holds the DB): clear any
-            # stale unpacked lock files a crashed worker left, so the seeded
-            # initial open doesn't crash-loop on a wedged DB.
-            swept = _sweep_locks(binary)
-            if swept:
-                _log(f"cleared {swept} stale lock file(s) from a crashed worker")
+    binary = os.path.abspath(os.path.expanduser(args.binary))
+    if not os.path.isfile(binary):
+        _log(f"no such file: {binary}")
+        return 2
+    if not os.access(os.path.dirname(binary), os.W_OK):
+        _log(f"directory not writable (IDA writes a .i64 there): "
+             f"{os.path.dirname(binary)}")
+        return 2
+    swept = _sweep_locks(binary)  # a crashed worker can leave the DB wedged
+    if swept:
+        _log(f"cleared {swept} stale lock file(s) from a crashed worker")
 
     # Hand off to the TUI (imported late so --help works without textual). It
-    # ensures the server (with on-screen progress), then opens/adopts the binary
-    # — idb_open is idempotent, so an already-open session is adopted instantly.
+    # spawns the worker behind its loading overlay while auto-analysis runs.
     try:
         from .app import IdaTui
     except ImportError as e:
         _log(f"the TUI needs textual; run with ~/ida-venv/bin/python  ({e})")
         return 1
     rpc_path = os.path.abspath(os.path.expanduser(args.rpc)) if args.rpc else None
-    IdaTui(url=args.url, db=args.db,
-           open_path=(binary if args.db is None else None), ttl=args.ttl,
-           ensure_server=(args.backend == "mcp" and not args.no_server),
-           backend=args.backend,
-           keepalive=not args.no_keepalive, rpc_path=rpc_path).run()
+    IdaTui(open_path=binary, keepalive=not args.no_keepalive,
+           rpc_path=rpc_path, ttl=args.ttl).run()
     return 0
 
 
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py
index 09335d7..62f474d 100644
--- a/tests/test_scenarios.py
+++ b/tests/test_scenarios.py
@@ -1651,15 +1651,9 @@ async def s_func_banners(c: Ctx):
 # --------------------------------------------------------------------------- #
 # Runner
 # --------------------------------------------------------------------------- #
-async def run(db, only=None, binary=None, backend="mcp"):
-    if backend == "worker":
-        # Own idalib worker: opens the binary in-process over a unix socket, no
-        # supervisor. Mirrors launch.py's worker branch.
-        app = IdaTui(url="", db=None, open_path=binary, ensure_server=False,
-                     backend="worker", keepalive=False)
-    else:
-        url = os.environ.get("IDA_MCP_URL", "http://127.0.0.1:8745/mcp")
-        app = IdaTui(url=url, db=db, keepalive=False)
+async def run(binary, only=None):
+    # Own idalib worker: opens the binary in-process over a unix socket.
+    app = IdaTui(open_path=binary, keepalive=False)
     async with app.run_test(size=(140, 44)) as pilot:
         c = Ctx(app, pilot)
         await c.boot()
@@ -1682,26 +1676,28 @@ async def run(db, only=None, binary=None, backend="mcp"):
 
 def main(argv):
     global STOP_AFTER
-    db = only = binary = None
-    backend = "mcp"
+    only = None
+    binary = None
     it = iter(argv)
     for a in it:
-        if a == "--db":
-            db = next(it)
-        elif a == "--only":
+        if a == "--only":
             only = next(it).split(",")
         elif a == "--stop-after":
             STOP_AFTER = next(it)
-        elif a == "--worker":
-            # --worker <binary>: run the suite against our own idalib worker
-            backend = "worker"
+        elif a in ("--worker", "--binary"):
             binary = os.path.abspath(os.path.expanduser(next(it)))
         elif a == "--list":
             for name, _ in SCENARIOS:
                 print(name)
             return 0
+        elif not a.startswith("-"):
+            binary = os.path.abspath(os.path.expanduser(a))
+    if binary is None:  # default target for the pilot
+        binary = os.path.join(
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            "targets", "echo")
     try:
-        asyncio.run(run(db, only, binary=binary, backend=backend))
+        asyncio.run(run(binary, only))
     except _StopSuite:
         print(f"  … stopped after '{STOP_AFTER}'")
     print(f"\n{PASS} passed, {FAIL} failed")