server+domain: heads walker + ListingModel — flat code/data listing (M1)

The keystone for a real disassembly-listing view (unlike DisasmModel, which
is one function, code-only). Two pieces:

server/patch_server.py: inject a `heads` tool. It walks item heads over a
segment with next_head/prev_head and renders each via generate_disasm_line,
so it returns a flat listing where code, data (db/dw/dd, strings, jump
tables) and undefined bytes all appear as typed rows {ea,kind,size,text,name}.
Unlike `disasm` (code-only, bails at the first data byte) it shows the whole
segment. Address-paged: chain forward via cursor.next, page up with back=true
(returns the N heads ending before addr, in forward order, + cursor.prev).

domain.py: ListingModel — a lazily-grown, segment-scoped head index
(FunctionIndex-style forward paging via cursor.next; line index == position
in the walked list). ensure_ea() gives random access to an address (resolving
a mid-item byte to its containing head). New Head dataclass, Program.listing()
(cached per segment) + segment_bounds(); section_of() now derives from it;
bump_items() clears the listing cache too.

Verified live: heads renders strings/jump-tables/unknown correctly, forward
chaining + back-paging work; ListingModel walks echo .text (5036 heads,
~276ms) with cached windows and mid-item address resolution. tests/test_domain
gains a [listing] section (27 passed). Pilot hex/region_define/disasm_nav green.

Note: adding a server tool needs a supervisor restart (workers respawn).

3 files changed, 299 insertions, 1 deletions

diff --git a/idatui/domain.py b/idatui/domain.py
index 53fd907..fd673ca 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -79,6 +79,28 @@ class Line:
         )
 
 
+@dataclass(frozen=True)
+class Head:
+    """One flat-listing item (from the ``heads`` server tool): a code
+    instruction, a data item, or an undefined byte run."""
+
+    ea: int
+    kind: str            # 'code' | 'data' | 'unknown'
+    size: int
+    text: str
+    name: str | None = None
+
+    @classmethod
+    def from_raw(cls, d: dict) -> "Head":
+        return cls(
+            ea=_as_int(d["ea"]),
+            kind=d.get("kind", "unknown"),
+            size=int(d.get("size", 0) or 0),
+            text=d.get("text", ""),
+            name=d.get("name"),
+        )
+
+
 @dataclass
 class Ref:
     addr: int
@@ -470,6 +492,127 @@ class DisasmModel:
 
 
 # --------------------------------------------------------------------------- #
+# Listing model: lazily-grown flat listing (code + data + undefined) per segment
+# --------------------------------------------------------------------------- #
+class ListingModel:
+    """A flat, IDA-style disassembly *listing* over one segment: code, data and
+    undefined heads interleaved, unlike ``DisasmModel`` (one function, code only).
+
+    Backed by the injected ``heads`` server tool, which walks item heads and
+    renders each via ``generate_disasm_line``. The segment is walked lazily in
+    forward pages (``FunctionIndex`` style); line index == position in the walked
+    head list. Random access to an address is O(distance-from-seg-start) the
+    first time (then cached) — the same tradeoff as ``disasm offset=N``. Grows
+    on demand as the viewport scrolls. Synchronous + thread-safe.
+    """
+
+    PAGE = 500  # heads per server call (well under the tool's 2000 cap)
+
+    def __init__(self, program: "Program", seg_start: int, seg_end: int,
+                 name: str | None = None):
+        self._prog = program
+        self.seg_start = seg_start
+        self.seg_end = seg_end
+        self.name = name or f"seg @ {seg_start:#x}"
+        self._heads: list[Head] = []
+        self._by_ea: dict[int, int] = {}
+        self._next: int | None = seg_start  # next address to fetch from
+        self._done = False
+        self._lock = threading.Lock()
+
+    def _load_next_page(self) -> int:
+        with self._lock:
+            if self._done or self._next is None:
+                return 0
+            frm = self._next
+        payload = self._prog.client.call("heads", addr=hex(frm), count=self.PAGE)
+        rows = payload.get("heads", []) if isinstance(payload, dict) else []
+        cur = payload.get("cursor", {}) if isinstance(payload, dict) else {}
+        with self._lock:
+            base = len(self._heads)
+            for i, r in enumerate(rows):
+                try:
+                    h = Head.from_raw(r)
+                except (KeyError, ValueError, TypeError):
+                    continue
+                self._by_ea.setdefault(h.ea, base + i)
+                self._heads.append(h)
+            nxt = cur.get("next")
+            if nxt is None:
+                self._done = True
+                self._next = None
+            else:
+                self._next = _as_int(nxt)
+            return len(rows)
+
+    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:
+            if self._load_next_page() == 0:
+                break
+
+    def ensure_ea(self, ea: int) -> int:
+        """Walk forward until the head containing ``ea`` is loaded; return its
+        line index (or the nearest head at/after it), or -1 if past the end."""
+        while True:
+            idx = self.index_of_ea(ea)
+            if idx >= 0:
+                return idx
+            with self._lock:
+                have = len(self._heads)
+                last_ea = self._heads[-1].ea 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
+                # at/after ea (a mid-item address lands on its containing head).
+                return self._first_at_or_after(ea)
+            if self._load_next_page() == 0:
+                return self._first_at_or_after(ea)
+
+    def _first_at_or_after(self, ea: int) -> int:
+        with self._lock:
+            heads = self._heads
+            for i, h in enumerate(heads):
+                if h.ea <= ea < h.ea + max(h.size, 1):
+                    return i
+                if h.ea > ea:
+                    return i
+        return -1
+
+    def load_all(self, progress: Callable[[int], None] | None = None) -> None:
+        while not self._done:
+            if self._load_next_page() == 0:
+                break
+            if progress:
+                progress(len(self._heads))
+
+    @property
+    def complete(self) -> bool:
+        with self._lock:
+            return self._done
+
+    def loaded(self) -> int:
+        with self._lock:
+            return len(self._heads)
+
+    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
+
+    def window(self, start: int, count: int) -> list[Head]:
+        self.ensure(start + count)
+        with self._lock:
+            return list(self._heads[start:start + count])
+
+    def index_of_ea(self, ea: int) -> int:
+        with self._lock:
+            return self._by_ea.get(ea, -1)
+
+
+# --------------------------------------------------------------------------- #
 # Hex model: block-cached byte view over the loaded image (VA-addressed)
 # --------------------------------------------------------------------------- #
 class HexModel:
@@ -565,6 +708,7 @@ class Program:
         )
         self._indices: dict[str | None, FunctionIndex] = {}
         self._disasm: dict[int, DisasmModel] = {}
+        self._listings: dict[int, ListingModel] = {}  # keyed by segment start
         self._decomp: dict[int, tuple[Decompilation, int]] = {}
         self._name_gen = 0  # bumped on rename; invalidates stale name caches
         self._sections: list[tuple[int, int, str]] | None = None
@@ -680,14 +824,33 @@ class Program:
     def section_of(self, ea: int) -> str | None:
         """Name of the segment/section containing ``ea`` (e.g. '.got', '.text',
         '.data.rel.ro', 'LOAD'), or None if unmapped."""
+        b = self.segment_bounds(ea)
+        return b[2] if b else None
+
+    def segment_bounds(self, ea: int) -> tuple[int, int, str] | None:
+        """(start, end, name) of the segment containing ``ea``, or None."""
         secs = self.sections()
         if not secs:
             return None
         i = bisect.bisect_right([s[0] for s in secs], ea) - 1
         if 0 <= i < len(secs) and secs[i][0] <= ea < secs[i][1]:
-            return secs[i][2]
+            return secs[i]
         return None
 
+    def listing(self, ea: int) -> ListingModel | None:
+        """Flat listing (code+data+undefined) for the segment containing ``ea``,
+        cached per segment. None if ``ea`` is unmapped."""
+        seg = self.segment_bounds(ea)
+        if seg is None:
+            return None
+        start, end, name = seg
+        with self._lock:
+            m = self._listings.get(start)
+            if m is None:
+                m = ListingModel(self, start, end, name)
+                self._listings[start] = m
+            return m
+
     # -- structs / local types -------------------------------------------- #
     def list_structs(self, filter: str = "") -> list[Struct]:
         """All local structs/unions (optionally name-substring filtered), sorted
@@ -863,6 +1026,7 @@ class Program:
             self._name_gen += 1
             self._indices.clear()
             self._decomp.clear()
+            self._listings.clear()
             models = list(self._disasm.values())
             self._disasm.clear()
         for m in models:
diff --git a/server/patch_server.py b/server/patch_server.py
index f42cc3a..a13e6cd 100644
--- a/server/patch_server.py
+++ b/server/patch_server.py
@@ -178,6 +178,104 @@ def file_regions() -> dict:
         out.append({"start": hex(seg.start_ea), "end": hex(seg.end_ea), "file_off": fo})
         seg = ida_segment.get_next_seg(seg.start_ea)
     return {"regions": out}
+
+
+def _idatui_head_row(ea):
+    """One flat-listing row for the head at ``ea``: kind (code/data/unknown),
+    byte size, rendered text, and any symbol name."""
+    import ida_bytes
+    import ida_lines
+    import ida_name
+
+    f = ida_bytes.get_flags(ea)
+    if ida_bytes.is_code(f):
+        kind = "code"
+    elif ida_bytes.is_data(f):
+        kind = "data"
+    else:
+        kind = "unknown"
+    line = ida_lines.generate_disasm_line(ea, 0)
+    text = ida_lines.tag_remove(line) if line else ""
+    text = " ".join(text.split())  # collapse IDA's column padding
+    row = {
+        "ea": hex(ea),
+        "kind": kind,
+        "size": int(ida_bytes.get_item_size(ea)),
+        "text": text,
+    }
+    nm = ida_name.get_ea_name(ea)
+    if nm:
+        row["name"] = nm
+    return row
+
+
+@tool
+@idasync
+def heads(
+    addr: Annotated[str, "Start address or name to walk from"],
+    count: Annotated[int, "Max heads to return (default 200, max 2000)"] = 200,
+    offset: Annotated[int, "Skip first N heads from addr (default 0)"] = 0,
+    end: Annotated[str, "Optional exclusive end address; default = segment end"] = "",
+    back: Annotated[bool, "Walk backwards: return the count heads ENDING just before addr, in forward order"] = False,
+) -> dict:
+    """Walk item heads from ``addr`` as a flat listing: every head is rendered
+    (code OR data OR undefined) via generate_disasm_line and stepped with
+    next_head/prev_head. Unlike ``disasm`` (code-only, bails at the first data
+    byte) this shows db/dw/dd/... lines for data and undefined regions — IDA's
+    real disassembly view. Address-paged: page forward by re-calling with
+    ``addr`` = the returned cursor.next; page up with ``back=true``."""
+    import ida_bytes
+    import ida_segment
+    import idaapi
+
+    count = 2000 if count > 2000 else (1 if count < 1 else count)
+    offset = max(int(offset), 0)
+    try:
+        start = parse_address(addr)
+    except Exception as e:
+        return {"addr": str(addr), "error": str(e), "heads": [], "cursor": {"done": True}}
+    seg = ida_segment.getseg(start)
+    if not seg:
+        return {"addr": str(addr), "error": "no segment", "heads": [], "cursor": {"done": True}}
+    lo, hi = seg.start_ea, seg.end_ea
+    if end:
+        try:
+            hi = min(hi, parse_address(end))
+        except Exception:
+            pass
+
+    rows = []
+    if back:
+        # Collect up to (count+offset) heads strictly before `start`, then take
+        # the window closest to `start`, returned in forward order.
+        walk = []
+        cur = ida_bytes.prev_head(start, lo)
+        while cur != idaapi.BADADDR and cur >= lo and len(walk) < count + offset:
+            walk.append(cur)
+            cur = ida_bytes.prev_head(cur, lo)
+        walk.reverse()
+        chosen = walk[: len(walk) - offset] if offset else walk
+        chosen = chosen[-count:]
+        rows = [_idatui_head_row(e) for e in chosen]
+        first = chosen[0] if chosen else start
+        pea = ida_bytes.prev_head(first, lo)
+        cursor = {"done": True} if pea == idaapi.BADADDR or pea < lo else {"prev": hex(pea)}
+        return {"addr": str(addr), "heads": rows, "cursor": cursor}
+
+    ea = ida_bytes.get_item_head(start)
+    for _ in range(offset):
+        if ea >= hi or ea == idaapi.BADADDR:
+            break
+        ea = ida_bytes.next_head(ea, hi)
+    more = False
+    while ea != idaapi.BADADDR and ea < hi:
+        if len(rows) >= count:
+            more = True
+            break
+        rows.append(_idatui_head_row(ea))
+        ea = ida_bytes.next_head(ea, hi)
+    cursor = {"next": hex(ea)} if more else {"done": True}
+    return {"addr": str(addr), "heads": rows, "cursor": cursor}
 '''
 
 SNIPPET = f"{BEGIN}\n{BODY.strip()}\n{END}\n"
diff --git a/tests/test_domain.py b/tests/test_domain.py
index 6875b66..24cb781 100644
--- a/tests/test_domain.py
+++ b/tests/test_domain.py
@@ -146,6 +146,42 @@ def main(argv):
         except KeyError as e:
             check("resolve symbol name", False, str(e))
 
+    # ---- flat listing (code + data + undefined heads) ------------------ #
+    print("\n[listing]")
+    seg = prog.segment_bounds(fattest.addr)
+    check("segment_bounds finds the .text segment", seg is not None and seg[0] <= fattest.addr < seg[1],
+          str(seg))
+    lm = prog.listing(fattest.addr)
+    check("listing() returns a model for a mapped address", lm is not None)
+    if lm is not None:
+        lm.ensure(20)
+        w = lm.window(0, 20)
+        check("listing window returns heads", len(w) == 20, str(len(w)))
+        eas = [h.ea for h in w]
+        check("listing head addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])),
+              str(eas[:4]))
+        check("listing heads carry a kind", all(h.kind in ("code", "data", "unknown") for h in w),
+              str({h.kind for h in w}))
+        check("listing head sizes positive", all(h.size >= 1 for h in w),
+              str([h.size for h in w[:6]]))
+        # random access to a mid-segment address lands on the containing head
+        mid = w[10].ea
+        li = lm.ensure_ea(mid)
+        check("ensure_ea lands on the exact head for a head address",
+              li >= 0 and lm.get(li).ea == mid, f"idx={li}")
+        # a mid-item byte resolves to its containing head
+        if w[4].size > 1:
+            inside = w[4].ea + 1
+            li2 = lm.ensure_ea(inside)
+            check("ensure_ea resolves a mid-item byte to its head",
+                  li2 >= 0 and lm.get(li2).ea == w[4].ea, f"idx={li2} ea={w[4].ea:#x}")
+        dt_cold, _ = ms(lambda: lm.window(0, 20))
+        check("listing window revisit is cached/instant", dt_cold < 5, f"{dt_cold:.1f}ms")
+        dt_all, _ = ms(lambda: lm.load_all())
+        check("listing load_all completes the segment", lm.complete and len(lm) > 20,
+              f"n={len(lm)} complete={lm.complete}")
+        print(f"    {seg[2]}: {len(lm)} heads walked in {dt_all:.0f}ms")
+
     prog.close()
     print(f"\n{PASS} passed, {FAIL} failed")
     return 1 if FAIL else 0