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.

1 files changed, 25 insertions, 204 deletions

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