launch: `ida-tui foo.elf` one-shot wrapper (server + locks + session)

A caveman entry point so you don't hand-craft the plumbing every time. It:
  * ensures the ida-pro-mcp supervisor is up — starts spawn.sh detached
    (start_new_session, tmux-free) and waits for the port if it's down;
  * recovers a binary wedged by a hard-killed worker — sweeps the stale
    unpacked .id0/.id1/.id2/.nam/.til next to the .i64 and retries (the packed
    .i64 is never touched);
  * adopts an already-open session for the same binary (idempotent), else
    idb_opens it with a sane idle-TTL;
  * launches the TUI attached to that session with keepalive on.

  ./ida-tui /path/to/binary     # open a binary and drive it
  ./ida-tui                     # attach to the sole session
  ./ida-tui --db <id>           # attach to a specific session

Pieces: idatui/launch.py (logic, reuses pane.py's server probe), a repo-root
`ida-tui` sh wrapper (resolves ~/ida-venv python, keeps idatui importable from
any cwd), and an `ida-tui` console-script in pyproject. --help works without
textual (app imported late). README documents both the one-liner and the
manual recovery. gitignore bin/ (binary targets, like targets/).

Verified live: ensure_server up-detection, open, adopt (same session id), and
lock-sweep all work against a running supervisor.

1 files changed, 234 insertions, 0 deletions

diff --git a/idatui/launch.py b/idatui/launch.py
new file mode 100644
index 0000000..c585668
--- /dev/null
+++ b/idatui/launch.py
@@ -0,0 +1,234 @@
+"""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.
+
+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
+
+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.
+"""
+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
+# reopen ("Failed to open database"). Safe to delete when nothing holds the DB.
+_LOCK_SUFFIXES = (".id0", ".id1", ".id2", ".nam", ".til")
+
+
+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) -> bool:
+    """Make sure the supervisor is listening; start it if not. Returns True when
+    the port is up. Only auto-starts a local server."""
+    host, port = _server_addr(url)
+    if _server_up(host, port):
+        return True
+    if host not in ("127.0.0.1", "localhost", "::1"):
+        _log(f"server at {host}:{port} is down and not local — cannot auto-start")
+        return False
+    _log(f"supervisor not running — starting it (log: {_server_log_path()})")
+    proc = _start_supervisor(target)
+    deadline = time.time() + timeout
+    spun = False
+    while time.time() < deadline:
+        if _server_up(host, port):
+            if spun:
+                _log("supervisor is up")
+            return True
+        if proc is not None and proc.poll() is not None:
+            _log("supervisor exited during startup — check the log above")
+            return False
+        spun = True
+        time.sleep(0.5)
+    _log(f"supervisor did not come up 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)
+                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)")
+    p.add_argument("--ttl", type=int, default=1800,
+                   help="worker idle-TTL seconds for the opened session (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")
+    args = p.parse_args(argv)
+
+    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
+
+    # 1) supervisor up (seed it with our target so its initial worker is useful)
+    if not args.no_server:
+        if not _ensure_server(args.url, binary):
+            return 1
+    elif not _server_up(*_server_addr(args.url)):
+        _log("supervisor is down and --no-server was given")
+        return 1
+
+    # 2) resolve the session (open/adopt the binary, or use --db / the sole one)
+    db = args.db
+    if binary is not None:
+        try:
+            client = IDAClient(args.url)
+            client.connect()
+            db = _open_binary(client, binary, args.ttl)
+        except IDAError as e:
+            _log(f"could not open {binary}: {e}")
+            return 1
+        finally:
+            try:
+                client.close()
+            except Exception:  # noqa: BLE001
+                pass
+
+    # 3) hand off to the TUI (imported late so --help works without textual)
+    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=db, keepalive=not args.no_keepalive,
+           rpc_path=rpc_path).run()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())