Stop ida-pro-mcp installing a sys.setprofile hook around every tool call. Its deadline mechanism profiles every python call/return so a pure-python tool loop can be interrupted; our tools are call-heavy, so it taxed the whole backend 3.3x. Worker now sets IDA_MCP_TOOL_TIMEOUT_SEC=0 and arms the deadline itself with one polling watchdog thread + ida_kernwin.set_cancelled() (the half that actually frees the IDA main thread). Also rewrote _idatui_spans to jump between colour tags instead of walking characters (byte-identical over 258k real lines).

Result: {"status":"keep","total_ms":26923.9,"lg_boot_ms":762.2,"lg_decomp_ms":2631.7,"lg_graph_ms":941.8,"lg_hex_ms":1052,"lg_index_ms":67.2,"lg_listing_cold_ms":440.6,"lg_listing_warm_ms":530.5,"lg_nav_ms":10598.3,"lg_palette_ms":4.6,"lg_render_ms":227.8,"lg_search_ms":5265.5,"pure_graph_ms":237.9,"sm_boot_ms":535.3,"sm_decomp_ms":631.8,"sm_graph_ms":702.1,"sm_hex_ms":841.1,"sm_index_ms":0,"sm_listing_cold_ms":268.2,"sm_listing_warm_ms":269.4,"sm_nav_ms":443.6,"sm_palette_ms":0.3,"sm_render_ms":277.8,"sm_search_ms":194.4,"fails":0}

1 files changed, 69 insertions, 1 deletions

diff --git a/idatui/worker.py b/idatui/worker.py
index a4e3509..556e69a 100644
--- a/idatui/worker.py
+++ b/idatui/worker.py
@@ -26,8 +26,68 @@ import pickle
 import socket
 import struct
 import sys
+import threading
+import time
 import uuid
 
+#: Seconds a single tool call may run before it is cancelled. 0 disables the
+#: deadline entirely.
+TOOL_TIMEOUT_SEC = float(os.environ.get("IDATUI_TOOL_TIMEOUT_SEC") or 60)
+
+# ida-pro-mcp enforces its own tool deadline by installing a `sys.setprofile`
+# hook for the duration of every call, so that a pure-python loop inside a tool
+# body can be interrupted. That hook runs a python function on EVERY python call
+# and return -- and our tools are exactly the call-heavy kind: `heads` renders
+# hundreds of items per request and measured 92us/row with the hook against
+# 28us/row without it. A 3.3x tax on the whole backend to bound loops that are
+# already bounded by their `count` argument.
+#
+# So: turn the upstream mechanism off and re-arm the half that does the real
+# work ourselves (see _Deadline). ida_kernwin.set_cancelled() is what actually
+# frees the IDA main thread -- decompile, auto_wait, find_bytes and friends poll
+# user_cancelled() and bail within a poll cycle -- and it costs nothing until it
+# fires.
+os.environ["IDA_MCP_TOOL_TIMEOUT_SEC"] = "0"
+
+
+class _Deadline:
+    """A single watchdog thread that cancels a tool call which overruns.
+
+    Arming is two attribute writes, because it is on the path of every call the
+    TUI makes (a scroll is dozens of them). The watchdog polls instead of being
+    signalled for the same reason: waking a thread per call costs more than the
+    0.25s of granularity it buys on a 60s deadline.
+    """
+
+    TICK = 0.25
+
+    def __init__(self, seconds: float) -> None:
+        import ida_kernwin
+        self._kernwin = ida_kernwin
+        self.seconds = seconds
+        self._until: float | None = None
+        t = threading.Thread(target=self._run, name="idatui-deadline",
+                             daemon=True)
+        t.start()
+
+    def _run(self) -> None:
+        while True:
+            time.sleep(self.TICK)
+            until = self._until
+            if until is not None and time.monotonic() >= until:
+                self._until = None
+                # THREAD_SAFE in the IDA SDK; upstream fires it off a Timer too.
+                self._kernwin.set_cancelled()
+
+    def arm(self) -> None:
+        # Clear unconditionally: the flag is sticky, and one left set would make
+        # every later user_cancelled() true forever.
+        self._kernwin.clr_cancelled()
+        self._until = time.monotonic() + self.seconds
+
+    def disarm(self) -> None:
+        self._until = None
+
 
 # --------------------------------------------------------------------------- #
 # framing
@@ -147,6 +207,7 @@ def _open_and_register(binpath: str, load_args: str = ""):
 def serve(sockpath: str, binpath: str, load_args: str = "") -> None:
     tools, module, save = _open_and_register(binpath, load_args)
     sid = uuid.uuid4().hex[:8]
+    deadline = _Deadline(TOOL_TIMEOUT_SEC) if TOOL_TIMEOUT_SEC > 0 else None
 
     def dispatch(name: str, args: dict):
         args = dict(args)
@@ -167,7 +228,14 @@ def serve(sockpath: str, binpath: str, load_args: str = "") -> None:
         fn = tools.get(name)
         if fn is None:
             raise KeyError(f"unknown tool: {name!r}")
-        result = fn(**args)
+        if deadline is None:
+            result = fn(**args)
+        else:
+            deadline.arm()
+            try:
+                result = fn(**args)
+            finally:
+                deadline.disarm()
         # Match the MCP server's structuredContent: a dict passes through, any
         # other return (list/scalar) is wrapped as {"result": ...}. domain.py
         # parses that exact shape (e.g. lookup_funcs -> payload["result"]).