reformat: ruff format + import sort, mechanically (see ruff.toml)

No behavior. Listed in .git-blame-ignore-revs (next commit).

1 files changed, 512 insertions, 213 deletions

diff --git a/idatui/rpc.py b/idatui/rpc.py
index 1262ebb..cfeb0ca 100644
--- a/idatui/rpc.py
+++ b/idatui/rpc.py
@@ -17,6 +17,7 @@ Method tiers:
   introspect   state, view, screen, functions
   (semantic verbs — open/goto/rename/... — layer on top in a later pass.)
 """
+
 from __future__ import annotations
 
 import asyncio
@@ -27,8 +28,8 @@ from typing import Any
 
 from rich.console import Console
 
-from ._sync import drain, settle
 from . import diag
+from ._sync import drain, settle
 from .app import DecompView, GraphView, HexView, ListingView, ViewMode
 
 PROTO_VERSION = 1
@@ -36,10 +37,31 @@ TYPE_DELAY_MS = 35  # default per-char delay for high-level typed ops (aesthetic
 
 # Verbs that dereference app.program — refused with a clear error before load.
 _PROGRAM_METHODS = {
-    "goto", "open", "rename", "comment", "retype", "follow", "xrefs", "symbols",
-    "structs", "search", "select", "save", "hex", "toggle_view",
-    "pseudocode", "disassembly", "xrefs_to", "xrefs_from", "resolve",
-    "define", "rename_many", "opfmt", "graph", "export", "find",
+    "goto",
+    "open",
+    "rename",
+    "comment",
+    "retype",
+    "follow",
+    "xrefs",
+    "symbols",
+    "structs",
+    "search",
+    "select",
+    "save",
+    "hex",
+    "toggle_view",
+    "pseudocode",
+    "disassembly",
+    "xrefs_to",
+    "xrefs_from",
+    "resolve",
+    "define",
+    "rename_many",
+    "opfmt",
+    "graph",
+    "export",
+    "find",
 }
 
 # Self-documenting method table (returned by the 'methods' verb).
@@ -55,7 +77,7 @@ METHODS = {
     "disassembly": "{target?,max?=2000} -> {total,lines:[{ea,text}]}",
     "xrefs_to": "{target,limit?=200} -> [{frm,to,type,fn_addr,fn_name}]",
     "xrefs_from": "{target,limit?=200} -> callees/refs; function-scoped for a "
-                  "function (decomp refs), address-scoped for a 0xADDR",
+    "function (decomp refs), address-scoped for a 0xADDR",
     "resolve": "{name} -> {ea}",
     "keys": "{keys:[str],settle?,timeout?} raw key injection (supports 'wait:<ms>')",
     "text": "{text,delay_ms?,settle?} type a literal string into the focused input",
@@ -69,17 +91,17 @@ METHODS = {
     "toggle_view": "disasm <-> pseudocode",
     "hex": "hex view",
     "graph": "{action?=show|open|close|toggle|zoom|block|entry|succ|pred,"
-             "target?,blocks?} the control-flow graph: 'show' reports its "
-             "structure (blocks, edges, cursor) without touching it; the others "
-             "drive it. 'block' takes target=<id|0xADDR>",
+    "target?,blocks?} the control-flow graph: 'show' reports its "
+    "structure (blocks, edges, cursor) without touching it; the others "
+    "drive it. 'block' takes target=<id|0xADDR>",
     "xrefs": "open the xref picker",
     "symbols": "{query?} open the symbol palette",
     "structs": "open the struct editor",
     "export": "{path?,types?=true} write the session's comments/names/types as "
-              "a markdown report -> {path,comments,names,types}",
+    "a markdown report -> {path,comments,names,types}",
     "find": "{query,mode?=auto|text|bytes,limit?=500,regex?,case?} search the "
-            "WHOLE database: disassembly text, or a byte pattern with "
-            "wildcards (48 8b ?? c3) -> {mode,hits:[{addr,head,line,func}]}",
+    "WHOLE database: disassembly text, or a byte pattern with "
+    "wildcards (48 8b ?? c3) -> {mode,hits:[{addr,head,line,func}]}",
     "search": "{term,direction?=1} incremental search in the code view",
     "select": "{index?} choose the highlighted/nth item in the open modal",
     "save": "persist the .i64 (Ctrl+S)",
@@ -90,42 +112,62 @@ METHODS = {
     "move": "{dir,n?=1} fast movement (down/up/.../pagedown)",
     "cursor": "{line?,col?} set the code-pane cursor directly",
     "define": "{kind:code|func|undef|thumb|thumbscan|data|string,target?} "
-              "(re)define bytes at target — the raw-image workflow",
+    "(re)define bytes at target — the raw-image workflow",
     "rename_many": "{items:[{addr,name}] | file:JSON} bulk-apply a symbol file "
-                   "in ONE call (no typing, no navigation)",
+    "in ONE call (no typing, no navigation)",
     "opfmt": "{mode?=cycle|back|show|hex|dec|oct|bin|char|offset|stack|"
-             "default,target?,word?,line?,col?} how the literal under the cursor is "
-             "DISPLAYED (IDA's 'o'); works on the listing and on pseudocode "
-             "numbers. 'show' reports the format and the stops without editing",
+    "default,target?,word?,line?,col?} how the literal under the cursor is "
+    "DISPLAYED (IDA's 'o'); works on the listing and on pseudocode "
+    "numbers. 'show' reports the format and the stops without editing",
 }
 
 #: `opfmt` modes that have a real key on the code views. Driving the key keeps
 #: the pane honest (a viewer sees the same thing a human would do); the named
 #: formats have no key, so those go through the view's action directly.
 _OPFMT_KEYS = {"cycle": "o", "back": "O"}
-_OPFMT_MODES = ("cycle", "back", "show", "hex", "dec", "oct", "bin", "char",
-                "offset", "stack", "default")
+_OPFMT_MODES = (
+    "cycle",
+    "back",
+    "show",
+    "hex",
+    "dec",
+    "oct",
+    "bin",
+    "char",
+    "offset",
+    "stack",
+    "default",
+)
 
 # `define` kinds -> the ListingView key that runs them. Driving the real key
 # keeps the pane honest (a viewer sees the same thing a human would do) and
 # reuses the app's own edit worker, which reports what actually happened.
 _DEFINE_KEYS = {
-    "code": "c",          # make code (runs until flow/undecodable)
-    "func": "p",          # make function
+    "code": "c",  # make code (runs until flow/undecodable)
+    "func": "p",  # make function
     "undef": "u",
-    "thumb": "t",         # flip ARM/Thumb at the cursor, then disassemble
-    "thumbscan": "T",     # find Thumb entry pointers in a vector table
+    "thumb": "t",  # flip ARM/Thumb at the cursor, then disassemble
+    "thumbscan": "T",  # find Thumb entry pointers in a vector table
     "data": "d",
     "string": "a",
 }
 
 # Movement keys — driven fast (no typed delay) so the pane still visibly moves.
 _MOVE_KEYS = {
-    "down": "j", "up": "k", "left": "h", "right": "l",
-    "word": "w", "wordback": "b", "bol": "0", "eol": "dollar_sign",
-    "top": "home", "bottom": "G",
-    "halfdown": "ctrl+d", "halfup": "ctrl+u",
-    "pagedown": "pagedown", "pageup": "pageup",
+    "down": "j",
+    "up": "k",
+    "left": "h",
+    "right": "l",
+    "word": "w",
+    "wordback": "b",
+    "bol": "0",
+    "eol": "dollar_sign",
+    "top": "home",
+    "bottom": "G",
+    "halfdown": "ctrl+d",
+    "halfup": "ctrl+u",
+    "pagedown": "pagedown",
+    "pageup": "pageup",
 }
 
 
@@ -148,8 +190,11 @@ def graph_info(app, blocks: bool = True) -> dict[str, Any]:
     rather than the box-drawing characters it is rendered as."""
     gv = app.query_one(GraphView)
     if gv.fc is None or gv.lay is None:
-        return {"open": app.is_graph, "loaded": False,
-                "note": "press space (or graph {action:'open'}) on a function"}
+        return {
+            "open": app.is_graph,
+            "loaded": False,
+            "note": "press space (or graph {action:'open'}) on a function",
+        }
     lay, fc = gv.lay, gv.fc
     out: dict[str, Any] = {
         "open": app.is_graph,
@@ -158,24 +203,30 @@ def graph_info(app, blocks: bool = True) -> dict[str, Any]:
         "zoom": gv.ZOOMS[gv._zoom],
         "canvas": {"w": lay.width, "h": lay.height},
         "stats": dict(lay.stats),
-        "cursor": {"block": gv.cursor_node, "row": gv.cursor_row,
-                   "ea": gv._cursor_ea(), "word": gv.word_under_cursor()},
+        "cursor": {
+            "block": gv.cursor_node,
+            "row": gv.cursor_row,
+            "ea": gv._cursor_ea(),
+            "word": gv.word_under_cursor(),
+        },
     }
     if blocks:
         rows = []
         for n in lay.nodes:
             b = gv._blocks.get(n.id)
-            rows.append({
-                "id": n.id,
-                "start": b.start if b else None,
-                "end": b.end if b else None,
-                "insns": len(b.rows) if b else 0,
-                "rank": n.rank,
-                "box": {"x": n.x, "y": n.y, "w": n.w, "h": n.h},
-                "succs": [{"id": i, "kind": k} for i, k in lay.succ.get(n.id, [])],
-                "preds": [{"id": i, "kind": k} for i, k in lay.pred.get(n.id, [])],
-                "selfloop": bool(b and any(d == n.id for d, _ in b.succs)),
-            })
+            rows.append(
+                {
+                    "id": n.id,
+                    "start": b.start if b else None,
+                    "end": b.end if b else None,
+                    "insns": len(b.rows) if b else 0,
+                    "rank": n.rank,
+                    "box": {"x": n.x, "y": n.y, "w": n.w, "h": n.h},
+                    "succs": [{"id": i, "kind": k} for i, k in lay.succ.get(n.id, [])],
+                    "preds": [{"id": i, "kind": k} for i, k in lay.pred.get(n.id, [])],
+                    "selfloop": bool(b and any(d == n.id for d, _ in b.succs)),
+                }
+            )
         out["blocks"] = rows
     return out
 
@@ -186,8 +237,20 @@ _MODALS = ("XrefsScreen", "SymbolPalette", "StructEditor", "ConfirmScreen")
 #: Handlers that did ``int(...)`` coped; the ones that compared directly blew up
 #: with e.g. "'<' not supported between instances of 'int' and 'str'". Coerce the
 #: known-numeric names once, centrally, instead of at every call site.
-_INT_PARAMS = ("lines", "limit", "max", "n", "index", "line", "col",
-               "occurrence", "delay_ms", "direction", "addr", "count")
+_INT_PARAMS = (
+    "lines",
+    "limit",
+    "max",
+    "n",
+    "index",
+    "line",
+    "col",
+    "occurrence",
+    "delay_ms",
+    "direction",
+    "addr",
+    "count",
+)
 _FLOAT_PARAMS = ("timeout",)
 
 
@@ -221,13 +284,16 @@ def _modal_snapshot(app) -> dict[str, Any] | None:
     if isinstance(items, list):
         try:
             from textual.widgets import OptionList
+
             hl = scr.query_one(OptionList).highlighted
         except Exception:  # noqa: BLE001
             hl = None
         info["highlighted"] = hl
         info["items"] = [
-            {"ea": (it[0] if isinstance(it[0], int) else None),
-             "label": str(it[1]) if len(it) > 1 else str(it)}
+            {
+                "ea": (it[0] if isinstance(it[0], int) else None),
+                "label": str(it[1]) if len(it) > 1 else str(it),
+            }
             for it in items[:64]
         ]
     return info
@@ -235,14 +301,23 @@ def _modal_snapshot(app) -> dict[str, Any] | None:
 
 def _cursor_info(app, w) -> dict[str, Any]:
     if isinstance(w, HexView):
-        return {"kind": "hex", "va": (w.cursor_va() if w.model else None),
-                "byte": w.cursor}
+        return {
+            "kind": "hex",
+            "va": (w.cursor_va() if w.model else None),
+            "byte": w.cursor,
+        }
     if isinstance(w, GraphView):
         # The graph cursor is (block, row), not a line index -- reporting it as
         # one would make a driver's `cursor line=` land somewhere arbitrary.
-        return {"kind": "graph", "ea": w._cursor_ea(), "block": w.cursor_node,
-                "row": w.cursor_row, "col": w.cursor_x,
-                "word": w.word_under_cursor(), "text": w._line_plain()}
+        return {
+            "kind": "graph",
+            "ea": w._cursor_ea(),
+            "block": w.cursor_node,
+            "row": w.cursor_row,
+            "col": w.cursor_x,
+            "word": w.word_under_cursor(),
+            "text": w._line_plain(),
+        }
     # disasm / decomp share the ColumnCursor surface
     word = None
     try:
@@ -254,9 +329,15 @@ def _cursor_info(app, w) -> dict[str, Any]:
         ea = app._line_ea_for(w)
     except Exception:  # noqa: BLE001
         pass
-    return {"kind": app._active, "line": w.cursor, "col": w.cursor_x,
-            "word": word, "ea": ea, "total": getattr(w, "total", None),
-            "scroll_y": round(w.scroll_offset.y)}
+    return {
+        "kind": app._active,
+        "line": w.cursor,
+        "col": w.cursor_x,
+        "word": word,
+        "ea": ea,
+        "total": getattr(w, "total", None),
+        "scroll_y": round(w.scroll_offset.y),
+    }
 
 
 def _where(app) -> str:
@@ -290,12 +371,12 @@ def snapshot(app) -> dict[str, Any]:
         pass
     return {
         "active": app._active,
-        "pref": app._code_mode(),   # kept for wire compat; a constant now
+        "pref": app._code_mode(),  # kept for wire compat; a constant now
         "function": ({"ea": cur.ea, "name": cur.name} if cur else None),
         "cursor": _cursor_info(app, w),
         "status": st,
         "filter": app._filter_term,
-        "binary": app._binary,      # None outside project mode
+        "binary": app._binary,  # None outside project mode
         "nav_depth": len(app._nav),
         "hops": list(getattr(app, "_hops", [])),
         "dirty": bool(app._dirty),
@@ -309,13 +390,18 @@ def view_lines(app, lines: int | None = None) -> dict[str, Any]:
     for hex use screen())."""
     w = _active_widget(app)
     if isinstance(w, HexView):
-        return {"active": "hex", "note": "use screen() for the hex grid",
-                "cursor": _cursor_info(app, w)}
+        return {
+            "active": "hex",
+            "note": "use screen() for the hex grid",
+            "cursor": _cursor_info(app, w),
+        }
     if isinstance(w, GraphView):
-        return {"active": "graph", "note": "use graph() for structure, "
-                                           "screen() for the drawing",
-                "cursor": _cursor_info(app, w),
-                "graph": graph_info(app, blocks=False)}
+        return {
+            "active": "graph",
+            "note": "use graph() for structure, screen() for the drawing",
+            "cursor": _cursor_info(app, w),
+            "graph": graph_info(app, blocks=False),
+        }
     top = round(w.scroll_offset.y)
     height = w.size.height or 40
     n = min(lines or height, max(w.total - top, 0))
@@ -323,21 +409,39 @@ def view_lines(app, lines: int | None = None) -> dict[str, Any]:
     for r in range(n):
         idx = top + r
         plain = w._line_plain(idx)
-        out.append({"i": idx, "cur": idx == w.cursor,
-                    "text": plain if plain is not None else ""})
-    return {"active": app._active, "top": top, "total": w.total,
-            "cursor": _cursor_info(app, w), "lines": out}
+        out.append(
+            {
+                "i": idx,
+                "cur": idx == w.cursor,
+                "text": plain if plain is not None else "",
+            }
+        )
+    return {
+        "active": app._active,
+        "top": top,
+        "total": w.total,
+        "cursor": _cursor_info(app, w),
+        "lines": out,
+    }
 
 
 def screen_text(app, fmt: str = "text") -> dict[str, Any]:
     """Render the whole screen exactly as shown. ``fmt``: 'text' (plain, default),
     'html' or 'svg' (colored — handy for an out-of-band web viewer)."""
     width, height = app.size
-    console = Console(width=width, height=height or 40, file=io.StringIO(),
-                      force_terminal=True, color_system="truecolor", record=True,
-                      legacy_windows=False, safe_box=False)
+    console = Console(
+        width=width,
+        height=height or 40,
+        file=io.StringIO(),
+        force_terminal=True,
+        color_system="truecolor",
+        record=True,
+        legacy_windows=False,
+        safe_box=False,
+    )
     render = app.screen._compositor.render_update(
-        full=True, screen_stack=app._background_screens, simplify=False)
+        full=True, screen_stack=app._background_screens, simplify=False
+    )
     console.print(render)
     out: dict[str, Any] = {"width": width, "height": height, "format": fmt}
     if fmt == "html":
@@ -388,8 +492,10 @@ def cursor_on(app, word: str, line: int | None = None, occurrence: int = 1) -> b
     if isinstance(w, HexView):
         raise ValueError("cursor_on: not supported in the hex view")
     if isinstance(w, GraphView):
-        raise ValueError("cursor_on: not supported in the graph view — use "
-                         "graph {action:'block'} or goto")
+        raise ValueError(
+            "cursor_on: not supported in the graph view — use "
+            "graph {action:'block'} or goto"
+        )
     if isinstance(w, DecompView):
         texts = list(w._texts)
     else:
@@ -412,7 +518,7 @@ def cursor_on(app, word: str, line: int | None = None, occurrence: int = 1) -> b
             if w.word_under_cursor() == word:
                 hits += 1
                 if hits >= max(1, occurrence):
-                    place_cursor(w)   # scrolls: an off-screen cursor edits blind
+                    place_cursor(w)  # scrolls: an off-screen cursor edits blind
                     return True
             col = t.find(word, col + 1)
     w.cursor, w.cursor_x = orig  # not found: leave the cursor untouched
@@ -455,8 +561,14 @@ def pseudocode(app, target=None) -> dict[str, Any]:
     if dea is None:
         return {"ea": None, "error": "no target"}
     d = app.program.decompile(dea)
-    return {"ea": dea, "name": (fn.name if fn else None), "failed": d.failed,
-            "error": d.error, "truncated": d.truncated, "code": d.code}
+    return {
+        "ea": dea,
+        "name": (fn.name if fn else None),
+        "failed": d.failed,
+        "error": d.error,
+        "truncated": d.truncated,
+        "code": d.code,
+    }
 
 
 def disassembly(app, target=None, max_lines: int = 2000) -> dict[str, Any]:
@@ -469,13 +581,26 @@ def disassembly(app, target=None, max_lines: int = 2000) -> dict[str, Any]:
     m = app.program.disasm(dea, fn.name if fn else None)
     total = m.total()
     lines = m.lines(0, min(total, max(1, max_lines)), prefetch=False)
-    return {"ea": dea, "name": (fn.name if fn else None), "total": total,
-            "lines": [{"ea": ln.ea, "text": ln.text} for ln in lines]}
+    return {
+        "ea": dea,
+        "name": (fn.name if fn else None),
+        "total": total,
+        "lines": [{"ea": ln.ea, "text": ln.text} for ln in lines],
+    }
 
 
 def _xref_dicts(xs, limit: int) -> list[dict[str, Any]]:
-    return [{"frm": x.frm, "to": x.to, "type": x.type, "kind": x.kind,
-             "fn_addr": x.fn_addr, "fn_name": x.fn_name} for x in xs[:limit]]
+    return [
+        {
+            "frm": x.frm,
+            "to": x.to,
+            "type": x.type,
+            "kind": x.kind,
+            "fn_addr": x.fn_addr,
+            "fn_name": x.fn_name,
+        }
+        for x in xs[:limit]
+    ]
 
 
 def xrefs_to(app, target, limit: int = 200) -> list[dict[str, Any]]:
@@ -497,9 +622,15 @@ def xrefs_from(app, target, limit: int = 200) -> list[dict[str, Any]]:
         for r in app.program.decompile(ea).refs[:limit]:
             tf = app.program.function_of(r.addr)
             is_func = bool(tf and tf.addr == r.addr)
-            out.append({"to": r.addr, "name": r.name or (tf.name if tf else None),
-                        "string": r.string, "is_func": is_func,
-                        "type": "code" if is_func else "data"})
+            out.append(
+                {
+                    "to": r.addr,
+                    "name": r.name or (tf.name if tf else None),
+                    "string": r.string,
+                    "is_func": is_func,
+                    "type": "code" if is_func else "data",
+                }
+            )
         return out
     return _xref_dicts(app.program.xrefs_from(ea), limit)
 
@@ -560,15 +691,25 @@ class RpcServer:
         except OSError:
             pass
 
-    async def _on_client(self, reader: asyncio.StreamReader,
-                         writer: asyncio.StreamWriter) -> None:
+    async def _on_client(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
         if self._busy:
             # No multi-driver support yet: refuse a second concurrent client
             # rather than let two drivers interleave mutations.
             try:
-                writer.write(json.dumps(
-                    {"id": None, "error": {"message": "busy: another client is "
-                     "connected (single-driver only)"}}).encode() + b"\n")
+                writer.write(
+                    json.dumps(
+                        {
+                            "id": None,
+                            "error": {
+                                "message": "busy: another client is "
+                                "connected (single-driver only)"
+                            },
+                        }
+                    ).encode()
+                    + b"\n"
+                )
                 await writer.drain()
                 writer.close()
             except Exception:  # noqa: BLE001
@@ -604,7 +745,11 @@ class RpcServer:
         except Exception as e:  # noqa: BLE001 — report, never kill the connection
             # ``str(KeyError("msg"))`` returns ``repr("msg")`` (adds quotes), which
             # mangles our friendly resolve messages; unwrap the single arg instead.
-            if isinstance(e, KeyError) and len(e.args) == 1 and isinstance(e.args[0], str):
+            if (
+                isinstance(e, KeyError)
+                and len(e.args) == 1
+                and isinstance(e.args[0], str)
+            ):
                 msg = e.args[0]
             else:
                 msg = str(e)
@@ -619,7 +764,8 @@ class RpcServer:
             # would go on to edit whatever the *previous* location was.
             raise TimeoutError(
                 f"{what or 'action'} did not complete within {timeout}s "
-                f"(still at {_where(self.app)}); retry with a larger timeout=")
+                f"(still at {_where(self.app)}); retry with a larger timeout="
+            )
         return snapshot(self.app)
 
     async def _graph(self, params, timeout):
@@ -641,18 +787,21 @@ class RpcServer:
                 return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)}
             if action == "close" and not app.is_graph:
                 return {**snapshot(app), "graph": graph_info(app, blocks=want_blocks)}
-            want = "graph" if action in ("open", "toggle") and \
-                not app.is_graph else None
+            want = (
+                "graph" if action in ("open", "toggle") and not app.is_graph else None
+            )
             res = await self._press(
                 ["space"],
-                (lambda: app.is_graph) if want else
-                (lambda: not app.is_graph),
-                timeout, f"graph {action}")
+                (lambda: app.is_graph) if want else (lambda: not app.is_graph),
+                timeout,
+                f"graph {action}",
+            )
             return {**res, "graph": graph_info(app, blocks=want_blocks)}
 
         if not app.is_graph:
-            raise ValueError(f"graph {action}: the graph is not open "
-                             f"(graph {{action:'open'}} first)")
+            raise ValueError(
+                f"graph {action}: the graph is not open (graph {{action:'open'}} first)"
+            )
         if action == "zoom":
             before = gv._zoom
             await self._press(["z"], lambda: gv._zoom != before, timeout, "graph zoom")
@@ -660,9 +809,12 @@ class RpcServer:
             await self._press(["0"], None, timeout, "graph entry")
         elif action in ("succ", "pred"):
             before = gv.cursor_node
-            await self._press(["J" if action == "succ" else "K"],
-                              lambda: gv.cursor_node != before, timeout,
-                              f"graph {action}")
+            await self._press(
+                ["J" if action == "succ" else "K"],
+                lambda: gv.cursor_node != before,
+                timeout,
+                f"graph {action}",
+            )
         elif action == "block":
             target = params.get("target")
             if target is None:
@@ -694,18 +846,23 @@ class RpcServer:
         """Open a prompt (a keystroke), optionally clear its prefill, type the
         value with the typed-out delay, submit. Returns after the prompt closes."""
         from textual.widgets import Input
+
         app = self.app
         await app._press_keys([open_key])
-        await settle(app, lambda: app.query_one(f"#{input_id}", Input).display, timeout=10)
+        await settle(
+            app, lambda: app.query_one(f"#{input_id}", Input).display, timeout=10
+        )
         inp = app.query_one(f"#{input_id}", Input)
         if not inp.display:
             # Say *why*. The old message always blamed the word under the cursor,
             # which sent readers hunting for a cursor problem when the real cause
             # was usually a modal eating the opening keystroke.
             modal = type(app.screen).__name__
-            why = (f"modal {modal!r} has focus and ate the {open_key!r} keystroke"
-                   if modal in _MODALS or modal != "Screen"
-                   else "no renameable token under the cursor")
+            why = (
+                f"modal {modal!r} has focus and ate the {open_key!r} keystroke"
+                if modal in _MODALS or modal != "Screen"
+                else "no renameable token under the cursor"
+            )
             raise RuntimeError(f"{input_id!r} prompt did not open: {why}")
         if clear:
             inp.value = ""
@@ -727,14 +884,14 @@ class RpcServer:
         app = self.app
         items = params.get("items")
         src = params.get("file")
-        if isinstance(items, str):        # `drive raw` hands params through as text
+        if isinstance(items, str):  # `drive raw` hands params through as text
             items = json.loads(items)
         if items is None:
             if not src:
                 raise ValueError("rename_many needs items=[{addr,name}] or file=<json>")
             with open(os.path.expanduser(str(src))) as f:
                 items = json.load(f)
-        if isinstance(items, dict):       # {"0x4370": "name"} is a natural shape too
+        if isinstance(items, dict):  # {"0x4370": "name"} is a natural shape too
             items = [{"addr": k, "name": v} for k, v in items.items()]
         if not isinstance(items, list) or not items:
             raise ValueError("rename_many: items must be a non-empty list")
@@ -745,8 +902,14 @@ class RpcServer:
                 skipped += 1
                 continue
             # Accept the field names symbol files actually use.
-            addr = next((it[k] for k in ("addr", "start", "ea", "address")
-                         if it.get(k) is not None), None)
+            addr = next(
+                (
+                    it[k]
+                    for k in ("addr", "start", "ea", "address")
+                    if it.get(k) is not None
+                ),
+                None,
+            )
             name = it.get("name") or it.get("label")
             if addr is None or not name:
                 skipped += 1
@@ -764,8 +927,15 @@ class RpcServer:
         # or the TUI freezes for the length of the batch.
         res = await asyncio.to_thread(app.program.client.invoke, "rename", batch=batch)
         summary = res.get("summary", {}) if isinstance(res, dict) else {}
-        failed = [r for r in (res.get("func") or []) if isinstance(r, dict)
-                  and r.get("error")] if isinstance(res, dict) else []
+        failed = (
+            [
+                r
+                for r in (res.get("func") or [])
+                if isinstance(r, dict) and r.get("error")
+            ]
+            if isinstance(res, dict)
+            else []
+        )
 
         # Names live in the IDB, but every cache in front of it is now stale --
         # including Hex-Rays', which is per-function and does NOT notice that a
@@ -779,18 +949,23 @@ class RpcServer:
         app.program.bump_names()
         app.program.invalidate_functions()
         app._func_index = None
-        app._load_functions()             # re-streams the function table
+        app._load_functions()  # re-streams the function table
         await settle(app, timeout=timeout)
         app._dirty = True
-        app._status(f"renamed {summary.get('ok', 0)} symbols"
-                    + (f", {len(failed)} failed" if failed else "")
-                    + "   (Ctrl+S to save)")
+        app._status(
+            f"renamed {summary.get('ok', 0)} symbols"
+            + (f", {len(failed)} failed" if failed else "")
+            + "   (Ctrl+S to save)"
+        )
         snap = snapshot(app)
         snap["rename_many"] = {
-            "requested": len(ops), "skipped": skipped,
-            "ok": summary.get("ok", 0), "failed": summary.get("failed", 0),
-            "errors": [{"addr": r.get("addr"), "error": r.get("error")}
-                       for r in failed[:10]],
+            "requested": len(ops),
+            "skipped": skipped,
+            "ok": summary.get("ok", 0),
+            "failed": summary.get("failed", 0),
+            "errors": [
+                {"addr": r.get("addr"), "error": r.get("error")} for r in failed[:10]
+            ],
         }
         return snap
 
@@ -810,13 +985,31 @@ class RpcServer:
     #: Verbs that drive the *main* app by injecting keystrokes. If a modal is on
     #: top it eats those keys, so they must refuse rather than silently no-op.
     _NEEDS_NO_MODAL = {
-        "goto", "open", "rename", "comment", "retype", "follow", "back",
-        "toggle_view", "hex", "save", "search", "move", "cursor", "cursor_on",
-        "define", "opfmt",
+        "goto",
+        "open",
+        "rename",
+        "comment",
+        "retype",
+        "follow",
+        "back",
+        "toggle_view",
+        "hex",
+        "save",
+        "search",
+        "move",
+        "cursor",
+        "cursor_on",
+        "define",
+        "opfmt",
     }
     #: Modals the driver is expected to interact with (they have their own verbs).
-    _DRIVABLE_MODALS = {"XrefsScreen", "SymbolPalette", "StructEditor",
-                        "ProjectPalette", "QuitScreen"}
+    _DRIVABLE_MODALS = {
+        "XrefsScreen",
+        "SymbolPalette",
+        "StructEditor",
+        "ProjectPalette",
+        "QuitScreen",
+    }
 
     def _modal_kind(self) -> str | None:
         scr = self.app.screen
@@ -833,15 +1026,20 @@ class RpcServer:
                     f"modal {modal!r} is on top and will swallow this verb's "
                     f"keystrokes; dismiss it first (close) or use its own verb "
                     f"(select/symbols/xrefs). Note: a binary with no entry "
-                    f"function can land in the symbol palette on startup.")
+                    f"function can land in the symbol palette on startup."
+                )
         if method in (None, "ping"):
             module = None
             try:
                 module = app._module() if app.client else None
             except Exception:  # noqa: BLE001
                 pass
-            return {"ok": True, "proto": PROTO_VERSION, "module": module,
-                    **_readiness(app)}
+            return {
+                "ok": True,
+                "proto": PROTO_VERSION,
+                "module": module,
+                **_readiness(app),
+            }
         if method == "methods":
             return METHODS
         if method == "quit":
@@ -856,14 +1054,18 @@ class RpcServer:
 
             def _go():
                 if dirty and save:
-                    app._on_quit_choice("save")   # saves, then exits
+                    app._on_quit_choice("save")  # saves, then exits
                 else:
                     app._on_quit_choice("discard")
 
             # answer first, then tear down (so this response still gets written)
             asyncio.get_running_loop().call_later(0.2, _go)
-            return {"ok": True, "quitting": True, "saving": bool(dirty and save),
-                    "dirty": dirty}
+            return {
+                "ok": True,
+                "quitting": True,
+                "saving": bool(dirty and save),
+                "dirty": dirty,
+            }
 
         if method in _PROGRAM_METHODS and app.program is None:
             raise ValueError("not ready: still connecting / loading functions")
@@ -902,8 +1104,10 @@ class RpcServer:
             if params.get("clear"):
                 diag.clear()
                 return {"cleared": True}
-            return {"recent": diag.recent(int(params.get("n", 10))),
-                    "log": os.environ.get("IDATUI_LOG") or None}
+            return {
+                "recent": diag.recent(int(params.get("n", 10))),
+                "log": os.environ.get("IDATUI_LOG") or None,
+            }
 
         if method == "trace":
             tc = app.trace_ctl
@@ -916,12 +1120,19 @@ class RpcServer:
                 if isinstance(v, str) and v.startswith("!"):
                     idx = int(float(v[1:]) * (t.length - 1) / 100.0)
                 else:
-                    idx = int(str(v).replace(",", ""), 0) if isinstance(v, str) else int(v)
+                    idx = (
+                        int(str(v).replace(",", ""), 0)
+                        if isinstance(v, str)
+                        else int(v)
+                    )
                 tc.seek(idx)
-            elif "goto" in params:      # first execution of an address/name
+            elif "goto" in params:  # first execution of an address/name
                 tgt = params["goto"]
-                ea = (int(str(tgt), 0) if str(tgt).lower().startswith("0x")
-                      else app.program.resolve(str(tgt)))
+                ea = (
+                    int(str(tgt), 0)
+                    if str(tgt).lower().startswith("0x")
+                    else app.program.resolve(str(tgt))
+                )
                 first = t.first_execution(ea)
                 if first is None:
                     raise ValueError(f"{tgt} never executed in this trace")
@@ -933,9 +1144,12 @@ class RpcServer:
                     (tc.step_over if over else tc.step)(1 if n > 0 else -1)
             await settle(app, timeout=float(params.get("timeout", 20.0)))
             snap = snapshot(app)
-            snap["trace"] = {"idx": tc.t, "length": t.length,
-                             "pc": hex(t.ip(tc.t)),
-                             "changed": sorted(t.changed(tc.t))}
+            snap["trace"] = {
+                "idx": tc.t,
+                "length": t.length,
+                "pc": hex(t.ip(tc.t)),
+                "changed": sorted(t.changed(tc.t)),
+            }
             return snap
 
         if method == "binaries":
@@ -943,12 +1157,20 @@ class RpcServer:
                 raise ValueError("not a project session (launch with --project)")
             counts = app._index.counts() if app._index is not None else {}
             resident = set(app._pool.resident()) if app._pool is not None else set()
-            return {"active": app._binary, "hops": list(app._hops),
-                    "binaries": [{"label": r.label, "source": r.source,
-                                  "active": r.label == app._binary,
-                                  "resident": r.label in resident,
-                                  "indexed": int(counts.get(r.label, 0))}
-                                 for r in app._project.refs]}
+            return {
+                "active": app._binary,
+                "hops": list(app._hops),
+                "binaries": [
+                    {
+                        "label": r.label,
+                        "source": r.source,
+                        "active": r.label == app._binary,
+                        "resident": r.label in resident,
+                        "indexed": int(counts.get(r.label, 0)),
+                    }
+                    for r in app._project.refs
+                ],
+            }
 
         if method == "switch":
             if app._project is None:
@@ -965,30 +1187,41 @@ class RpcServer:
             else:
                 # Same path a project search hit takes, so it records a hop and
                 # Esc comes back here.
-                app._switch_then_goto(label, int(str(addr), 0)
-                                      if isinstance(addr, str) else int(addr))
-            await settle(app, lambda: app._binary == label
-                         and app._func_index is not None
-                         and app._func_index.complete,
-                         timeout=float(params.get("timeout", 300.0)))
+                app._switch_then_goto(
+                    label, int(str(addr), 0) if isinstance(addr, str) else int(addr)
+                )
+            await settle(
+                app,
+                lambda: (
+                    app._binary == label
+                    and app._func_index is not None
+                    and app._func_index.complete
+                ),
+                timeout=float(params.get("timeout", 300.0)),
+            )
             return snapshot(app)
 
         # -- structured introspection (heavy: run off the UI loop) -------- #
         loop = asyncio.get_running_loop()
         if method == "pseudocode":
-            return await loop.run_in_executor(None, pseudocode, app, params.get("target"))
+            return await loop.run_in_executor(
+                None, pseudocode, app, params.get("target")
+            )
         if method == "disassembly":
             mx = int(params.get("max", 2000))
             return await loop.run_in_executor(
-                None, disassembly, app, params.get("target"), mx)
+                None, disassembly, app, params.get("target"), mx
+            )
         if method == "xrefs_to":
             lim = int(params.get("limit", 200))
             return await loop.run_in_executor(
-                None, xrefs_to, app, params.get("target"), lim)
+                None, xrefs_to, app, params.get("target"), lim
+            )
         if method == "xrefs_from":
             lim = int(params.get("limit", 200))
             return await loop.run_in_executor(
-                None, xrefs_from, app, params.get("target"), lim)
+                None, xrefs_from, app, params.get("target"), lim
+            )
         if method == "resolve":
             return await loop.run_in_executor(None, resolve, app, params.get("name"))
 
@@ -998,15 +1231,24 @@ class RpcServer:
 
         # optional ergonomic: place the cursor on a token before an edit/follow
         if method in ("rename", "retype", "follow") and params.get("word"):
-            if not cursor_on(app, str(params["word"]), params.get("line"),
-                             int(params.get("occurrence", 1))):
-                raise ValueError(f"cursor_on: token {params['word']!r} not found "
-                                 "in the current view")
+            if not cursor_on(
+                app,
+                str(params["word"]),
+                params.get("line"),
+                int(params.get("occurrence", 1)),
+            ):
+                raise ValueError(
+                    f"cursor_on: token {params['word']!r} not found in the current view"
+                )
             await drain(app)
 
         if method == "cursor_on":
-            found = cursor_on(app, str(params["word"]), params.get("line"),
-                              int(params.get("occurrence", 1)))
+            found = cursor_on(
+                app,
+                str(params["word"]),
+                params.get("line"),
+                int(params.get("occurrence", 1)),
+            )
             await drain(app)
             snap = snapshot(app)
             snap["found"] = found
@@ -1024,7 +1266,8 @@ class RpcServer:
                 # on the function the caller *used* to be looking at.
                 raise TimeoutError(
                     f"goto {target!r} did not land within {timeout}s "
-                    f"(still at {_where(app)}); retry with a larger timeout=")
+                    f"(still at {_where(app)}); retry with a larger timeout="
+                )
             return snapshot(app)
 
         if method == "define":
@@ -1032,80 +1275,95 @@ class RpcServer:
             if kind not in _DEFINE_KEYS:
                 raise ValueError(
                     f"unknown define kind {kind!r}; one of "
-                    f"{', '.join(sorted(_DEFINE_KEYS))}")
+                    f"{', '.join(sorted(_DEFINE_KEYS))}"
+                )
             target = params.get("target")
             if target not in (None, ""):
                 # Land on the address first. A raw image is mostly *undefined*,
                 # so the target usually has no name and no function — the goto
                 # predicate can't be address-based, only "we moved".
-                await self._fill_prompt("g", "goto", str(target), delay,
-                                        clear=False)
+                await self._fill_prompt("g", "goto", str(target), delay, clear=False)
                 await settle(app, timeout=timeout)
             if app.is_hex:
                 # backslash leaves hex for the code view (which may be decomp).
-                await self._press(["backslash"],
-                                  lambda: not app.is_hex, timeout,
-                                  "leave the hex view")
+                await self._press(
+                    ["backslash"], lambda: not app.is_hex, timeout, "leave the hex view"
+                )
             if app.is_decomp:
                 # These bindings live on the listing; in the decompiler the key
                 # would be swallowed or do something else entirely.
-                await self._press(["tab"], lambda: app.is_listing,
-                                  timeout, "switch to the listing")
+                await self._press(
+                    ["tab"], lambda: app.is_listing, timeout, "switch to the listing"
+                )
             if not app.is_listing:
                 raise RuntimeError(
                     f"define needs the listing view, but the active pane is "
-                    f"{app._active!r}")
-            snap = await self._press([_DEFINE_KEYS[kind]], timeout=timeout,
-                                     what=f"define {kind}")
+                    f"{app._active!r}"
+                )
+            snap = await self._press(
+                [_DEFINE_KEYS[kind]], timeout=timeout, what=f"define {kind}"
+            )
             snap["define"] = {"kind": kind, "status": snap.get("status", "")}
             return snap
 
         if method == "opfmt":
             mode = str(params.get("mode", "cycle")).lower()
             if mode not in _OPFMT_MODES:
-                raise ValueError(f"unknown opfmt mode {mode!r}; one of "
-                                 f"{', '.join(_OPFMT_MODES)}")
+                raise ValueError(
+                    f"unknown opfmt mode {mode!r}; one of {', '.join(_OPFMT_MODES)}"
+                )
             target = params.get("target")
             if target not in (None, ""):
-                await self._fill_prompt("g", "goto", str(target), delay,
-                                        clear=False)
+                await self._fill_prompt("g", "goto", str(target), delay, clear=False)
                 await settle(app, timeout=timeout)
             if app.is_hex:
-                await self._press(["backslash"], lambda: not app.is_hex,
-                                  timeout, "leave the hex view")
+                await self._press(
+                    ["backslash"], lambda: not app.is_hex, timeout, "leave the hex view"
+                )
             view = _active_widget(app)
             if isinstance(view, HexView):
                 raise RuntimeError("opfmt needs a code view, not the hex view")
             if params.get("word"):
                 # Land the column on the literal first: WHICH operand gets
                 # reformatted is decided by where the cursor is.
-                if not cursor_on(app, str(params["word"]), params.get("line"),
-                                 int(params.get("occurrence", 1) or 1)):
+                if not cursor_on(
+                    app,
+                    str(params["word"]),
+                    params.get("line"),
+                    int(params.get("occurrence", 1) or 1),
+                ):
                     raise RuntimeError(
                         f"{params['word']!r} is not on screen in this view, so "
-                        f"there is no literal to reformat")
+                        f"there is no literal to reformat"
+                    )
                 await drain(app)
             elif params.get("line") is not None or params.get("col") is not None:
                 place_cursor(view, params.get("line"), params.get("col"))
                 await drain(app)
             before = _where(app)
             if mode in _OPFMT_KEYS:
-                snap = await self._press([_OPFMT_KEYS[mode]], timeout=timeout,
-                                         what=f"opfmt {mode}")
+                snap = await self._press(
+                    [_OPFMT_KEYS[mode]], timeout=timeout, what=f"opfmt {mode}"
+                )
             else:
                 view.focus()
                 view.action_op_format(mode)
                 await settle(app, timeout=timeout)
                 snap = snapshot(app)
-            snap["opfmt"] = {"mode": mode, "at": before,
-                             "status": snap.get("status", "")}
+            snap["opfmt"] = {
+                "mode": mode,
+                "at": before,
+                "status": snap.get("status", ""),
+            }
             return snap
 
         if method == "rename_many":
             return await self._rename_many(params, timeout)
 
         if method == "rename":
-            await self._fill_prompt("n", "rename", str(params["name"]), delay, clear=True)
+            await self._fill_prompt(
+                "n", "rename", str(params["name"]), delay, clear=True
+            )
             await settle(app, timeout=timeout)
             return snapshot(app)
         if method == "comment":
@@ -1115,19 +1373,21 @@ class RpcServer:
             # the Input widget.  The app's _do_comment converts the two-char
             # sequence '\n' into a real newline for IDA, so we escape here.
             ctext = str(params["text"]).replace("\n", "\\n")
-            await self._fill_prompt("semicolon", "comment", ctext, 0,
-                                    clear=True)
+            await self._fill_prompt("semicolon", "comment", ctext, 0, clear=True)
             await settle(app, timeout=timeout)
             return snapshot(app)
         if method == "retype":
-            await self._fill_prompt("y", "retype", str(params["proto"]), delay, clear=True)
+            await self._fill_prompt(
+                "y", "retype", str(params["proto"]), delay, clear=True
+            )
             await settle(app, timeout=timeout)
             return snapshot(app)
 
         if method == "follow":
             depth = len(app._nav)
-            return await self._press(["enter"], lambda: len(app._nav) > depth,
-                                     timeout, "follow")
+            return await self._press(
+                ["enter"], lambda: len(app._nav) > depth, timeout, "follow"
+            )
         if method == "back":
             return await self._press(["escape"], timeout=timeout)
         if method == "toggle_view":
@@ -1158,17 +1418,23 @@ class RpcServer:
             # call that LEAVES hex could never be satisfied and always timed
             # out -- a driver could open the hex view but never close it.
             was_hex = app.is_hex
-            return await self._press(["backslash"], lambda: app.is_hex != was_hex,
-                                     timeout, "hex")
+            return await self._press(
+                ["backslash"], lambda: app.is_hex != was_hex, timeout, "hex"
+            )
         if method == "graph":
             return await self._graph(params, timeout)
         if method == "xrefs":
             return await self._press(
-                ["x"], lambda: type(app.screen).__name__ == "XrefsScreen",
-                timeout, "xrefs")
+                ["x"],
+                lambda: type(app.screen).__name__ == "XrefsScreen",
+                timeout,
+                "xrefs",
+            )
         if method == "symbols":
             await app._press_keys(["ctrl+n"])
-            await settle(app, lambda: type(app.screen).__name__ == "SymbolPalette", timeout=10)
+            await settle(
+                app, lambda: type(app.screen).__name__ == "SymbolPalette", timeout=10
+            )
             q = params.get("query")
             if q:
                 await app._press_keys(_text_to_keys(str(q), delay))
@@ -1176,10 +1442,14 @@ class RpcServer:
             return snapshot(app)
         if method == "structs":
             return await self._press(
-                ["ctrl+t"], lambda: type(app.screen).__name__ == "StructEditor",
-                timeout, "structs")
+                ["ctrl+t"],
+                lambda: type(app.screen).__name__ == "StructEditor",
+                timeout,
+                "structs",
+            )
         if method == "find":
             from . import search as _search
+
             q = str(params.get("query", ""))
             forced = params.get("mode")
             forced = None if forced in (None, "auto") else str(forced)
@@ -1190,33 +1460,57 @@ class RpcServer:
                     raise ValueError(f"find: {problem}")
                 cleaned = _search.normalise_pattern(cleaned)
             hits, err, truncated = await asyncio.to_thread(
-                app.program.search, cleaned, mode,
+                app.program.search,
+                cleaned,
+                mode,
                 limit=int(params.get("limit", 500)),
-                regex=bool(params.get("regex")), case=bool(params.get("case")))
+                regex=bool(params.get("regex")),
+                case=bool(params.get("case")),
+            )
             if err:
                 raise ValueError(f"find: {err}")
-            return {"mode": mode, "query": cleaned, "truncated": truncated,
-                    "hits": [{"addr": hex(h.addr), "head": hex(h.head),
-                              "line": h.line, "func": h.func,
-                              "seg": h.seg} for h in hits]}
+            return {
+                "mode": mode,
+                "query": cleaned,
+                "truncated": truncated,
+                "hits": [
+                    {
+                        "addr": hex(h.addr),
+                        "head": hex(h.head),
+                        "line": h.line,
+                        "func": h.func,
+                        "seg": h.seg,
+                    }
+                    for h in hits
+                ],
+            }
         if method == "export":
             # Deliberately NOT driven through the prompt: this is the one verb
             # whose whole point is the file it leaves behind, and a driver needs
             # the path back, not a screenshot of a prompt closing.
             from . import findings
+
             path = params.get("path")
             app.journal.load(app.program)
             app.journal.flush(app.program)
             out, f = await asyncio.to_thread(
-                findings.export, app.program, app._open_path or "",
+                findings.export,
+                app.program,
+                app._open_path or "",
                 str(path) if path else None,
-                types=bool(params.get("types", True)), journal=app.journal)
+                types=bool(params.get("types", True)),
+                journal=app.journal,
+            )
             app._status(f"exported findings → {out}", priority=True)
             await drain(app)
-            return {"path": out, "comments": len(f.comments),
-                    "names": len(findings._user_names(f)),
-                    "types": len(f.types), "functions": f.n_functions,
-                    "bytes": os.path.getsize(out) if os.path.exists(out) else 0}
+            return {
+                "path": out,
+                "comments": len(f.comments),
+                "names": len(findings._user_names(f)),
+                "types": len(f.types),
+                "functions": f.n_functions,
+                "bytes": os.path.getsize(out) if os.path.exists(out) else 0,
+            }
         if method == "close":
             return await self._press(["escape"], timeout=timeout)
         if method == "save":
@@ -1224,13 +1518,16 @@ class RpcServer:
 
         if method == "search":
             term = str(params.get("term", ""))
-            open_key = "slash" if int(params.get("direction", 1)) >= 0 else "question_mark"
+            open_key = (
+                "slash" if int(params.get("direction", 1)) >= 0 else "question_mark"
+            )
             await self._fill_prompt(open_key, "search", term, delay, clear=True)
             await settle(app, timeout=timeout)
             return snapshot(app)
 
         if method == "select":
             from textual.widgets import OptionList
+
             scr = app.screen
             if type(scr).__name__ not in _MODALS:
                 raise ValueError("select: no modal list is open")
@@ -1249,8 +1546,10 @@ class RpcServer:
         if method == "move":
             key = _MOVE_KEYS.get(str(params.get("dir")))
             if key is None:
-                raise ValueError(f"unknown move dir: {params.get('dir')!r} "
-                                 f"(one of {sorted(_MOVE_KEYS)})")
+                raise ValueError(
+                    f"unknown move dir: {params.get('dir')!r} "
+                    f"(one of {sorted(_MOVE_KEYS)})"
+                )
             n = max(1, int(params.get("n", 1)))
             await app._press_keys([key] * n)
             if params.get("settle", True):