listing: make every undefined byte its own row, so you can carve anywhere

Loading a blob and pressing `c` at 0 disassembles one instruction; everything
after it collapsed into a single row — "db 2044 dup(?)" — with no cursor
position anywhere inside it. There was no way to start a second instruction
stream at an arbitrary offset, which is most of what carving a firmware image
IS. In IDA every undefined byte is its own line and you just put the cursor on
one.

The collapse existed for a real reason (see the comment in the heads tool): a
.bss or a fresh blob would otherwise be millions of one-byte rows, and this
model materialises what it walks. Expanding physically would also make
`g <far address>` walk every byte in between.

So the run stays ONE physical head and PRESENTS as N logical rows. _row_at is a
prefix sum over heads, _phys() maps a row back to (head, byte offset), and the
text for an interior row is synthesised on demand — "db 4Ah", the actual value,
because the byte values are the whole point when you're looking for a stream.
Memory is unchanged (libcrypto: 1 head for its 80-byte .bss, 61MB RSS), and
index_of_ea into the middle of a run is 0.01ms via bisect.

Now: cursor on any byte, `c`, and you get an instruction; the bytes before it
stay individually addressable.

Two bugs found on the way:

* IDAToolError takes (tool, message) and five call sites in domain.py passed one
  string. Every one of those error paths raised TypeError INSTEAD of the real
  error — "define code @ 0x4020: Failed to create instruction" reached the user
  as "IDAToolError.__init__() missing 1 required positional argument". Fixed all
  five; the message that finally came through is what identified the next issue.
* Searching now walks one row per undefined byte, so _index_for_search is capped
  at 400k lines and says when it truncated, rather than grinding through a
  multi-megabyte blob nobody wants to text-search.

tests: test_blob_ui.py +8 — a run presents one row per byte, each is a single
addressable byte showing its value, an interior address resolves to its own row,
`c` on a chosen byte carves there, the carved row spans the instruction, and
neighbouring bytes stay addressable. Uses PLANTED A64 instructions, because
whether random bytes decode is chance and a test that depends on chance is
worthless on the run where it fails. 19/0 blob, 202/0 scenarios, 30/0 project UI.

3 files changed, 210 insertions, 24 deletions

diff --git a/idatui/app.py b/idatui/app.py
index f1b4dec..4cf0ee0 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -783,13 +783,24 @@ class DisasmView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=True
             return
         texts: list[str] = []
         off, total = 0, self.total
+        # A freshly-loaded blob is one row per undefined byte, so "all lines" can
+        # be millions of `db 4Ah` that nobody searches for. Index a bounded
+        # prefix rather than hang; say so instead of silently finding nothing.
+        LIMIT = 400_000
+        capped = total > LIMIT
+        total = min(total, LIMIT)
         while off < total:
-            lines = model.lines(off, DisasmModel.BLOCK, prefetch=False)
+            lines = model.lines(off, min(DisasmModel.BLOCK, total - off),
+                                prefetch=False)
             if not lines:
                 break
             texts.extend(self._fmt(ln) for ln in lines)
             off += len(lines)
         self._search_texts = texts
+        if capped:
+            self.app.call_from_thread(
+                self._app_status,
+                f"search covers the first {LIMIT:,} lines of {self.total:,}")
         self.app.call_from_thread(done)
 
     @work(thread=True, exclusive=True, group="disasm-prime")
diff --git a/idatui/domain.py b/idatui/domain.py
index 681542e..846af89 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -583,6 +583,16 @@ class ListingModel:
         self.name = name or f"seg @ {seg_start:#x}"
         self._heads: list[Head] = []
         self._by_ea: dict[int, int] = {}
+        # Logical rows != physical heads. A run of undefined bytes arrives as ONE
+        # head ("db 2044 dup(?)") because materialising millions of one-byte rows
+        # for a .bss would be absurd — but you must still be able to put the
+        # cursor on any byte in it and press `c`, exactly as in IDA. So a run of
+        # N bytes PRESENTS as N rows and the text for each is synthesised on
+        # demand. _row_at[i] is the logical row where physical head i starts.
+        self._row_at: list[int] = []
+        self._head_eas: list[int] = []      # parallel to _heads, for bisect
+        self._rows = 0                      # total logical rows loaded
+        self._ubytes: dict[int, bytes] = {}  # lazily-read bytes for those rows
         self._next: int | None = seg_start  # next address to fetch from
         self._done = False
         self._max_raw = 0  # widest opcode length (bytes) seen, for the op column
@@ -649,14 +659,16 @@ class ListingModel:
                 continue
         page = self._attach_opcode_bytes(page)
         with self._lock:
-            base = len(self._heads)
-            for i, h in enumerate(page):
+            for h in page:
                 # Banner/label rows (function headers, separators, code labels)
                 # are display-only; don't index them so navigation lands on the
                 # real code/data head at that address.
                 if h.kind not in ("sep", "funchdr", "label"):
-                    self._by_ea.setdefault(h.ea, base + i)
+                    self._by_ea.setdefault(h.ea, self._rows)
+                self._row_at.append(self._rows)
+                self._head_eas.append(h.ea)
                 self._heads.append(h)
+                self._rows += self._span(h)
             nxt = cur.get("next")
             if nxt is None:
                 self._done = True
@@ -665,9 +677,67 @@ class ListingModel:
                 self._next = _as_int(nxt)
             return len(rows)
 
+    @staticmethod
+    def _span(h: Head) -> int:
+        """How many logical rows head ``h`` occupies."""
+        return h.size if (h.kind == "unknown" and h.size > 1) else 1
+
+    def _phys(self, row: int) -> tuple[int, int]:
+        """(physical head index, byte offset into it) for logical ``row``."""
+        import bisect
+        i = bisect.bisect_right(self._row_at, row) - 1
+        if i < 0:
+            return (-1, 0)
+        return (i, row - self._row_at[i])
+
+    def _unknown_bytes(self, ea: int, n: int) -> bytes:
+        """Bytes behind an undefined run, read in blocks and cached.
+
+        Undefined rows are the ones you carve, so their VALUES are the whole
+        point — "db ?" with no byte tells you nothing about where an instruction
+        stream might start.
+        """
+        BLK = 1024
+        out = bytearray()
+        a = ea
+        while len(out) < n:
+            b0 = (a // BLK) * BLK
+            blk = self._ubytes.get(b0)
+            if blk is None:
+                try:
+                    blk = self._prog.read_bytes(b0, BLK)
+                except Exception:  # noqa: BLE001
+                    blk = b""
+                self._ubytes[b0] = blk
+            off = a - b0
+            take = min(BLK - off, n - len(out))
+            chunk = blk[off:off + take] if blk else b""
+            if not chunk:
+                break
+            out += chunk
+            a += len(chunk)
+        return bytes(out)
+
+    def _row_head(self, i: int, off: int) -> Head:
+        """The Head for one logical row: the physical head, or a synthesised
+        single-byte row inside an undefined run.
+
+        The run's FIRST row is synthesised too. Leaving "db 2044 dup(?)" there
+        would say the row covers 2044 bytes when it now covers one, and the
+        column of byte values would start an address late.
+        """
+        h = self._heads[i]
+        if self._span(h) == 1:
+            return h
+        ea = h.ea + off
+        b = self._unknown_bytes(ea, 1)
+        text = f"db {b[0]:02X}h" if b else "db ?"
+        return Head(ea=ea, kind="unknown", size=1, text=text,
+                    name=h.name if off == 0 else None)
+
     def ensure(self, n: int) -> None:
-        """Ensure at least ``n`` heads are loaded (or all, if fewer exist)."""
-        while not self._done and len(self._heads) < n:
+        """Ensure at least ``n`` logical rows are loaded (or all, if fewer)."""
+        while not self._done and self._rows < n:
             if self._load_next_page() == 0:
                 break
 
@@ -679,8 +749,9 @@ class ListingModel:
             if idx >= 0:
                 return idx
             with self._lock:
-                have = len(self._heads)
-                last_ea = self._heads[-1].ea if self._heads else -1
+                have = self._rows
+                last_ea = (self._heads[-1].ea + max(self._heads[-1].size, 1) - 1
+                           if self._heads else -1)
                 done = self._done
             if done or (have and last_ea >= ea):
                 # Loaded past ea without an exact head hit: return the first head
@@ -691,12 +762,19 @@ class ListingModel:
 
     def _first_at_or_after(self, ea: int) -> int:
         with self._lock:
-            heads = self._heads
-            for i, h in enumerate(heads):
+            j = self._head_index_at(ea)
+            if j >= 0:
+                h = self._heads[j]
                 if h.ea <= ea < h.ea + max(h.size, 1):
-                    return i
+                    off = (ea - h.ea) if self._span(h) > 1 else 0
+                    return self._row_at[j] + off
+            for i, h in enumerate(self._heads):
+                if h.ea <= ea < h.ea + max(h.size, 1):
+                    # Inside an undefined run, land on the exact BYTE.
+                    off = (ea - h.ea) if self._span(h) > 1 else 0
+                    return self._row_at[i] + off
                 if h.ea > ea:
-                    return i
+                    return self._row_at[i]
         return -1
 
     def load_all(self, progress: Callable[[int], None] | None = None) -> None:
@@ -704,7 +782,7 @@ class ListingModel:
             if self._load_next_page() == 0:
                 break
             if progress:
-                progress(len(self._heads))
+                progress(self._rows)
 
     @property
     def complete(self) -> bool:
@@ -713,23 +791,58 @@ class ListingModel:
 
     def loaded(self) -> int:
         with self._lock:
-            return len(self._heads)
+            return self._rows
 
     def __len__(self) -> int:
         return self.loaded()
 
     def get(self, i: int) -> Head | None:
         with self._lock:
-            return self._heads[i] if 0 <= i < len(self._heads) else None
+            if not (0 <= i < self._rows):
+                return None
+            j, off = self._phys(i)
+            if j < 0:
+                return None
+            span = self._span(self._heads[j])
+            h = self._heads[j]
+        # Synthesis reads bytes, so do it OUTSIDE the lock: an RPC under the
+        # model lock deadlocks the page loader that is filling it.
+        return self._row_head(j, off) if span > 1 else h
 
     def window(self, start: int, count: int) -> list[Head]:
+        """``count`` logical rows from ``start`` (synthesising undefined ones)."""
         self.ensure(start + count)
         with self._lock:
-            return list(self._heads[start:start + count])
+            rows = min(self._rows, start + count)
+            spans = [self._phys(i) for i in range(max(start, 0), max(rows, 0))]
+            heads = self._heads
+            plain = [(j, off, heads[j]) for j, off in spans if j >= 0]
+        return [self._row_head(j, off) if self._span(h) > 1 else h
+                for j, off, h in plain]
 
     def index_of_ea(self, ea: int) -> int:
         with self._lock:
-            return self._by_ea.get(ea, -1)
+            hit = self._by_ea.get(ea)
+            if hit is not None:
+                return hit
+            # An address INSIDE an undefined run is a real row now, not a
+            # mid-item address: that is what makes `g <addr>` + `c` work
+            # anywhere in a blob. Heads are address-ordered, so bisect rather
+            # than scan — a big listing has hundreds of thousands of them and
+            # this is on the navigation path.
+            j = self._head_index_at(ea)
+            if j >= 0:
+                h = self._heads[j]
+                if self._span(h) > 1 and h.ea <= ea < h.ea + h.size:
+                    return self._row_at[j] + (ea - h.ea)
+        return -1
+
+    def _head_index_at(self, ea: int) -> int:
+        """Index of the physical head containing ``ea`` (caller holds the lock)."""
+        import bisect
+        eas = self._head_eas
+        i = bisect.bisect_right(eas, ea) - 1
+        return i if 0 <= i < len(self._heads) else -1
 
     # -- DisasmModel-compatible accessors (unified model) ------------------ #
     def cached_line(self, idx: int) -> Head | None:
@@ -741,7 +854,7 @@ class ListingModel:
 
     def is_cached(self, start: int, count: int) -> bool:
         with self._lock:
-            return start + count <= len(self._heads)
+            return start + count <= self._rows
 
     def ensure_async(self, start: int, count: int) -> None:
         pass  # the background grower streams the rest in; nothing to prefetch
@@ -1251,14 +1364,14 @@ class Program:
         res = self._first_result(
             self.client.call("define_code", items=[{"addr": hex(ea)}]))
         if res.get("error"):
-            raise IDAToolError(f"define code @ {ea:#x}: {res['error']}")
+            raise IDAToolError("define_code", f"@ {ea:#x}: {res['error']}")
 
     def define_func(self, ea: int) -> None:
         """Create a function starting at ``ea`` (IDA's 'p')."""
         res = self._first_result(
             self.client.call("define_func", items=[{"addr": hex(ea)}]))
         if res.get("error"):
-            raise IDAToolError(f"create function @ {ea:#x}: {res['error']}")
+            raise IDAToolError("define_func", f"@ {ea:#x}: {res['error']}")
 
     def undefine(self, ea: int, size: int | None = None) -> None:
         """Undefine the item at ``ea`` back to raw bytes (IDA's 'u')."""
@@ -1267,7 +1380,7 @@ class Program:
             item["size"] = int(size)
         res = self._first_result(self.client.call("undefine", items=[item]))
         if res.get("error"):
-            raise IDAToolError(f"undefine @ {ea:#x}: {res['error']}")
+            raise IDAToolError("undefine", f"@ {ea:#x}: {res['error']}")
 
     def make_data(self, ea: int, type_decl: str, name: str | None = None) -> None:
         """Create a typed data item at ``ea`` (IDA's 'd', but typed). ``type_decl``
@@ -1278,7 +1391,7 @@ class Program:
         res = self._first_result(self.client.call("make_data", items=[item]))
         if res.get("ok") is False or res.get("error"):
             raise IDAToolError(
-                f"make data @ {ea:#x}: {res.get('error') or 'rejected'}")
+                "make_data", f"@ {ea:#x}: {res.get('error') or 'rejected'}")
 
     def make_string(self, ea: int, length: int = 0, kind: str = "c") -> str:
         """Create a string literal at ``ea`` (IDA's 'A'); auto-length when 0.
@@ -1287,7 +1400,7 @@ class Program:
         res = r if isinstance(r, dict) else {}
         if not res.get("ok"):
             raise IDAToolError(
-                f"make string @ {ea:#x}: {res.get('error') or 'rejected'}")
+                "make_string", f"@ {ea:#x}: {res.get('error') or 'rejected'}")
         return res.get("text", "")
 
     def region_label(self, ea: int) -> str:
diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py
index e8d2533..400c9ab 100644
--- a/tests/test_blob_ui.py
+++ b/tests/test_blob_ui.py
@@ -44,9 +44,22 @@ async def wait(pred, pilot, t=240.0):
 
 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))
+        # -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
+            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(os.urandom(64 * 1024))     # random: IDA will find no code
+            f.write(bytes(data))
 
         # Skip the dialog by answering up front; this test is about what
         # happens AFTER a described blob turns out to contain nothing.
@@ -85,6 +98,55 @@ async def run() -> int:
             check("the hint survives navigating", "no functions" in status2,
                   status2[:90])
 
+            # -- byte-granular carving ---------------------------------- #
+            # An undefined run arrives as ONE head ("db N dup(?)"). It has to
+            # PRESENT as one row per byte or there is no cursor position to
+            # press `c` at, which is how IDA works and the only way to find an
+            # instruction stream that doesn't start at the run's first byte.
+            m = lst.model
+            check("an undefined run presents one row per byte",
+                  m.loaded() >= 4096 and len(m._heads) < 64,
+                  f"rows={m.loaded()} physical heads={len(m._heads)}")
+            rows = m.window(0, 4)
+            check("each row is a single addressable byte",
+                  [h.ea for h in rows] == [0x4000, 0x4001, 0x4002, 0x4003]
+                  and all(h.size == 1 for h in rows),
+                  f"{[(hex(h.ea), h.size) for h in rows]}")
+            check("and shows its value, not a placeholder",
+                  all(h.text.startswith("db ") and "dup" not in h.text
+                      for h in rows), f"{[h.text for h in rows]}")
+
+            # The point of all this: land on an arbitrary byte and convert it.
+            i = m.index_of_ea(0x4021)
+            check("an address inside the run resolves to its own row",
+                  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
+            lst.cursor = m.index_of_ea(target)
+            lst._scroll_cursor_into_view()
+            await pilot.pause(0.1)
+            check("the cursor sits on the byte we aimed at",
+                  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.
+            m = lst.model
+            h = m.get(m.index_of_ea(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}")
+            if h is not None and h.kind == "code":
+                print(f"       carved {target:#x}: {h.text}")
+                check("the carved row spans the instruction, not one byte",
+                      h.size == 4, f"size={h.size}")
+                check("bytes before it stay individually addressable",
+                      m.get(m.index_of_ea(target - 1)).size == 1
+                      and m.get(m.index_of_ea(target - 1)).ea == target - 1)
+
             await pilot.press("ctrl+l")
             opened = await wait(lambda: isinstance(app.screen, ConfirmScreen), pilot, 20)
             check("Ctrl+L offers to reload with different options", opened,