TX optimize sweeps live and asks to apply after the evidence is in

The six-prompt chain (path, samples, min, max, step, apply-up-front)
collapses to one: pick the path, and the live sweep screen takes it
from there. Levels land in an SNR bar chart as they are measured,
ascending by power with the current best starred, while the optimizer
announces its phase — coarse sweep, refining around the leader,
verifying — through a new on_phase callback. Reliability and trace
counts sit beside every bar, and a level nothing got through at reads
as an explicit no-reply row.

The apply decision moves to where it belongs: after the sweep, a
floating dialog offers the winner against the node's original power
(Cancel left, Apply right, Enter commits), and `a` re-offers it if
first declined. The sweep itself runs with apply off, so Esc at any
point stops the search and the optimizer's unwind restores the
original transmit power.

The sweep persists under one runs row (levels and traces alike), the
scripted CLI keeps its full flag set, chart, and --apply semantics,
and the tool moves in with Trace under Diagnostics.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

meshterm/services/tx_optimizer.py

@@ -32,6 +32,11 @@ from . import trace_runner
 
 LevelCallback = Callable[[int, int, TxLevelResult], None]
 PersistLevel = Callable[[TxLevelResult], None]
+PhaseCallback = Callable[[str], None]
+
+#: The search phases, in order, as reported to ``on_phase``: the coarse grid sweep, the
+#: integer fill-in around the coarse winner, and the extra-samples re-measure of the leader.
+PHASES = ("coarse", "refine", "verify")
 
 #: On a near-tie in target SNR (within this many dB of the best), the lower TX power wins.
 DEFAULT_SNR_TOLERANCE_DB = 1.0
@@ -142,6 +147,7 @@ async def optimize_tx_power(
     cooldown_s: float = 1.0,
     snr_tolerance: float = DEFAULT_SNR_TOLERANCE_DB,
     on_level: Optional[LevelCallback] = None,
+    on_phase: Optional[PhaseCallback] = None,
     persist_level: Optional[PersistLevel] = None,
     persist_trace: Optional[Callable] = None,
 ) -> TxOptResult:
@@ -168,6 +174,8 @@ async def optimize_tx_power(
         cooldown_s: Delay between individual traces (duty-cycle safety).
         snr_tolerance: dB band for the lower-power tie-break (see :func:`select_best`).
         on_level: Optional progress callback ``(completed, total, level_result)``.
+        on_phase: Optional callback announcing each search phase as it begins (one of
+            :data:`PHASES`), so a live view can say *what kind* of measuring is happening.
         persist_level: Optional callback to store each level's aggregated result.
         persist_trace: Optional callback to store each individual trace.
 
@@ -221,12 +229,16 @@ async def optimize_tx_power(
         return level
 
     try:
+        if on_phase is not None:
+            on_phase("coarse")
         for tx in coarse:
             await measure_level(tx)
 
         best = select_best(list(levels.values()), snr_tolerance=snr_tolerance)
 
         if refine:
+            if on_phase is not None:
+                on_phase("refine")
             lo = max(tx_min, best.tx_power - coarse_step + 1)
             hi = min(tx_max, best.tx_power + coarse_step - 1)
             for tx in range(lo, hi + 1):
@@ -237,6 +249,8 @@ async def optimize_tx_power(
         if verify:
             # Re-measure the leader with extra samples; if it was an outlier the larger
             # sample will pull it back and a steadier neighbor can take over.
+            if on_phase is not None:
+                on_phase("verify")
             await measure_level(best.tx_power)
             best = select_best(list(levels.values()), snr_tolerance=snr_tolerance)
 

meshterm/tools/tx_optimize.py

@@ -3,8 +3,13 @@
 You force a path (just like ``trace``) ending at the **target** node where SNR is
 measured. The node one hop *before* the target — one you hold admin rights on — is the
 node whose transmit power gets swept and tuned. Admin passwords are remembered between
-runs, progress streams live like a trace, and the winning power is written back to the
-node when the sweep finishes.
+runs.
+
+In the interactive menu one prompt picks the path and the live sweep screen
+(:mod:`meshterm.ui.tx_screen`) takes it from there: levels land in a bar chart as they
+are measured, and the apply decision is made *after* the sweep, over the evidence. On the
+CLI it stays a scriptable one-shot with the full flag set (range, step, samples,
+``--apply``, the HTML chart), and progress streams like a trace.
 """
 
 from __future__ import annotations
@@ -37,65 +42,121 @@ class TxOptimizeTool(Tool):
     order = 15  # right after Trace, its measurement sibling
 
     async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]:
-        """Interactively gather the path, sampling, range, and apply choices.
+        """Pick the path for the live sweep (everything else defaults, tuned in-screen).
 
         Args:
             ctx: Shared application context.
 
         Returns:
-            A parameter dict, or ``None`` if the user cancelled.
+            ``{"live": True, "path": spec}``, or ``None`` if the user cancelled.
         """
         device = await ctx.device()
         contacts = await device.get_contacts()
         choices = [c.name for c in contacts]
 
-        path_prompt = (
-            "Path to the target (comma-separated contacts/hex, ending at the target;\n"
-            "the node just before the target is the one we'll tune):"
+        prompt = (
+            "Comma-separated contacts/hex, ending at the target — "
+            "the node just before it is tuned"
         )
         validate_path = lambda v: _validate_link_path(v, contacts)  # noqa: E731
         if choices:
-            path_spec = await ctx.ui.autocomplete(path_prompt, choices, validate=validate_path)
+            path_spec = await ctx.ui.autocomplete(
+                "Path to the target", choices, prompt=prompt, validate=validate_path
+            )
         else:
-            path_spec = await ctx.ui.text(path_prompt, validate=validate_path)
+            path_spec = await ctx.ui.text(
+                "Path to the target", prompt=prompt, validate=validate_path
+            )
         if not path_spec:
             return None
+        return {"live": True, "path": path_spec.strip()}
 
-        samples = await ctx.ui.text(
-            f"Traces per TX level? (1-{MAX_SAMPLES})", default="3", validate=_is_valid_sample_count
-        )
-        if samples is None:
-            return None
-        tx_min = await ctx.ui.text(
-            "Lowest TX power to try:", default=str(ctx.settings.tx_opt_min), validate=_is_int
-        )
-        if tx_min is None:
-            return None
-        tx_max = await ctx.ui.text(
-            "Highest TX power to try:", default=str(ctx.settings.tx_opt_max), validate=_is_int
-        )
-        if tx_max is None:
-            return None
-        step = await ctx.ui.text("Coarse step:", default="3", validate=_is_valid_sample_count)
-        if step is None:
-            return None
-        apply = await ctx.ui.confirm(
-            "Set the winning TX power on the node afterward?", default=True
-        )
-        if apply is None:
-            return None
+    async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Run — without the outer run row on the live path.
 
-        return {
-            "path": path_spec.strip(),
-            "samples": int(samples),
-            "tx_min": int(tx_min),
-            "tx_max": int(tx_max),
-            "step": int(step),
-            "apply": bool(apply),
-            "viz": True,
-        }
+        The live screen opens its own ``runs`` row for the sweep (matching what a
+        scripted invocation records), so wrapping the screen session in another row
+        would double-log it. Scripted runs keep the base class's logging.
+
+        Args:
+            ctx: Shared application context.
+            params: Parameters for this invocation.
+
+        Returns:
+            The :class:`ToolResult` from :meth:`run`.
+        """
+        if params.get("live"):
+            return await self.run(ctx, params)
+        return await super().execute(ctx, params)
 
     async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Open the live sweep (menu) or run the scripted optimization (CLI).
+
+        Args:
+            ctx: Shared application context.
+            params: ``live`` + ``path`` from the menu prompt; or ``path``, ``samples``,
+                ``tx_min``, ``tx_max``, ``step``, ``apply``, optional ``password``/
+                ``viz``, and the injected ``_run_id`` from the CLI.
+
+        Returns:
+            A :class:`ToolResult` with the optimum and any chart path.
+        """
+        if params.get("live"):
+            return await self._run_live(ctx, str(params["path"]))
+        return await self._run_cli(ctx, params)
+
+    # -- interactive (menu) ---------------------------------------------------------
+
+    async def _run_live(self, ctx: AppContext, path_spec: str) -> ToolResult:
+        """Resolve the link, log in, and hand off to the live sweep screen.
+
+        Args:
+            ctx: Shared application context.
+            path_spec: The user's comma-separated path entry (names and/or hex).
+
+        Returns:
+            A :class:`ToolResult` echoing the sweep's recorded summary.
+        """
+        from ..ui.tx_screen import open_tx_optimize
+
+        device = await ctx.device()
+        contacts = await device.get_contacts()
+        path = trace_runner.parse_trace_path(path_spec, contacts)
+        admin_node, target_label = _resolve_link(path, contacts)
+        await self._login(ctx, admin_node, {})
+
+        summary = await open_tx_optimize(
+            ctx, admin_node=admin_node, target_label=target_label, path=path
+        )
+        return ToolResult(summary=summary)
+
+    async def _login(
+        self, ctx: AppContext, admin_node: Contact, params: dict[str, Any]
+    ) -> None:
+        """Authenticate against the admin node, remembering a working password.
+
+        Args:
+            ctx: Shared application context.
+            admin_node: The node we're about to tune.
+            params: Tool params (may carry an explicit ``password`` on the CLI).
+
+        Raises:
+            DeviceCommandError: If the login is rejected (the stored password, now
+                known bad, is forgotten so the next run asks fresh).
+        """
+        device = await ctx.device()
+        password = await self._resolve_password(ctx, admin_node, params)
+        if not await device.admin_login(admin_node, password):
+            ctx.admin_store.forget(admin_node)  # bad password: don't keep reusing it
+            raise DeviceCommandError(
+                f"admin login to {admin_node.name!r} failed (wrong password?). "
+                "The saved password was cleared; re-run to enter a new one."
+            )
+        ctx.admin_store.remember(admin_node, password)
+
+    # -- scripted (CLI) ---------------------------------------------------------------
+
+    async def _run_cli(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
         """Resolve the link, log in, sweep TX power, render results, and apply the winner.
 
         Args:
@@ -119,15 +180,8 @@ class TxOptimizeTool(Tool):
         if int(params.get("samples", 3)) > MAX_SAMPLES:
             ctx.ui.note(f"[warn]capping at {MAX_SAMPLES} traces per level[/warn]")
 
-        password = await self._resolve_password(ctx, admin_node, params)
         ctx.ui.note(f"[muted]logging in to[/muted] [brand]{admin_node.name}[/brand] [muted]…[/muted]")
-        if not await device.admin_login(admin_node, password):
-            ctx.admin_store.forget(admin_node)  # bad password: don't keep reusing it
-            raise DeviceCommandError(
-                f"admin login to {admin_node.name!r} failed (wrong password?). "
-                "The saved password was cleared; re-run to enter a new one."
-            )
-        ctx.admin_store.remember(admin_node, password)
+        await self._login(ctx, admin_node, params)
 
         ctx.ui.note(
             f"[muted]tuning[/muted] [brand]{admin_node.name}[/brand] "
@@ -349,42 +403,6 @@ def _validate_link_path(value: str, contacts: list[Contact]) -> bool | str:
     return True
 
 
-def _is_valid_sample_count(value: str) -> bool | str:
-    """Validate a positive count in ``1..MAX_SAMPLES`` for a questionary text answer.
-
-    Args:
-        value: Raw input.
-
-    Returns:
-        ``True`` if valid, otherwise an error message string.
-    """
-    try:
-        count = int(value)
-    except ValueError:
-        return "Enter a whole number."
-    if count < 1:
-        return "Enter a number greater than zero."
-    if count > MAX_SAMPLES:
-        return f"Maximum {MAX_SAMPLES}."
-    return True
-
-
-def _is_int(value: str) -> bool | str:
-    """Validate that a questionary answer is a whole number.
-
-    Args:
-        value: Raw input.
-
-    Returns:
-        ``True`` if valid, otherwise an error message string.
-    """
-    try:
-        int(value)
-        return True
-    except ValueError:
-        return "Enter a whole number."
-
-
 def _fmt_snr(snr: Optional[float]) -> str:
     """Format an optional SNR for the progress description.
 

meshterm/ui/tx_screen.py

@@ -0,0 +1,463 @@
+"""The live TX-optimization screen: watch a remote node's power sweep land level by level.
+
+This is the interactive face of the ``tx-optimize`` tool (the scripted CLI keeps its
+one-shot table and HTML chart). Where the old flow front-loaded six prompts — including
+"apply the winner?" before anything had been measured — this screen follows the app's
+living-screen pattern: one prompt picks the path, the sweep starts immediately, and every
+measured level lands in a bar chart as it completes, phase by phase (coarse → refine →
+verify), with the current best starred. The apply decision moves to *after* the evidence
+is in: when the sweep finishes, a floating dialog offers the winner (Cancel left, Apply
+right, Enter commits — the platform-dialog convention), and ``a`` re-offers it later if
+the user first backs out of the dialog.
+
+Backing out mid-sweep is safe: the optimizer's unwind restores the admin node's original
+transmit power, and every level already measured is persisted under the sweep's ``runs``
+row exactly as a scripted run records it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING, Any, Callable, Optional
+
+from rich.console import Group, RenderableType
+from rich.table import Table
+from rich.text import Text
+
+from ..core.models import Contact, TxLevelResult, TxOptResult
+from ..services import tx_optimizer
+from .theme import snr_style
+from .trace_screen import snr_bar
+from .tui.render import render_lines
+from .tui.screen import Screen
+from .tui.spinner import Spinner
+
+if TYPE_CHECKING:
+    from ..context import AppContext
+
+#: Seconds between spinner frames while the sweep is measuring.
+_SPINNER_INTERVAL = 0.12
+
+#: Human phrasing for each optimizer phase reported through ``on_phase``.
+_PHASE_LABELS = {
+    "coarse": "coarse sweep",
+    "refine": "refining around the leader",
+    "verify": "verifying the leader",
+}
+
+
+class TxSweepScreen(Screen):
+    """A full-screen live view of one remote-admin TX-power sweep.
+
+    The screen is fed by the optimizer's callbacks (each measured level, each phase
+    change) and by the controller marking completion/failure/apply; it renders state and
+    handles keys only. Esc backs out at any time — mid-sweep that cancels the search and
+    restores the node's original power; after completion, ``a`` (re-)offers the apply
+    dialog.
+    """
+
+    floating = False
+
+    def __init__(
+        self,
+        *,
+        admin_label: str,
+        target_label: str,
+        path: str,
+        tx_min: int,
+        tx_max: int,
+        step: int,
+        samples: int,
+        session: Any,
+        offer_apply: Callable[[], None],
+    ) -> None:
+        """Create the sweep screen (the controller starts the sweep alongside it).
+
+        Args:
+            admin_label: Display name of the node being tuned.
+            target_label: Display name of the node the SNR is measured at.
+            path: The forced one-way path (comma-separated hashes) out to the target.
+            tx_min: Lowest TX power the sweep explores.
+            tx_max: Highest TX power the sweep explores.
+            step: Coarse-sweep grid spacing.
+            samples: Traces measured per level.
+            session: The running TUI session (for repaints).
+            offer_apply: Re-opens the apply dialog (bound by the controller); wired to
+                the ``a`` key once a completed sweep has a winner that wasn't applied.
+        """
+        super().__init__()
+        self.title = f"tx optimize · {admin_label} → {target_label}"
+        self._admin_label = admin_label
+        self._target_label = target_label
+        self._path = path
+        self._tx_min = tx_min
+        self._tx_max = tx_max
+        self._step = step
+        self._samples = samples
+        self._session = session
+        self._offer_apply = offer_apply
+        self._spinner = Spinner()
+        self._phase: Optional[str] = None
+        self._done = 0
+        self._total = 0
+        self._levels: dict[int, TxLevelResult] = {}
+        self._best_tx: Optional[int] = None
+        self._result: Optional[TxOptResult] = None
+        self._applied = False
+        self._error: Optional[str] = None
+
+    # --- controller feed --------------------------------------------------------
+
+    @property
+    def running(self) -> bool:
+        """Whether the sweep is still measuring."""
+        return self._result is None and self._error is None
+
+    @property
+    def footer_hint(self) -> str:  # type: ignore[override]
+        """The footer keys, tracking sweep state."""
+        if self.running:
+            return "measuring… · ↑↓ scroll · Esc stop & restore"
+        if self._can_apply():
+            return "a apply winner · ↑↓ scroll · Esc back"
+        return "↑↓ scroll · Esc back"
+
+    def tick(self) -> None:
+        """Advance the measuring spinner (driven by the controller's animation timer)."""
+        self._spinner.tick()
+
+    def on_phase(self, phase: str) -> None:
+        """Record the optimizer entering a search phase (see ``tx_optimizer.PHASES``)."""
+        self._phase = phase
+        self._session.invalidate()
+
+    def on_level(self, done: int, total: int, level: TxLevelResult) -> None:
+        """Record one measured level (replacing any earlier pass at the same power)."""
+        self._done, self._total = done, total
+        self._levels[level.tx_power] = level
+        self._best_tx = tx_optimizer.select_best(list(self._levels.values())).tx_power
+        self._session.invalidate()
+
+    def complete(self, result: TxOptResult) -> None:
+        """Mark the sweep finished and adopt its final selection."""
+        self._result = result
+        self._best_tx = result.best_tx
+        self._session.invalidate()
+
+    def fail(self, error: str) -> None:
+        """Mark the sweep failed, keeping whatever levels already landed on screen."""
+        self._error = error
+        self._session.invalidate()
+
+    def mark_applied(self) -> None:
+        """Record that the winner was written to the admin node."""
+        self._applied = True
+        self._session.invalidate()
+
+    def _can_apply(self) -> bool:
+        """Whether ``a`` should offer the winner (finished, got a result, not yet set)."""
+        return (
+            self._result is not None
+            and self._result.best_snr is not None
+            and not self._applied
+        )
+
+    # --- input --------------------------------------------------------------------
+
+    def handle(self, action: str, data: str = "") -> None:
+        """Offer the apply dialog, scroll, or dismiss."""
+        if action == "text" and data.lower() == "a":
+            if self._can_apply():
+                self._offer_apply()
+        elif action == "up":
+            self.scroll_lines(-1)
+        elif action == "down":
+            self.scroll_lines(1)
+        elif action == "pageup":
+            self.scroll_pages(-1)
+        elif action in ("pagedown", "space"):
+            self.scroll_pages(1)
+        elif action in ("home", "ctrl_home"):
+            self.scroll_to_top()
+        elif action in ("end", "ctrl_end"):
+            self.scroll_to_bottom()
+        elif action == "escape":
+            self.resolve(None)
+
+    # --- rendering ------------------------------------------------------------------
+
+    def render_body(self, width: int) -> list[str]:
+        """Render the link header, phase line, level chart, and outcome."""
+        lines = render_lines(Group(*self._sections()), width)
+        self._scroll_total = max(1, len(lines))
+        return lines
+
+    def _sections(self) -> list[RenderableType]:
+        """Assemble the screen's stacked sections for the current state."""
+        sections: list[RenderableType] = [self._header(), Text(), self._phase_line()]
+        if self._levels:
+            sections += [Text(), self._levels_table()]
+        outcome = self._outcome()
+        if outcome is not None:
+            sections += [Text(), outcome]
+        return sections
+
+    def _header(self) -> Text:
+        """The tuned link and the sweep's fixed parameters, label-aligned."""
+        header = Text.assemble(
+            ("tuning     ", "muted"), (self._admin_label, "brand"),
+            (" → ", "muted"), (self._target_label, "brand"), ("\n", ""),
+            ("via        ", "muted"), (self._path, ""), ("\n", ""),
+            ("sweep      ", "muted"),
+            (f"TX {self._tx_min}–{self._tx_max} · step {self._step} · "
+             f"{self._samples} trace{'s' if self._samples != 1 else ''}/level", ""),
+        )
+        if self._result is not None and self._result.original_tx is not None:
+            header.append("\n")
+            header.append("was        ", style="muted")
+            header.append(f"TX {self._result.original_tx}")
+        return header
+
+    def _phase_line(self) -> Text:
+        """The live phase/progress line, or the terminal state once the sweep ends."""
+        if self._error is not None:
+            return Text(f"✗ sweep failed: {self._error}", style="err")
+        if self._result is not None:
+            return Text("✓ sweep complete", style="ok")
+        line = self._spinner.text()
+        label = _PHASE_LABELS.get(self._phase or "", "starting…")
+        line.append(f"  {label}", style="muted")
+        if self._total:
+            line.append(f" · level {self._done}/{self._total}", style="muted")
+        return line
+
+    def _levels_table(self) -> Table:
+        """Every measured level as a bar chart, ascending by TX, the best starred."""
+        table = Table(box=None, padding=(0, 1, 0, 0), expand=False, header_style="muted")
+        table.add_column("TX", justify="right", min_width=4)
+        table.add_column("SNR AT TARGET")
+        table.add_column("", justify="right")  # numeric SNR
+        table.add_column("RELIABILITY", justify="right")
+        table.add_column("TRACES", justify="right", style="muted")
+        for level in sorted(self._levels.values(), key=lambda lv: lv.tx_power):
+            is_best = level.tx_power == self._best_tx
+            tx_cell = Text()
+            tx_cell.append("★ " if is_best else "  ", style="ok")
+            tx_cell.append(str(level.tx_power), style="brand" if is_best else "")
+            snr = level.target_snr
+            if snr is None:
+                bar_cell: Text = Text("✗ no reply", style="err")
+                snr_cell = Text("—", style="muted")
+            else:
+                bar_cell = snr_bar(snr)
+                snr_cell = Text(f"{snr:+.1f} dB", style=snr_style(snr))
+            rate = level.success_rate
+            rate_style = "ok" if rate >= 1.0 else ("warn" if rate > 0 else "err")
+            table.add_row(
+                tx_cell,
+                bar_cell,
+                snr_cell,
+                Text(f"{rate:.0%}", style=rate_style),
+                f"{level.successes}/{level.samples}",
+            )
+        return table
+
+    def _outcome(self) -> Optional[Text]:
+        """The completed sweep's verdict and apply status, or ``None`` while measuring."""
+        result = self._result
+        if result is None:
+            return None
+        if result.best_snr is None:
+            return Text.assemble(
+                ("! ", "warn"),
+                ("no traces reached ", ""),
+                (self._target_label, "brand"),
+                (" at any level — check the path ends at the target and is reachable.", ""),
+            )
+        snr_text = Text(f"{result.best_snr:+.1f} dB", style=snr_style(result.best_snr))
+        outcome = Text.assemble(
+            ("best       ", "muted"), (f"TX {result.best_tx}", "brand"), ("  ·  ", "muted"),
+        )
+        outcome.append_text(snr_text)
+        outcome.append(f" at {self._target_label}  ·  ", style="muted")
+        outcome.append(f"{result.best_success_rate:.0%} reliable")
+        outcome.append("\n")
+        outcome.append("status     ", style="muted")
+        if self._applied:
+            outcome.append(f"✓ TX {result.best_tx} set on {self._admin_label}", style="ok")
+        else:
+            restored = (
+                f" (node left at TX {result.original_tx})"
+                if result.original_tx is not None
+                else ""
+            )
+            outcome.append(f"not applied{restored} — press a to apply", style="muted")
+        return outcome
+
+
+async def open_tx_optimize(
+    ctx: "AppContext",
+    *,
+    admin_node: Contact,
+    target_label: str,
+    path: str,
+) -> dict[str, Any]:
+    """Run one live TX-power sweep for an already-authenticated admin node.
+
+    The caller (the ``tx-optimize`` tool) has resolved the path and logged in to
+    ``admin_node``; this opens the sweep screen, drives the optimizer against it with
+    ``apply=False`` (the node is restored to its original power when the search ends or
+    is cancelled), persists every level and trace under one ``runs`` row, and raises the
+    apply dialog once the evidence is in. Cancelling mid-sweep (Esc) stops the search;
+    the optimizer's unwind restores the original power before the screen returns.
+
+    Args:
+        ctx: The shared application context (must be running the interactive TUI surface).
+        admin_node: The contact being tuned (already logged in).
+        target_label: Display name of the node the SNR is measured at.
+        path: The forced one-way path (comma-separated hashes) ending at the target.
+
+    Returns:
+        The summary dict recorded on the sweep's run row (empty if cancelled before any
+        level landed).
+
+    Raises:
+        RuntimeError: If called outside the interactive menu (no full-screen session).
+    """
+    from .surface import TuiUi
+
+    if not isinstance(ctx.ui, TuiUi):  # pragma: no cover - guarded by the menu-only caller
+        raise RuntimeError("the live TX sweep is only available in the menu")
+    session = ctx.ui.session
+    device = await ctx.device()
+
+    tx_min = ctx.settings.tx_opt_min
+    tx_max = ctx.settings.tx_opt_max
+    step = 3
+    samples = 3
+    summary: dict[str, Any] = {}
+
+    def offer_apply() -> None:
+        asyncio.ensure_future(apply_dialog())
+
+    screen = TxSweepScreen(
+        admin_label=admin_node.name,
+        target_label=target_label,
+        path=path,
+        tx_min=tx_min,
+        tx_max=tx_max,
+        step=step,
+        samples=samples,
+        session=session,
+        offer_apply=offer_apply,
+    )
+
+    run_id = ctx.repo.start_run(
+        "tx-optimize",
+        {"path": path, "samples": samples, "step": step, "tx_min": tx_min, "tx_max": tx_max},
+        ctx.profile_name,
+    )
+
+    async def apply_dialog() -> None:
+        """Offer the winner in a floating dialog; write it to the node on Apply."""
+        result = screen._result
+        if result is None or result.best_snr is None or screen._applied:
+            return
+        was = f" (currently {result.original_tx})" if result.original_tx is not None else ""
+        prompt = Text.assemble(
+            ("Set TX ", ""), (str(result.best_tx), "brand"),
+            (f" on {admin_node.name}?{was}", ""),
+        )
+        # Platform-dialog convention: the safe way out left, the committing action right
+        # and default, so Enter applies and Esc backs out.
+        choice = await session.button_dialog(
+            prompt,
+            [("Cancel", "cancel"), ("Apply", "apply")],
+            title="Apply winner",
+            default=1,
+            footer_hint="Esc cancel · Enter apply",
+        )
+        if choice != "apply":
+            return
+        await device.set_remote_tx_power(admin_node, result.best_tx)
+        ctx.log.info("set TX power %s on %s", result.best_tx, admin_node.name)
+        screen.mark_applied()
+        summary["applied"] = True
+        ctx.repo.finish_run(run_id, "ok", summary)
+
+    async def sweep() -> None:
+        """Drive the optimizer to completion, then raise the apply dialog."""
+        ticker = asyncio.ensure_future(_animate(session, screen))
+        try:
+            result = await tx_optimizer.optimize_tx_power(
+                device,
+                target_label,
+                admin_node,
+                path,
+                tx_min=tx_min,
+                tx_max=tx_max,
+                coarse_step=step,
+                samples_per_level=samples,
+                apply=False,  # the decision moves to the dialog, after the evidence is in
+                cooldown_s=ctx.settings.trace_cooldown_s,
+                on_level=screen.on_level,
+                on_phase=screen.on_phase,
+                persist_level=lambda lv: ctx.repo.record_tx_sample(run_id, lv),
+                persist_trace=lambda t: ctx.repo.record_trace(run_id, t),
+            )
+        except asyncio.CancelledError:
+            ctx.repo.finish_run(run_id, "error", {"error": "cancelled"})
+            raise
+        except Exception as exc:  # noqa: BLE001 - shown on the screen, not crashed through
+            screen.fail(str(exc))
+            ctx.repo.finish_run(run_id, "error", {"error": str(exc)})
+            return
+        finally:
+            ticker.cancel()
+            try:
+                await ticker
+            except asyncio.CancelledError:
+                pass
+            except Exception:  # noqa: BLE001 - a spinner hiccup must never break the sweep
+                pass
+        summary.update(
+            {
+                "target": result.target,
+                "admin_node": result.admin_node,
+                "path": result.path,
+                "best_tx": result.best_tx,
+                "best_snr": result.best_snr,
+                "best_success_rate": round(result.best_success_rate, 3),
+                "original_tx": result.original_tx,
+                "applied": False,
+                "levels_measured": len(result.levels),
+            }
+        )
+        ctx.repo.finish_run(run_id, "ok", summary)
+        screen.complete(result)
+        await apply_dialog()
+
+    worker = asyncio.ensure_future(sweep())
+    try:
+        await session.run_screen(screen)
+    finally:
+        if not worker.done():
+            # Esc mid-sweep: stop the search and let the optimizer's unwind restore the
+            # node's original power. The ring spinner covers that last device command.
+            worker.cancel()
+            async with ctx.ui.busy_overlay():
+                try:
+                    await worker
+                except asyncio.CancelledError:
+                    pass
+                except Exception:  # noqa: BLE001 - the sweep is over; nothing to surface
+                    pass
+    return summary
+
+
+async def _animate(session: Any, screen: TxSweepScreen) -> None:
+    """Advance the measuring spinner and repaint on a steady cadence, until cancelled."""
+    while True:
+        await asyncio.sleep(_SPINNER_INTERVAL)
+        screen.tick()
+        session.invalidate()

tests/test_live_screens.py

@@ -0,0 +1,313 @@
+"""Unit tests for the live Trace and TX-optimize screens.
+
+These drive the two full-screen tools' pure logic — burst/sweep state machines, key
+handling, and rendering — against fake sessions and injected runners, so they run fast
+and headless (the same approach as ``test_tui``).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from meshterm.core.models import Hop, TraceResult, TraceStats, TxLevelResult, TxOptResult, utcnow
+from meshterm.ui.trace_screen import TraceScreen, snr_bar
+from meshterm.ui.tx_screen import TxSweepScreen
+
+
+class _FakeSession:
+    """The one session capability the screens use directly: requesting a repaint."""
+
+    def __init__(self) -> None:
+        self.repaints = 0
+
+    def invalidate(self) -> None:
+        self.repaints += 1
+
+
+def _trace(*snrs: float, success: bool = True, target: str = "Alice") -> TraceResult:
+    """Build a trace with one hop per SNR reading."""
+    hops = [Hop(index=i, node=f"{i:02x}{i:02x}", snr=snr) for i, snr in enumerate(snrs)]
+    return TraceResult(
+        target=target, success=success, hops=hops, round_trip_ms=200.0, path_hash_bytes=2
+    )
+
+
+def _plain(lines: list[str]) -> str:
+    """Join rendered ANSI lines and strip nothing — assertions use substring checks."""
+    return "\n".join(lines)
+
+
+# --- snr_bar -------------------------------------------------------------------
+
+
+def test_snr_bar_scales_with_signal_quality() -> None:
+    """A stronger signal fills more of the track; None renders an empty muted track."""
+    weak = snr_bar(-12.0).plain
+    strong = snr_bar(8.0).plain
+    assert weak.count("▆") < strong.count("▆")
+    assert len(weak) == len(strong)  # the track width is constant
+    assert "▆" not in snr_bar(None).plain
+
+
+def test_snr_bar_clamps_out_of_range_readings() -> None:
+    """Readings beyond the display range clamp to the ends instead of over/underflowing."""
+    assert snr_bar(99.0).plain.count("▆") == snr_bar(10.0).plain.count("▆")
+    assert snr_bar(-99.0).plain.count("▆") == 1  # a heard hop always shows something
+
+
+# --- TraceScreen ----------------------------------------------------------------
+
+
+def _trace_screen(burst=None, edit_path=None, previous=None) -> tuple[TraceScreen, _FakeSession]:
+    session = _FakeSession()
+
+    async def default_burst(samples, path_spec, on_trace):  # noqa: ANN001
+        for _ in range(samples):
+            on_trace(_trace(5.0, 2.0))
+
+    async def default_edit(current):  # noqa: ANN001
+        return current
+
+    screen = TraceScreen(
+        "Alice",
+        device_label="Us",
+        device_hash="aabb" + "00" * 30,
+        resolve=lambda label: label,
+        session=session,
+        burst=burst or default_burst,
+        edit_path=edit_path or default_edit,
+        previous=previous,
+    )
+    return screen, session
+
+
+async def test_trace_screen_burst_streams_results_into_the_log() -> None:
+    """A burst appends each landed trace; the log and aggregates render them."""
+    screen, _ = _trace_screen()
+    screen.start_burst()
+    await asyncio.sleep(0)  # let the worker run
+    await screen._worker
+    body = _plain(screen.render_body(100))
+    assert "success rate" in body and "3/3" in body
+    assert "#3" in body  # newest-first numbering
+    assert "Per-hop medians" in body
+
+
+async def test_trace_screen_seeds_route_from_previous_trace() -> None:
+    """Before any fresh reply, the stored route shows, marked as previous."""
+    old = _trace(4.0)
+    old.timestamp = utcnow() - timedelta(hours=3)
+    screen, _ = _trace_screen(previous=old)
+    body = _plain(screen.render_body(100))
+    assert "(previous" in body
+    # A fresh success replaces the seeded route and drops the marker.
+    screen._on_trace(_trace(6.0))
+    body = _plain(screen.render_body(100))
+    assert "(previous" not in body
+
+
+async def test_trace_screen_only_one_burst_at_a_time() -> None:
+    """Enter during a burst is a no-op; the running flag gates re-entry."""
+    started = 0
+    release = asyncio.Event()
+
+    async def burst(samples, path_spec, on_trace):  # noqa: ANN001
+        nonlocal started
+        started += 1
+        await release.wait()
+
+    screen, _ = _trace_screen(burst=burst)
+    screen.start_burst()
+    assert screen._running
+    screen.handle("enter")  # ignored while running
+    release.set()
+    await screen._worker
+    assert started == 1
+    assert not screen._running
+
+
+async def test_trace_screen_burst_failure_reads_inline() -> None:
+    """A failed burst reports its error in the log area instead of crashing the screen."""
+
+    async def burst(samples, path_spec, on_trace):  # noqa: ANN001
+        raise RuntimeError("no route")
+
+    screen, _ = _trace_screen(burst=burst)
+    screen.start_burst()
+    await screen._worker
+    assert "trace failed: no route" in _plain(screen.render_body(100))
+
+
+async def test_trace_screen_samples_cycle_only_when_idle() -> None:
+    """`s` cycles the burst size through the odd sample counts, but never mid-burst."""
+    screen, _ = _trace_screen()
+    assert screen._samples == 3
+    screen.handle("text", "s")
+    assert screen._samples == 5
+    screen._running = True
+    screen.handle("text", "s")
+    assert screen._samples == 5  # locked while a burst is in flight
+
+
+async def test_trace_screen_path_dialog_updates_the_spec() -> None:
+    """`p` opens the injected path prompt; its result becomes the next burst's path."""
+    asked: list[str] = []
+
+    async def edit(current: str):  # noqa: ANN001
+        asked.append(current)
+        return "3d,f2"
+
+    screen, _ = _trace_screen(edit_path=edit)
+    screen.handle("text", "p")
+    await asyncio.sleep(0)
+    assert asked == [""]
+    assert screen._path_spec == "3d,f2"
+    assert "3d,f2" in _plain(screen.render_body(100))
+
+
+async def test_trace_screen_escape_cancels_the_inflight_burst() -> None:
+    """Esc resolves the screen and cancels a burst that is still measuring."""
+    release = asyncio.Event()
+
+    async def burst(samples, path_spec, on_trace):  # noqa: ANN001
+        await release.wait()
+
+    screen, _ = _trace_screen(burst=burst)
+    screen.start_burst()
+    screen.future = asyncio.get_running_loop().create_future()
+    screen.handle("escape")
+    assert screen.future.result() is None
+    with pytest.raises(asyncio.CancelledError):
+        await screen._worker
+
+
+# --- TxSweepScreen -----------------------------------------------------------------
+
+
+def _level(tx: int, snr: float | None, successes: int = 2, samples: int = 2) -> TxLevelResult:
+    return TxLevelResult(
+        tx_power=tx,
+        samples=samples,
+        successes=successes,
+        target_snr=snr,
+        score=snr if snr is not None else float("-inf"),
+        stats=TraceStats.from_traces("Alice", []),
+    )
+
+
+def _sweep_screen(offer_apply=None) -> tuple[TxSweepScreen, _FakeSession]:
+    session = _FakeSession()
+    screen = TxSweepScreen(
+        admin_label="Repeater",
+        target_label="Alice",
+        path="a1b2,d4e5",
+        tx_min=12,
+        tx_max=28,
+        step=3,
+        samples=3,
+        session=session,
+        offer_apply=offer_apply or (lambda: None),
+    )
+    return screen, session
+
+
+def _result(best_tx: int = 19, best_snr: float | None = 8.8) -> TxOptResult:
+    return TxOptResult(
+        target="Alice",
+        admin_node="Repeater",
+        path="a1b2,d4e5",
+        original_tx=20,
+        best_tx=best_tx,
+        best_snr=best_snr,
+        best_success_rate=1.0,
+        applied=False,
+        levels=[_level(best_tx, best_snr)],
+    )
+
+
+def test_sweep_screen_stars_the_running_best() -> None:
+    """Each landed level renders ascending by TX with the current best starred."""
+    screen, _ = _sweep_screen()
+    screen.on_phase("coarse")
+    screen.on_level(1, 7, _level(12, -2.0))
+    screen.on_level(2, 7, _level(18, 7.5))
+    body = _plain(screen.render_body(100))
+    assert "coarse sweep" in body and "level 2/7" in body
+    starred = next(line for line in body.splitlines() if "★" in line)
+    assert "18" in starred
+
+
+def test_sweep_screen_shows_failed_levels_distinctly() -> None:
+    """A level nothing got through at reads as a no-reply row, not an empty bar."""
+    screen, _ = _sweep_screen()
+    screen.on_level(1, 7, _level(28, None, successes=0))
+    assert "✗ no reply" in _plain(screen.render_body(100))
+
+
+def test_sweep_screen_completion_offers_the_winner() -> None:
+    """After completion the outcome shows and `a` re-offers the apply dialog."""
+    offered: list[bool] = []
+    screen, _ = _sweep_screen(offer_apply=lambda: offered.append(True))
+    screen.handle("text", "a")  # mid-sweep: nothing to offer yet
+    assert offered == []
+    screen.complete(_result())
+    body = _plain(screen.render_body(100))
+    assert "sweep complete" in body and "TX 19" in body and "press a to apply" in body
+    assert "a apply winner" in screen.footer_hint
+    screen.handle("text", "a")
+    assert offered == [True]
+
+
+def test_sweep_screen_apply_updates_status_and_retires_the_key() -> None:
+    """Marking the winner applied flips the status line and stops offering `a`."""
+    offered: list[bool] = []
+    screen, _ = _sweep_screen(offer_apply=lambda: offered.append(True))
+    screen.complete(_result())
+    screen.mark_applied()
+    assert "✓ TX 19 set on Repeater" in _plain(screen.render_body(100))
+    assert "a apply winner" not in screen.footer_hint
+    screen.handle("text", "a")
+    assert offered == []
+
+
+def test_sweep_screen_no_result_never_offers_apply() -> None:
+    """A sweep where nothing got through warns and keeps the apply key retired."""
+    screen, _ = _sweep_screen()
+    screen.complete(_result(best_snr=None))
+    body = _plain(screen.render_body(100))
+    assert "no traces reached" in body
+    assert "a apply winner" not in screen.footer_hint
+
+
+def test_sweep_screen_failure_keeps_measured_levels_on_screen() -> None:
+    """A mid-sweep error is reported while the levels already measured stay visible."""
+    screen, _ = _sweep_screen()
+    screen.on_level(1, 7, _level(12, -2.0))
+    screen.fail("link lost")
+    body = _plain(screen.render_body(100))
+    assert "sweep failed: link lost" in body
+    assert "-2.0" in body
+
+
+def test_sweep_screen_escape_resolves() -> None:
+    """Esc resolves the screen (the controller then cancels and restores)."""
+    screen, _ = _sweep_screen()
+
+    class _Fut:
+        def __init__(self) -> None:
+            self.value = None
+            self._done = False
+
+        def done(self) -> bool:
+            return self._done
+
+        def set_result(self, value) -> None:  # noqa: ANN001
+            self.value = value
+            self._done = True
+
+    screen.future = _Fut()
+    screen.handle("escape")
+    assert screen.future.done()

tests/test_skeleton.py

@@ -342,6 +342,29 @@ async def test_tx_optimizer_no_apply_restores_original(tmp_path: Path) -> None:
     assert await device.get_remote_tx_power(admin) == 15  # restored, not the winner
 
 
+async def test_tx_optimizer_reports_phases_in_order() -> None:
+    """on_phase announces each enabled search stage, in coarse → refine → verify order."""
+    from meshterm.services import tx_optimizer
+
+    device, admin, target, path = await _setup_link()
+    phases: list[str] = []
+
+    await tx_optimizer.optimize_tx_power(
+        device, target.name, admin, path,
+        samples_per_level=2, coarse_step=6, apply=False, cooldown_s=0,
+        on_phase=phases.append,
+    )
+    assert phases == list(tx_optimizer.PHASES)
+
+    phases.clear()
+    await tx_optimizer.optimize_tx_power(
+        device, target.name, admin, path,
+        samples_per_level=2, coarse_step=6, refine=False, verify=False,
+        apply=False, cooldown_s=0, on_phase=phases.append,
+    )
+    assert phases == ["coarse"]  # disabled stages are never announced
+
+
 def test_select_best_prefers_reliability_then_lowest_power() -> None:
     """A 100%-reliable level beats a flakier higher-SNR one; ties go to lower TX."""
     from meshterm.core.models import TraceStats, TxLevelResult