Merge the IDA Code Mode port

Replaces the private idalib worker (idatui/worker.py + worker_client.py, with
server/patch_server.py injecting tools into ida-pro-mcp) with an ordinary
client of ida_codemode.client.DatabaseHandle. A database open by an IDA GUI is
reused; otherwise Code Mode starts or shares a managed idalib worker. The TUI
no longer owns an IDA process, and closing it releases only its lease.

Based on Duncan Ogilvie's port, rebased onto ~150 commits of local work it
predated. The rebase itself was mechanical; landing it was not. Nine defects
had to be fixed before the feature set was whole again, none of which the
patch's own tests could catch:

  - DatabaseHandle.open() takes image_base, not loading_address: every
    connect() would have raised TypeError on the first call
  - five operations our tree had grown were simply missing (flowchart, so the
    graph view was dead; op_format/pc_nums/pc_num_format, so 'o'/'O' were;
    survey_binary)
  - set_comments wrote only the disassembly comment, so comments never
    appeared in pseudocode
  - xref_query returned rows in raw IDA order, and 'follow the call' silently
    followed the fall-through instead
  - rename accepted one edit per category, so bulk symbol import was dead
  - decompile ran decomp_map's full per-column ctree sweep to fill in a
    per-line address anchor
  - heads shipped without operand extents or the digest protocol
  - the package became unimportable without ida_codemode installed, which
    killed the offline test suites

Verified against the pre-codemode tag rather than against assumptions: the
full suite is 788 passed / 0 failed, and the pilot's 301 checks match the old
backend exactly. Performance is within 2x on the listing hot path and faster
on decompile, disasm and connect, after fixing two runtime costs that are
documented for upstream in docs/CODEMODE_UPSTREAM.md.

Test runtime came down from ~9m20s to 115s along the way -- not by removing
checks, but by removing four kinds of waiting-on-a-guess that were also
hiding real failures.

1 files changed, 83 insertions, 26 deletions

diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py
index f60e914..1055e5a 100644
--- a/tests/test_blob_ui.py
+++ b/tests/test_blob_ui.py
@@ -17,10 +17,13 @@ import sys
 import tempfile
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
 from textual.widgets import Input, Static  # noqa: E402
 
 from idatui.app import ConfirmScreen, IdaTui, ListingView  # noqa: E402
+from _fixtures import staged, synthetic  # noqa: E402
+from idatui._sync import settle  # noqa: E402
 
 PASS = FAIL = 0
 
@@ -46,28 +49,63 @@ async def wait(pred, pilot, t=240.0):
     return False
 
 
-async def run() -> int:
-    with tempfile.TemporaryDirectory() as tmp:
-        # Random bytes so IDA finds no functions... but with REAL AArch64
-        # instructions planted at a known offset. Whether arbitrary random bytes
-        # happen to decode is chance, and a test that depends on chance tells you
-        # nothing on the run where it fails.
-        data = bytearray(os.urandom(64 * 1024))
+#: File offset of the planted instruction run -> ea 0x4000 + PLANTED.
+PLANTED = 0x40
+
+
+def _blob_bytes() -> bytes:
+    """A DETERMINISTIC pseudo-random blob with real AArch64 instructions planted.
+
+    Seeded, not os.urandom: the bytes must be identical every run or the
+    pristine-database cache can never apply (this suite used to pay ~40s of
+    auto-analysis per run because the content, and the path, changed each time).
+    Determinism also removes a genuine flake -- whether 64KB of chance bytes
+    contains something IDA reads as a function is luck, and "and really has no
+    functions" is asserted below.
+    """
+    import random
+    data = bytearray(random.Random(0xB10BCAFE).randbytes(64 * 1024))
         # -parm puts IDA in AArch64 mode, so these are A64 encodings; the ARM32
         # spelling of a nop (0xE1A00000) is NOT decodable there and made this
         # test fail for a reason that had nothing to do with what it checks.
-        planted = 0x40                       # file offset -> ea 0x4040
-        for k, insn in enumerate((0xD503201F,   # nop
-                                  0xD503201F,   # nop
-                                  0xD65F03C0)):  # ret  <- the run must stop here
-            data[planted + k * 4:planted + k * 4 + 4] = insn.to_bytes(4, "little")
-        blob = os.path.join(tmp, "rnd.bin")
-        with open(blob, "wb") as f:
-            f.write(bytes(data))
+    for k, insn in enumerate((0xD503201F,   # nop
+                              0xD503201F,   # nop
+                              0xD65F03C0)):  # ret  <- the run must stop here
+        data[PLANTED + k * 4:PLANTED + k * 4 + 4] = insn.to_bytes(4, "little")
+    return bytes(data)
+
+
+#: -parm puts IDA in AArch64 mode (the ARM32 spelling of a nop is not decodable
+#: there); -b400 sets the image base. The cached database must be built with the
+#: SAME switches, so both go through one factory.
+BLOB_ARGS = "-parm -b400"
+
 
+def _blob_app(path):
+    return IdaTui(open_path=path, keepalive=False, load_args=BLOB_ARGS)
+
+
+def head_at(lst, ea):
+    """The listing row for ``ea`` off the LIVE model, or None.
+
+    Always re-reads ``lst.model``: an edit may rebuild the model, and holding
+    the old object shows pre-edit rows -- which looks exactly like the edit
+    silently failing.
+    """
+    m = lst.model
+    if m is None:
+        return None
+    i = m.index_of_ea(ea)
+    return m.get(i) if i is not None and i >= 0 else None
+
+
+async def run() -> int:
+    blob_src = synthetic("rnd.bin", _blob_bytes)
+    # staged() analyses once ever and copies the result in on later runs.
+    async with staged(blob_src, _blob_app) as blob:
         # Skip the dialog by answering up front; this test is about what
         # happens AFTER a described blob turns out to contain nothing.
-        app = IdaTui(open_path=blob, keepalive=False, load_args="-parm -b400")
+        app = _blob_app(blob)
         async with app.run_test(size=(140, 44)) as pilot:
             ok = await wait(lambda: app._func_index is not None
                             and app._func_index.complete, pilot)
@@ -126,7 +164,7 @@ async def run() -> int:
                   i >= 0 and m.get(i).ea == 0x4021,
                   f"row={i} ea={m.get(i).ea if i >= 0 else None}")
 
-            target = 0x4000 + planted        # a NOP we put there ourselves
+            target = 0x4000 + PLANTED        # a NOP we put there ourselves
             lst.cursor = m.index_of_ea(target)
             lst._scroll_cursor_into_view()
             await pilot.pause(0.1)
@@ -134,12 +172,18 @@ async def run() -> int:
                   lst._cursor_ea() == target,
                   f"{lst._cursor_ea():#x} want {target:#x}")
             await pilot.press("c")
-            await pilot.pause(2.0)
-            # Defining an item REBUILDS the listing model, so re-read it from the
-            # view: holding the old object shows the pre-edit rows and looks
-            # exactly like the edit silently failing.
+            # settle(), not a fixed sleep AND not a bare predicate: an edit can
+            # look done for a moment and then be replaced when a queued listing
+            # rebuild lands, so the gate has to be "the row is code AND the app
+            # has stopped working". settle() is the same helper the app's own
+            # RPC layer uses, so tests and driver agree on what "done" means.
+            await settle(app, lambda: (lambda h: h is not None and h.kind == "code")(
+                head_at(lst, target)), timeout=30)
+            # Re-read the model: defining an item rebuilds it, and holding the
+            # old object shows pre-edit rows -- which looks exactly like the
+            # edit silently failing.
             m = lst.model
-            h = m.get(m.index_of_ea(target))
+            h = head_at(lst, target)
             check("`c` on a chosen byte carves an instruction there",
                   h is not None and h.kind == "code",
                   f"kind={h.kind if h else None} text={h.text if h else None!r}")
@@ -184,9 +228,17 @@ async def run() -> int:
             for ch in "note":
                 await pilot.press(ch)
             await pilot.press("enter")
-            await wait(lambda: lst.model is not old and lst.model is not None,
-                       pilot, 30)
-            await pilot.pause(0.4)
+            # Wait for the COMMENT ITSELF to show up, not for the model object to
+            # be replaced: a comment now re-renders the listing in place (the
+            # walk is kept), so `model is not old` never becomes true and this
+            # burned its full 30s timeout on every run -- after which the check
+            # below passed vacuously, because nothing had happened at all.
+            # The prompt closing plus quiescence is the real end of the edit.
+            # (The listing re-renders its text lazily, so the comment is not
+            # necessarily visible in model rows the moment the worker returns --
+            # which is why this waits for the app, not for the text.)
+            await settle(app, lambda: not app.query_one("#comment", Input).display,
+                         timeout=30)
             check("commenting leaves the view where it was",
                   lst.model.get(round(lst.scroll_offset.y)).ea == ctop
                   and lst._cursor_ea() == ccur,
@@ -208,7 +260,12 @@ async def run() -> int:
             check("scrolled somewhere with rows above us",
                   round(lst.scroll_offset.y) > 0, f"top={lst.scroll_offset.y}")
             await pilot.press("c")
-            await pilot.pause(2.5)
+            # No predicate here on purpose: this spot is random data, so the
+            # carve may legitimately produce nothing and "the row became code"
+            # would never hold (it timed out for 30s and then passed anyway).
+            # What is being checked is that the VIEW did not move, so the gate
+            # is simply "the app has finished reacting".
+            await settle(app, timeout=30)
             m2 = lst.model
             top_after = m2.get(round(lst.scroll_offset.y)).ea
             check("carving leaves the scroll position where it was",