listing: render opcode bytes (shared engine with disasm) — M4/UX step 1

The flat listing was a subset of the disasm view (no opcodes). Now the listing
carries an opcode-bytes column with the same format + o toggle: Head gains a raw
field, ListingModel fills it for code heads via one bulk read per page (bounded,
variable-length safe) + tracks the widest opcode; ListingView renders/pads the
column, settling width as pages stream in. render_line/_line_plain share the
addr+opcodes+label+mnem/text layout with DisasmView.

Test harness all_funcs() reloads the index if a prior bump_items() cleared it
(was crashing biggest() with max([])).

Verified: code heads carry raw (max_raw_len 11); listing suite + continuous_view
opcode-parity 26/0.

3 files changed, 96 insertions, 9 deletions

diff --git a/idatui/app.py b/idatui/app.py
index 415d00a..b556352 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -900,6 +900,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         Binding("pageup", "page(-1)", "PgUp", show=False),
         Binding("home", "goto_top", "Top", show=False),
         Binding("G,end", "goto_bottom", "Bottom", show=False),
+        Binding("o", "toggle_opcodes", "Opcodes", show=False),
         Binding("c", "define_code", "Code", show=False),
         Binding("d", "make_data", "Data", show=False),
         Binding("a", "make_string", "Str", show=False),
@@ -930,6 +931,8 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         self._term = ""
         self._matches: list[int] = []
         self._ranges: dict[int, list[tuple[int, int]]] = {}
+        self._show_ops = True   # opcode-bytes column (toggle 'o'), like disasm
+        self._op_w = 0          # char width of the hex-bytes field (excl. gap)
 
     # -- text helpers ------------------------------------------------------ #
     def _head(self, idx: int) -> Head | None:
@@ -939,12 +942,21 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
     def _name_prefix(h: Head) -> str:
         return f"{h.name}  " if h.name else ""
 
+    def _op_field(self, h: Head) -> str:
+        """The padded opcode-bytes column (empty when hidden / no bytes). Shared
+        format with the disasm view so cursor/search offsets line up."""
+        if not self._show_ops or self._op_w <= 0:
+            return ""
+        raw = h.raw or b""
+        return " ".join(f"{b:02X}" for b in raw).ljust(self._op_w) + "  "
+
     def _line_plain(self, idx: int) -> str | None:
         h = self._head(idx)
         if h is None:
             return None
         indent = "    " if h.kind == "member" else ""
-        return f"{h.ea:08X}  " + indent + self._name_prefix(h) + h.text
+        return (f"{h.ea:08X}  " + self._op_field(h) + indent
+                + self._name_prefix(h) + h.text)
 
     # -- public API -------------------------------------------------------- #
     def load(self, model: ListingModel, name: str, cursor: int = 0,
@@ -985,9 +997,30 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         else:
             self._scroll_cursor_into_view()
         self._pending_scroll_y = None
+        self._update_op_w()
         self.refresh()
         self.post_message(ListingView.CursorMoved(self.cursor, self._cursor_ea()))
 
+    def _update_op_w(self) -> bool:
+        """Recompute the opcode-column width from the widest code head seen.
+        Returns True if it changed (callers refresh + drop stale search index)."""
+        w = 0
+        if self.model is not None and self._show_ops:
+            mx = self.model.max_raw_len()
+            w = max(mx * 3 - 1, 0) if mx > 0 else 0
+        if w != self._op_w:
+            self._op_w = w
+            return True
+        return False
+
+    def action_toggle_opcodes(self) -> None:
+        self._show_ops = not self._show_ops
+        self._update_op_w()
+        self._ranges = {}  # column layout changed -> stale match offsets
+        self._clamp_x()
+        self.refresh()
+        self._app_status("opcodes " + ("on" if self._show_ops else "off"))
+
     @work(thread=True, exclusive=True, group="listing-grow")
     def _grow(self) -> None:
         """Stream the rest of the segment's heads in the background, growing the
@@ -1012,6 +1045,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
             return
         self.total = total
         self.virtual_size = Size(0, total)
+        self._update_op_w()  # more code streamed in -> widen the op column
         self.refresh()
 
     # -- search hooks ----------------------------------------------------- #
@@ -1051,6 +1085,11 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
             strip = Strip([Segment(f"  {idx:>8}  …", _S_DIM)])
         else:
             segs: list[Segment] = [Segment(f"{h.ea:08X}  ", _S_ADDR)]
+            op = self._op_field(h)
+            if op:
+                segs.append(Segment(op, _S_OPBYTES))
+            if h.kind == "member":
+                segs.append(Segment("    ", _S_MEMBER))
             if h.name:
                 segs.append(Segment(f"{h.name}  ", _S_LABEL))
             if h.kind == "code":
@@ -1061,7 +1100,7 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
             elif h.kind == "data":
                 segs.append(Segment(h.text, _S_DATA))
             elif h.kind == "member":
-                segs.append(Segment("    " + h.text, _S_MEMBER))
+                segs.append(Segment(h.text, _S_MEMBER))
             else:
                 segs.append(Segment(h.text, _S_UNK))
             strip = Strip(segs)
diff --git a/idatui/domain.py b/idatui/domain.py
index 3fe0c51..86fb47e 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -92,10 +92,11 @@ class Head:
     instruction, a data item, or an undefined byte run."""
 
     ea: int
-    kind: str            # 'code' | 'data' | 'unknown'
+    kind: str            # 'code' | 'data' | 'unknown' | 'member'
     size: int
     text: str
     name: str | None = None
+    raw: bytes | None = None  # opcode/item bytes (filled in for code by the model)
 
     @classmethod
     def from_raw(cls, d: dict) -> "Head":
@@ -538,11 +539,45 @@ class ListingModel:
         self._by_ea: dict[int, int] = {}
         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
         self._lock = threading.Lock()
         # Serializes page loads so a background grower and an in-view search can
         # both drive loading without double-fetching the same page.
         self._load_lock = threading.Lock()
 
+    # Opcode bytes are only worth showing for code; cap the bulk read so a page
+    # containing a huge coalesced undefined run doesn't pull megabytes.
+    _OP_SPAN_CAP = 1 << 16
+
+    def _attach_opcode_bytes(self, page: list[Head]) -> list[Head]:
+        """Fill ``raw`` (opcode bytes) for the code heads in ``page`` via one
+        bulk read over their extent (variable-length safe)."""
+        code = [h for h in page if h.kind == "code" and h.size > 0]
+        if not code:
+            return page
+        lo = code[0].ea
+        hi = code[-1].ea + code[-1].size
+        if hi - lo <= 0 or hi - lo > self._OP_SPAN_CAP:
+            return page
+        data = self._prog.read_bytes(lo, hi - lo)
+        biggest = self._max_raw
+        out = []
+        for h in page:
+            if h.kind == "code" and h.size > 0:
+                off = h.ea - lo
+                b = bytes(data[off:off + h.size])
+                biggest = max(biggest, len(b))
+                out.append(replace(h, raw=b))
+            else:
+                out.append(h)
+        with self._lock:
+            self._max_raw = biggest
+        return out
+
+    def max_raw_len(self) -> int:
+        with self._lock:
+            return self._max_raw
+
     def load_next_page(self) -> int:
         """Load one more page of heads; returns how many were added."""
         return self._load_next_page()
@@ -559,13 +594,16 @@ class ListingModel:
         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 {}
+        page = []
+        for r in rows:
+            try:
+                page.append(Head.from_raw(r))
+            except (KeyError, ValueError, TypeError):
+                continue
+        page = self._attach_opcode_bytes(page)
         with self._lock:
             base = len(self._heads)
-            for i, r in enumerate(rows):
-                try:
-                    h = Head.from_raw(r)
-                except (KeyError, ValueError, TypeError):
-                    continue
+            for i, h in enumerate(page):
                 self._by_ea.setdefault(h.ea, base + i)
                 self._heads.append(h)
             nxt = cur.get("next")
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py
index b921c54..b441a2c 100644
--- a/tests/test_scenarios.py
+++ b/tests/test_scenarios.py
@@ -117,7 +117,10 @@ class Ctx:
 
     # -- discovery -------------------------------------------------------- #
     def all_funcs(self):
-        return self.prog.functions().all_loaded()
+        idx = self.prog.functions()
+        if len(idx) == 0 and not idx.complete:
+            idx.load_all()  # a prior bump_items() cleared the index cache
+        return idx.all_loaded()
 
     def find_func(self, pred, limit=400):
         for f in self.all_funcs()[:limit]:
@@ -1629,6 +1632,13 @@ async def s_continuous_view(c: Ctx):
     kinds = {c.lst.model.get(i).kind for i in range(len(c.lst.model))}
     c.check("continuous listing interleaves code with data/undefined",
             "code" in kinds and ("data" in kinds or "unknown" in kinds), str(kinds))
+    # rendering parity with disasm: code lines carry opcode bytes
+    cidx = next((i for i in range(len(c.lst.model))
+                 if c.lst.model.get(i).kind == "code"), None)
+    c.check("continuous listing renders opcode bytes (parity with disasm)",
+            cidx is not None and c.lst.model.get(cidx).raw
+            and c.lst._op_field(c.lst.model.get(cidx)).strip() != "",
+            f"raw={c.lst.model.get(cidx).raw if cidx is not None else None!r}")
 
 
 # --------------------------------------------------------------------------- #