app: lift trace navigation out of IdaTui

First cut at the 4000-line class. Trace is the cleanest seam: 348 contiguous
lines, one coherent job (where we are in time and everything that moves us),
and two suites already covering it.

TraceController owns the state now -- the trace, the timestamp, the trail maps.
IdaTui keeps the keys, because Textual only merges BINDINGS from DOMNode
subclasses and a mixin's would be silently dropped, and it keeps the @work entry
points, because the worker machinery wants a DOMNode host. Both are one-line
delegates.

_trace/_t/_trail_map/_trail_map_ea/_trail_line_of stay readable on the app as
properties: the pilot suite and rpc.py read the position by those names, and a
property means one owner rather than a copy that can drift. rpc.py itself now
goes through the controller.

The parallel line-map that _apply_split_map used to poke into five attributes is
now one adopt_map() call -- same single shared index, but the sharing is stated
rather than implied by two places assigning the same fields.

731 checks, unchanged.

1 files changed, 416 insertions, 0 deletions

diff --git a/idatui/trace_ctl.py b/idatui/trace_ctl.py
new file mode 100644
index 0000000..5f17e49
--- /dev/null
+++ b/idatui/trace_ctl.py
@@ -0,0 +1,416 @@
+"""Execution-trace navigation: the controller behind `--trace FILE`.
+
+`idatui.trace` is the *reader* (parse the Tenet delta log, answer questions about
+it). This is the half that used to live in `IdaTui`: the trace's position in
+time, the trail painted onto the listing/pseudocode/hex, and every key that
+moves through it (`[` `]` `{` `}` `<` `>` `W`).
+
+Why it's a plain object and not a mixin: Textual only merges ``BINDINGS`` from
+``DOMNode`` subclasses, so a mixin's bindings are silently dropped. The keys
+therefore stay declared on ``IdaTui``, whose ``action_*`` methods are one-line
+delegates into here. Same reason the ``@work(thread=True)`` entry point stays on
+the app -- Textual's worker machinery wants a ``DOMNode`` host.
+
+The controller owns the trace state. ``IdaTui`` keeps forwarding properties
+(``app._trace``, ``app._t``, ``app._trail_map``...) because the pilot suite and
+the RPC layer read them by those names; see ``IdaTui._trace``.
+"""
+from __future__ import annotations
+
+import bisect
+import os
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:                                    # pragma: no cover
+    from .app import IdaTui
+
+_app_mod = None
+
+
+def _views():
+    """The widget classes, imported lazily.
+
+    ``app.py`` imports this module at the top, so importing it back at module
+    scope would be a cycle. Nothing here needs the classes until a trace is
+    actually driven, by which point ``app`` is fully imported.
+    """
+    global _app_mod
+    if _app_mod is None:
+        from . import app as _m
+        _app_mod = _m
+    return _app_mod
+
+
+class TraceController:
+    """Where we are in the trace, and everything that moves us."""
+
+    def __init__(self, app: "IdaTui", path: str = "") -> None:
+        self.app = app
+        self.path = path or ""       # the Tenet trace to explore, if any
+        self.trace = None            # the loaded Trace, once analysed
+        self.t = 0                   # current timestamp in that trace
+        self.trail_map = []          # decomp_map for trail_map_ea
+        self.trail_map_ea = None
+        self.trail_line_of: dict[int, int] = {}   # ea -> pseudocode line
+        self.trail_eas: list[int] = []            # sorted keys of trail_line_of
+        self.trail_span = None                    # ea span of that function
+        self.pending_line = None     # step waiting on a re-decompile
+
+    @property
+    def armed(self) -> bool:
+        """A trace was asked for but hasn't been parsed yet."""
+        return bool(self.path) and self.trace is None
+
+    def _need_trace(self) -> bool:
+        """Complain once, in one voice, rather than at four call sites."""
+        if self.trace is None:
+            self.app._status("no trace loaded (--trace FILE)")
+            return False
+        return True
+
+    # -- loading ------------------------------------------------------------ #
+    def load(self) -> None:
+        """Parse the trace and line it up with the database.
+
+        Runs after the function index exists: rebasing needs the database's
+        addresses, and without it nothing in the trace matches anything on
+        screen (our echo trace runs at 0x7ffff6faa000; the database has that
+        code at 0x2000).
+
+        Called on a worker thread, so every touch of the UI hops back.
+        """
+        from .trace import Trace
+        app = self.app
+        path = self.path
+        try:
+            def note(n):
+                app.call_from_thread(
+                    app._status, f"trace: {n:,} instructions\u2026")
+            trace = Trace.load(path, progress=note)
+        except OSError as e:
+            app.call_from_thread(app._status, f"trace: {e}")
+            return
+        if not trace.length:
+            app.call_from_thread(
+                app._status, f"trace: {os.path.basename(path)} is empty")
+            return
+        idx = app._func_index
+        addrs = [f.addr for f in idx.all_loaded()] if idx is not None else []
+        slide = trace.rebase(addrs)
+        trace.apply_slide(slide)
+        hit = sum(1 for f in (idx.all_loaded() if idx else [])
+                  if trace.executions(f.addr))
+        app.call_from_thread(self.ready, trace, slide, hit)
+
+    def ready(self, trace, slide: int, hit: int) -> None:
+        app = self.app
+        self.trace = trace
+        self.t = 0
+        dock = app.query_one(_views().TraceDock)
+        dock.display = True
+        dock.show(trace, 0)
+        where = (f"rebased {slide:+#x}" if slide else "no rebase needed")
+        app._status(f"trace: {trace.length:,} instructions, {hit} functions "
+                    f"touched ({where})", priority=True)
+        self.seek(0, follow=True)
+
+    # -- trace navigation --------------------------------------------------- #
+    def seek(self, idx: int, follow: bool = True) -> None:
+        """Move to timestamp ``idx``; ``follow`` takes the code view with it."""
+        app = self.app
+        t = self.trace
+        if t is None or not t.length:
+            return
+        # A seek invalidates any navigation still in flight. They run in workers
+        # and finish out of order: the trace's opening seek lands on the entry
+        # point, takes a while, and used to arrive AFTER later seeks — dragging
+        # the cursor back to _start while the trace was elsewhere, permanently.
+        #
+        # Bumped HERE and not in _goto_ea. Doing it for every navigation is the
+        # more general rule ("the last thing you asked for wins") but it also
+        # lets an ordinary follow be dropped by whatever navigates next, and the
+        # only evidence I have is about seeks. Narrow fix for the measured bug.
+        app._nav_seq += 1
+        self.t = max(0, min(int(idx), t.length - 1))
+        app.query_one(_views().TraceDock).show(t, self.t)
+        self.paint_trail()
+        if not follow:
+            return
+        pc = t.ip(self.t)
+        if app._split and self.seek_split(pc):
+            return
+        # Stay in whichever view you're reading. Without prefer_decomp a step
+        # from the pseudocode navigates to an address, which opens the listing —
+        # so stepping through C threw you out of C on the first keypress.
+        app._goto_ea(pc, push=False,
+                     prefer_decomp=(app._active == "decomp"))
+
+    def seek_split(self, pc: int) -> bool:
+        """Put BOTH panes on ``pc``. True if handled.
+
+        Normal navigation moves one pane and gives the companion a band, never a
+        cursor — that rule exists so the two can't chase each other. A trace step
+        isn't navigation though: time is a single global position, and both views
+        are showing the same instant, so both cursors belong on it.
+
+        The scroll anchoring is unchanged: after placing the cursors, the usual
+        _sync_split still bands the companion and aligns it to the driver's
+        screen row, so the eye tracks straight across.
+        """
+        app = self.app
+        M = _views()
+        lst = app.query_one(M.ListingView)
+        if lst.model is None:
+            return False
+        row = lst.model.ensure_ea(pc)
+        if row is None or row < 0:
+            return False        # not in this listing (other segment): full nav
+        lst.cursor = row
+        lst._scroll_cursor_into_view()
+
+        # Has execution actually left the decompiled function? Ask the map the
+        # trail painting keeps, which is keyed to what the decompiler currently
+        # HOLDS. _split_range comes from the guarded async path and lags, so a
+        # stale one made every step look like a function change: the decompiler
+        # bounced main -> PLT stub -> main, each bounce costing a synchronous
+        # 769-line map fetch on the UI thread.
+        span = self.trail_span
+        inside = (pc in self.trail_line_of
+                  or (span is not None and span[0] <= pc <= span[1]))
+        if not inside:
+            self.pending_line = pc
+            app._resync_decomp_async(pc)
+            return True
+        self.place_decomp_at(pc)
+        app._sync_split(app._active)
+        return True
+
+    def place_decomp_at(self, pc: int) -> None:
+        """Move the pseudocode cursor to the line covering ``pc``.
+
+        Uses the map the trail painting already keeps (keyed to the decompiler's
+        CURRENTLY loaded function), not the split view's _split_ea2line. That one
+        is refreshed by a guarded async path — it drops a result if _cur moved
+        while it was in flight — and a burst of steps moves _cur constantly, so
+        during stepping it is frequently a map of the function you just left.
+        """
+        app = self.app
+        dec = app.query_one(_views().DecompView)
+        line = None
+        if self.trail_map_ea == dec.loaded_ea and self.trail_line_of:
+            # EXACT match only. The decompiler doesn't attribute every
+            # instruction to a line (about half of main's aren't), and the
+            # tempting fallback — the nearest mapped instruction at or before
+            # the pc — is unsound: C lines are not monotonic in address, so
+            # 0x24a8 early in main resolved to line 708, "sub_2040();", near the
+            # end. A cursor that jumps to an unrelated statement is worse than
+            # one that waits; the trail still marks where we are.
+            line = self.trail_line_of.get(pc)
+        if line is None:
+            line = app._split_ea2line.get(pc)
+        if line is None:
+            line = dec.line_for_ea(pc)
+        if line is not None:
+            dec.goto(line, dec.cursor_x)
+
+    # -- painting ----------------------------------------------------------- #
+    def paint_trail(self) -> None:
+        """Push the execution trail into the code views.
+
+        Recomputed per seek rather than per repaint: it's ~200 lookups, and a
+        repaint happens far more often than a step.
+        """
+        app = self.app
+        M = _views()
+        t = self.trace
+        if t is None:
+            return
+        hx = app._try_view(M.HexView)     # None until it's mounted
+        if hx is not None:
+            hx.trace, hx.trace_idx = t, self.t
+            if hx.display:
+                hx.refresh()
+        trail = t.trail(self.t)
+        lst = app._try_view(M.ListingView)
+        if lst is not None:
+            lst.trail = trail
+            lst.refresh()
+        self.paint_trail_decomp(trail)
+
+    def paint_trail_decomp(self, trail: dict) -> None:
+        """Map the instruction trail onto pseudocode lines.
+
+        This is the thing Tenet can't do: it paints disassembly, because that's
+        where a trace's addresses live. We already have decomp_map (built for
+        the split view) saying which instructions each pseudocode line covers,
+        so the same trail lands on C.
+
+        A line covers many instructions, so it takes the strongest kind present:
+        'now' wins over 'past' wins over 'future' — if the instruction you are
+        standing on is part of this line, this line is where you are.
+        """
+        app = self.app
+        dec = app._try_view(_views().DecompView)
+        if dec is None:
+            return
+        ea = dec.loaded_ea
+        if not dec.display or ea is None or app.program is None:
+            dec.trail = {}
+            return
+        if self.trail_map_ea != ea:
+            # One index, built once per decompiled function and shared with the
+            # split view (_apply_split_map fills the same fields). decomp_map is
+            # an RPC and stepping is interactive, so paying it per keystroke —
+            # or twice, once for each of two parallel maps — would be felt.
+            try:
+                app._apply_split_map(ea, app.program.decomp_map(ea))
+            except Exception:  # noqa: BLE001
+                self.trail_map, self.trail_map_ea = [], ea
+                self.trail_line_of, self.trail_eas = {}, []
+                self.trail_span = None
+        rank = {"future": 0, "past": 1, "now": 2}
+        lines: dict[int, str] = {}
+        for i, eas in enumerate(self.trail_map or []):
+            best = None
+            for a in eas:
+                k = trail.get(a)
+                if k is not None and (best is None or rank[k] > rank[best]):
+                    best = k
+            if best is not None:
+                lines[i] = best
+        dec.trail = lines
+        dec.refresh()
+        pend, self.pending_line = self.pending_line, None
+        if pend is not None and app._split:
+            # The function was still decompiling when the step happened; land
+            # now that its line map exists.
+            self.place_decomp_at(pend)
+            app._sync_split(app._active)
+
+    def adopt_map(self, ea: int, m: list, ea2line: dict, span) -> None:
+        """Take the line map the split view just built.
+
+        ONE index, shared with the split view: the trace path used to keep a
+        parallel copy of exactly this, fetched separately and keyed differently,
+        which is how the two ended up describing different functions.
+        """
+        self.trail_map, self.trail_map_ea = m, ea
+        self.trail_line_of = dict(ea2line)
+        self.trail_eas = sorted(self.trail_line_of)
+        self.trail_span = span
+
+    # -- stepping ----------------------------------------------------------- #
+    def step(self, delta: int) -> None:
+        if not self._need_trace():
+            return
+        self.seek(self.t + delta)
+
+    def step_over(self, direction: int) -> None:
+        """Step over a call by following the stack pointer.
+
+        A call pushes, so the callee runs with SP BELOW where we started;
+        stepping until SP comes back up lands after the call returns. Cheaper
+        and more robust than recognising call instructions per architecture,
+        which is what the mode makes it: if this instruction doesn't call
+        anything, SP is already >= the start and it degenerates to one step.
+        """
+        if not self._need_trace():
+            return
+        t = self.trace
+        sp_name = "rsp" if "rsp" in t.reg_at else ("esp" if "esp" in t.reg_at else "sp")
+        sp0 = t.register(sp_name, self.t)
+        i = self.t + direction
+        limit = 200000          # a runaway search must not hang the UI
+        while 0 <= i < t.length and limit > 0:
+            sp = t.register(sp_name, i)
+            if sp0 is None or sp is None or sp >= sp0:
+                break
+            i += direction
+            limit -= 1
+        self.seek(max(0, min(i, t.length - 1)))
+
+    def seek_hit(self, direction: int) -> None:
+        """Seek to the next/previous time the focused view's subject was touched.
+
+        Two different questions with one pair of keys, because the answer to
+        "which thing?" is already on screen: in a code view it's the instruction
+        under the cursor ("when else did this run?"), in hex it's the byte under
+        the cursor ("who else touched this?").
+        """
+        app = self.app
+        M = _views()
+        if not self._need_trace():
+            return
+        t = self.trace
+        if app._active == "hex":
+            hx = app._try_view(M.HexView)
+            va = hx.cursor_va() if hx is not None else None
+            if va is None:
+                return
+            stamps = t.memory_accesses(va, 1)
+            what = f"access to {va:#x}"
+        else:
+            view = app._active_code_view()
+            if isinstance(view, M.DecompView):
+                # A C line is not one address, so ask about the whole statement:
+                # "when else did this line run?" is the question, and it's the
+                # union of its instructions' executions. Falling back to the
+                # line's single /*ea*/ marker would answer a narrower question
+                # and often no question at all, since most lines have no marker.
+                line = view.cursor
+                eas = []
+                if (self.trail_map_ea == view.loaded_ea
+                        and 0 <= line < len(self.trail_map or [])):
+                    eas = list(self.trail_map[line])
+                if not eas:
+                    one = view._line_ea(line)
+                    eas = [one] if one is not None else []
+                if not eas:
+                    app._status("this line has no instructions to seek on",
+                                priority=True)
+                    return
+                stamps = sorted({x for e in eas for x in t.executions(e)})
+                what = f"execution of C line {line + 1}"
+            else:
+                ea = view._cursor_ea() if view is not None else None
+                if ea is None:
+                    app._status("no address on this line", priority=True)
+                    return
+                stamps = list(t.executions(ea))
+                what = f"execution of {ea:#x}"
+        if not stamps:
+            app._status(f"no {what} in this trace", priority=True)
+            return
+        if direction > 0:
+            i = bisect.bisect_right(stamps, self.t)
+        else:
+            i = bisect.bisect_left(stamps, self.t) - 1
+        if not (0 <= i < len(stamps)):
+            edge = "last" if direction > 0 else "first"
+            app._status(f"already at the {edge} {what} "
+                        f"({len(stamps)} in the trace)", priority=True)
+            return
+        self.seek(stamps[i])
+        app._status(f"{what}: {i + 1} of {len(stamps)}  @ t={stamps[i]:,}",
+                    priority=True)
+
+    def seek_reg_write(self) -> None:
+        """W: which instruction set each register to its current value."""
+        app = self.app
+        if not self._need_trace():
+            return
+        t = self.trace
+        rows = []
+        for name in t.registers:
+            v = t.register(name, self.t)
+            if v is None:
+                continue
+            rows.append((name, v, t.last_write(name, self.t),
+                         t.next_write(name, self.t)))
+        if rows:
+            app.push_screen(_views().RegWriteScreen(rows, self.t),
+                            self._on_reg_write_chosen)
+
+    def _on_reg_write_chosen(self, idx) -> None:  # type: ignore[no-untyped-def]
+        if idx is not None:
+            self.seek(int(idx))