diff --git a/meshterm/core/models.py b/meshterm/core/models.py index f379a51..56d60e9 100644 --- a/meshterm/core/models.py +++ b/meshterm/core/models.py @@ -15,6 +15,14 @@ from typing import Optional #: Label used for our own (local) device when framing a trace path's endpoints. LOCAL_DEVICE_LABEL = "us" +#: Sentinel ``target`` recorded on traces run by the *Trace path* feature, which walks a +#: hand-composed route with no destination node at all (a trace is one walked path; the +#: notion of a target is pure UX). The parentheses keep it from colliding with contact +#: names. History queries key on it: ``Repository.latest_trace`` seeds the path screen +#: from it, while ``Repository.traced_targets`` excludes it so composed walks never +#: surface in the target picker's "Recently traced" list. +PATH_TRACE_TARGET = "(path)" + # MeshCore advert types — the low nibble of an advert's flag byte — classify the kind of # node an advert announces. Repeaters are fixed infrastructure and carry the mesh, so # location-aware features (e.g. the map) prioritise them over ordinary leaf nodes. diff --git a/meshterm/persistence/repository.py b/meshterm/persistence/repository.py index ef25606..8b097d7 100644 --- a/meshterm/persistence/repository.py +++ b/meshterm/persistence/repository.py @@ -14,6 +14,7 @@ from pathlib import Path from typing import Any, Optional from ..core.models import ( + PATH_TRACE_TARGET, ChatMessage, HeardNode, Hop, @@ -379,6 +380,8 @@ class Repository: This feeds the target picker's *Recently traced* section, so the list is deliberately short and recency-ordered: an all-time tally only ever grows, burying current work under stale names (``--mock`` targets included). + Target-less path walks (recorded under :data:`~meshterm.core.models. + PATH_TRACE_TARGET`) are excluded — they aren't destinations one can pick. Args: limit: Maximum number of distinct targets to return. @@ -387,9 +390,9 @@ class Repository: Distinct target names, most recently traced first. """ rows = self._conn.execute( - "SELECT target, MAX(id) AS latest FROM traces " + "SELECT target, MAX(id) AS latest FROM traces WHERE target != ? " "GROUP BY target ORDER BY latest DESC LIMIT ?", - (limit,), + (PATH_TRACE_TARGET, limit), ).fetchall() return [row["target"] for row in rows] diff --git a/meshterm/tools/trace.py b/meshterm/tools/trace.py index 16b6094..38092c6 100644 --- a/meshterm/tools/trace.py +++ b/meshterm/tools/trace.py @@ -1,17 +1,22 @@ -"""The ``trace`` tool: path traces to a target, live in the menu, one-shot on the CLI. +"""The trace tools: path traces, live in the menu, one-shot on the CLI. -Every invocation transmits exactly *one* trace — repeaters penalize (and can blacklist) -nodes that burst traffic, so repeat sampling is always a human decision, never a loop. +A trace is one walked path — the protocol has no destination field, so "target" is +purely a UX notion — and the menu splits the feature along exactly that line: -In the interactive menu this opens the live trace screen -(:mod:`meshterm.ui.trace_screen`): pick a target from a recency-ordered list and the -screen opens armed but idle — Enter runs one trace under an abortable tracing dialog -(press again to sample more; the screen aggregates the session), ``p`` composes a forced -path hop by hop from the topology observed in received traffic (and, standing on a -repeater with stored admin credentials, can fetch that repeater's live neighbour table -for suggestions our own radio never heard), and ``x`` explores ranked path scenarios -(with an optional probe that measures each candidate once and offers the winner). On the -CLI it stays a scriptable one-shot: run one trace, print the route and per-hop SNR, exit. +* **Trace target** (``trace``): *can I reach this node?* Pick a target from a + recency-ordered list; the live screen composes/explores symmetric routes that turn + at the target and come home over the mirrored hops. +* **Trace path** (``trace-path``): *how far can a route I build carry?* No target and + no picker — the whole walk is composed hop by hop and only has to end within our + own earshot. Traces record under the ``(path)`` sentinel, keeping composed walks + out of the target picker's history. + +Both open the live trace screen (:mod:`meshterm.ui.trace_screen`) armed but idle: +nothing transmits until Trace is committed, which runs the chosen sample count +(1–8 traces, paced between transmissions — repeaters penalize, and can blacklist, +nodes that burst traffic, so multi-trace runs always sleep a cooldown between sends). +On the CLI each stays a scriptable one-shot: run one trace, print the route and +per-hop SNR, exit. Both front ends persist identically: one ``runs`` row per trace, recorded under it, so stored history reads the same no matter where it came from. @@ -25,7 +30,7 @@ import typer from rich.text import Text from ..context import AppContext -from ..core.models import LOCAL_DEVICE_LABEL, Contact, TraceStats +from ..core.models import LOCAL_DEVICE_LABEL, PATH_TRACE_TARGET, Contact, TraceStats from ..services import trace_runner from ..ui.widgets import stats_panel from .base import Tool, ToolResult, register @@ -42,8 +47,8 @@ class TraceTool(Tool): """Trace the path to a target and watch per-hop SNR, live or scripted.""" name = "trace" - title = "Trace" - help = "Trace the path to a target and watch per-hop SNR live" + title = "Trace target" + help = "Trace a target over a mirrored route and watch per-hop SNR" category = "Diagnostics" order = 10 @@ -175,7 +180,7 @@ class TraceTool(Tool): items.append(Choice(title="Back", value=_BACK)) choice = await ctx.ui.select( - "Trace — pick a target", items, default=default, wrap=False + "Trace target — pick a target", items, default=default, wrap=False ) if choice in (None, _BACK): return None @@ -186,8 +191,9 @@ class TraceTool(Tool): async def _run_cli(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Run one trace, persist it, and print the route and per-hop summary. - One transmission per invocation is a hard rule — repeaters penalize (and can - blacklist) nodes that burst traffic — so sampling means invoking again. + One transmission per invocation is a hard rule on the CLI — repeaters + penalize (and can blacklist) nodes that burst traffic — so scripted sampling + means invoking again (with your own pacing between runs). Args: ctx: Shared application context. @@ -196,69 +202,10 @@ class TraceTool(Tool): Returns: A :class:`ToolResult` with the trace's outcome. """ - target = params["target"] - run_id = params["_run_id"] - device = await ctx.device() - - # Our own device's name labels both endpoints of the path (first hop's origin, - # last hop's destination); fall back to a neutral label if it's unavailable. - self_info = await device.get_self_info() - device_label = str(self_info.get("name") or LOCAL_DEVICE_LABEL) - # Our own public key, so the route's endpoints (us) carry a hash like every - # other hop, addressed at the same path-hash width. - device_hash = str(self_info.get("public_key") or "") or None - - # Resolve repeater hashes in the results to contact names where we know them, - # so the tables read as names instead of opaque hex prefixes. - contacts = await device.get_contacts() - resolve = trace_runner.make_node_resolver(contacts) - - # Resolve an optional forced path (names/hex) to the hex string the radio wants. - # Leaving it blank is fully supported: the device reuses the route it already - # learned for the target (or floods if it has none). - path: Optional[str] = None - path_spec = params.get("path") - if path_spec: - path = trace_runner.parse_trace_path(path_spec, contacts) - ctx.ui.note(f"[muted]forcing path:[/muted] [brand]{path}[/brand]") - else: - ctx.ui.note("[muted]path: auto (device-routed)[/muted]") - - with ctx.ui.progress("trace") as progress: - task = progress.add_task(f"tracing {target}", total=1) - result = await device.run_trace(target, path=path) - ctx.repo.record_trace(run_id, result) - progress.advance(task) - - # The stats panel renders route + per-hop readings for the single trace (its - # medians collapse to the readings themselves). - stats = TraceStats.from_traces(target, [result]) - ctx.ui.show( - stats_panel( - stats, - device_label, - resolve, - route=result if result.success else None, - device_hash=device_hash, - ) + return await _trace_once_cli( + ctx, params["_run_id"], target=params["target"], path_spec=params.get("path") ) - summary: dict[str, Any] = { - "target": target, - "success": result.success, - "hops": result.hop_count if result.success else None, - "min_snr": result.min_snr, - "rtt_ms": result.round_trip_ms, - } - if path: - summary["path"] = path - via = f" via [brand]{path}[/brand]" if path else "" - if result.success: - message = f"[ok]✓[/ok] traced [brand]{target}[/brand]{via}" - else: - message = f"[err]✗[/err] no reply from [brand]{target}[/brand]{via}" - return ToolResult(summary=summary, message=message) - def register_cli(self, app: typer.Typer) -> None: """Register the ``trace`` subcommand. @@ -283,6 +230,172 @@ class TraceTool(Tool): run_tool_command(self, tool_params) +@register +class TracePathTool(Tool): + """Walk a hand-composed route — no target — and see how far it carries.""" + + name = "trace-path" + title = "Trace path" + help = "Walk a hand-composed route — out and back your way" + category = "Diagnostics" + order = 11 + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """No parameters to gather — a path walk has no target to pick. + + Args: + ctx: Shared application context. + + Returns: + ``{"live": True}`` — the route itself is composed on the screen. + """ + return {"live": True} + + async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Run — without the outer run row on the live path (see :class:`TraceTool`). + + 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 path-walk screen (menu) or run one persisted trace (CLI). + + Args: + ctx: Shared application context. + params: ``live`` from the menu; or ``path`` (str, required) and the + injected ``_run_id`` from the CLI. + + Returns: + A :class:`ToolResult` with the walk's outcome. + """ + if params.get("live"): + from ..ui.trace_screen import open_trace_path + + traces = await open_trace_path(ctx) + return ToolResult(summary={"traces": traces}) + return await _trace_once_cli( + ctx, params["_run_id"], target=PATH_TRACE_TARGET, path_spec=params["path"] + ) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``trace-path`` subcommand. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + @app.command( + name=self.name, + help="Walk a composed route once (no target) and show per-hop SNR", + ) + def _trace_path( + path: str = typer.Option( + ..., + "--path", + "-p", + help="The whole walk: comma-separated contact names/hex prefixes " + "(must end within earshot of this node)", + ), + ) -> None: + run_tool_command(self, {"path": path}) + + +async def _trace_once_cli( + ctx: AppContext, run_id: int, *, target: str, path_spec: Optional[str] +) -> ToolResult: + """Run one persisted trace and print the route and per-hop summary (CLI body). + + Shared by both scripted trace commands: ``trace`` passes its target (path + optional — the device routes without one), ``trace-path`` passes the + :data:`~meshterm.core.models.PATH_TRACE_TARGET` sentinel with a required path. + + Args: + ctx: Shared application context. + run_id: The already-open run row to record under. + target: The label the trace persists under. + path_spec: The forced route as typed (names/hex, comma-separated), if any. + + Returns: + A :class:`ToolResult` with the trace's outcome. + """ + is_walk = target == PATH_TRACE_TARGET + device = await ctx.device() + + # Our own device's name labels both endpoints of the path (first hop's origin, + # last hop's destination); fall back to a neutral label if it's unavailable. + self_info = await device.get_self_info() + device_label = str(self_info.get("name") or LOCAL_DEVICE_LABEL) + # Our own public key, so the route's endpoints (us) carry a hash like every + # other hop, addressed at the same path-hash width. + device_hash = str(self_info.get("public_key") or "") or None + + # Resolve repeater hashes in the results to contact names where we know them, + # so the tables read as names instead of opaque hex prefixes. + contacts = await device.get_contacts() + resolve = trace_runner.make_node_resolver(contacts) + + # Resolve an optional forced path (names/hex) to the hex string the radio wants. + # Leaving it blank is fully supported for a *target* trace: the device reuses the + # route it already learned (or floods if it has none). A path walk always has one. + path: Optional[str] = None + if path_spec: + path = trace_runner.parse_trace_path(path_spec, contacts) + ctx.ui.note(f"[muted]forcing path:[/muted] [brand]{path}[/brand]") + else: + ctx.ui.note("[muted]path: auto (device-routed)[/muted]") + + with ctx.ui.progress("trace") as progress: + task = progress.add_task( + "walking the path" if is_walk else f"tracing {target}", total=1 + ) + result = await device.run_trace(target, path=path) + ctx.repo.record_trace(run_id, result) + progress.advance(task) + + # The stats panel renders route + per-hop readings for the single trace (its + # medians collapse to the readings themselves). + stats = TraceStats.from_traces(target, [result]) + ctx.ui.show( + stats_panel( + stats, + device_label, + resolve, + route=result if result.success else None, + device_hash=device_hash, + ) + ) + + summary: dict[str, Any] = { + "target": target, + "success": result.success, + "hops": result.hop_count if result.success else None, + "min_snr": result.min_snr, + "rtt_ms": result.round_trip_ms, + } + if path: + summary["path"] = path + via = f" via [brand]{path}[/brand]" if path else "" + if is_walk: + if result.success: + message = f"[ok]✓[/ok] the path came home{via}" + else: + message = f"[err]✗[/err] no reply{via}" + elif result.success: + message = f"[ok]✓[/ok] traced [brand]{target}[/brand]{via}" + else: + message = f"[err]✗[/err] no reply from [brand]{target}[/brand]{via}" + return ToolResult(summary=summary, message=message) + + def _recent_targets(stored: list[str], contacts: list[Contact]) -> list[str]: """Clean the stored recent-target names for the picker's *Recently traced* section. diff --git a/meshterm/ui/path_composer.py b/meshterm/ui/path_composer.py index 3703232..9a6b317 100644 --- a/meshterm/ui/path_composer.py +++ b/meshterm/ui/path_composer.py @@ -1,11 +1,25 @@ """The trace path composer: build a forced route hop by hop, guided by observed links. -A floating dialog over the live trace screen. The route under construction reads across -the top — ``us → hop → … → target ( → mirror → us)`` — and beneath it sits a suggestion -list: the nodes the topology evidence says the path's current tail can hear, strongest -observed link first (see :meth:`~meshterm.services.topology.MeshTopology.next_hops`). -Every link is bidirectional evidence, so a path that was ever *received* through two -nodes proposes that link in either direction. +A floating dialog over a live trace screen. The route under construction reads across +the top — ``us → hop → … → us`` — and beneath it sits a suggestion list: the nodes the +topology evidence says the path's current tail can hear, strongest observed link first +(see :meth:`~meshterm.services.topology.MeshTopology.next_hops`). Every link is +bidirectional evidence, so a path that was ever *received* through two nodes proposes +that link in either direction. + +The composer serves both trace features, and which one owns it decides how the route's +return half is written (the trace protocol has no separate return-path field — the spec +sent to the device is one walk that ends within our earshot): + +* **Target mode** (*Trace target* — a ``target_id`` is pinned): only the outbound leg + is composed. The target stays pinned as the turning point and the return is the + outbound hops mirrored — the preview resolves and dims that half, since it isn't + yours to choose. An *Auto* action hands routing back to the device. +* **Path mode** (*Trace path* — no target at all): the whole walk is yours, hop by + hop, until the route comes back within earshot of us — only the final landing on + our own node is filled in (and dimmed). The preview opens as just ``us → us``. + There is no *Auto* action, because with no destination there is nothing for the + device to route to. Interaction, following the reorder screen's cursor-over-rows-and-actions pattern: @@ -18,22 +32,12 @@ Interaction, following the reorder screen's cursor-over-rows-and-actions pattern neighbours* row asks that repeater over the mesh for its own neighbour table — fresh second-vantage evidence exactly where composing ran out of it. The dialog resolves :class:`FetchNeighbours` and the owning flow fetches, refreshes the - topology, and reopens the composer mid-thought (hops and mode preserved). -* Enter on **Use this path** commits the composed spec; **Auto** hands routing back to - the device; Esc cancels with no change. + topology, and reopens the composer mid-thought (hops preserved). +* Enter on **Use this path** commits the composed spec; Esc cancels with no change. -The trace protocol has no separate return-path field — the spec sent to the device is -one walk, out to the target and back into our earshot — and the ⇄ action row picks how -the return half is written: - -* **Symmetric** (the default): only the outbound leg is composed. The target stays - pinned as the turning point and the return is the outbound hops mirrored — the - preview resolves and dims that half, since it isn't yours to choose. -* **Asymmetric**: the whole walk is yours, hop by hop, target included, until the route - comes back within earshot of us — only the final landing on our own node is filled in. - Toggling out of symmetric materializes the mirrored return as editable hops, so the - route is unchanged until you reshape it; toggling back keeps what precedes the target - as the outbound leg and regenerates the mirror. +Nodes render exactly as the trace screen's route line does — ``Name (hash)`` with the +name in the brand colour and the hash muted in parentheses — so a node reads the same +wherever it appears. """ from __future__ import annotations @@ -51,7 +55,8 @@ from .tui.screen import Screen from .widgets import _age_seconds, _format_age, _shorten_hash #: Sentinel spec meaning "no forced path — let the device route" (the trace screen's -#: empty-spec convention). +#: empty-spec convention). Only meaningful in target mode; a path walk has no target +#: for the device to route to, so path mode never resolves it. AUTO_SPEC = "" #: How many suggestions to show at most; beyond this the evidence is too weak to matter @@ -60,7 +65,6 @@ _MAX_SUGGESTIONS = 12 #: Action-row sentinels (kept distinct from suggestion rows, which carry node ids). _USE = "use" -_MODE = "mode" _AUTO = "auto" _CANCEL = "cancel" @@ -86,11 +90,11 @@ class FetchNeighbours: class PathComposerScreen(Screen): - """Compose a forced trace path step by step, symmetric or hand-routed. + """Compose a forced trace path step by step, mirrored at a target or hand-routed. - Resolves with the finished spec (comma-separated hex — in symmetric mode the - outbound hops, the target, then the mirrored return; in asymmetric mode the - composed walk verbatim), :data:`AUTO_SPEC` for device routing, or + Resolves with the finished spec (comma-separated hex — in target mode the outbound + hops, the target, then the mirrored return; in path mode the composed walk + verbatim), :data:`AUTO_SPEC` for device routing (target mode only), or :data:`~meshterm.ui.tui.screen.CANCEL`. """ @@ -99,47 +103,50 @@ class PathComposerScreen(Screen): def __init__( self, *, - target_id: str, - target_hash: str, - target_label: str, device_label: str, topology: MeshTopology, width_bytes: int, + target_id: Optional[str] = None, + target_hash: Optional[str] = None, + target_label: Optional[str] = None, + device_hash: Optional[str] = None, hops: Optional[list[str]] = None, fetch_nodes: frozenset[str] = frozenset(), - symmetric: bool = True, ) -> None: """Build the composer. + A pinned target (all three ``target_*`` arguments) selects target mode: the + target is excluded from suggestions (the path implicitly turns at it) and the + emitted spec appends it plus the mirrored return. With no target the composer + is in path mode — the whole walk is composed by hand and committed verbatim. + Args: - target_id: The target's canonical id. In symmetric mode it is excluded from - suggestions (the path implicitly turns at it); in asymmetric mode it is - suggested like any node, since the user routes through it themselves. + device_label: Our own node's name, opening and closing the route preview. + topology: The evidence graph suggestions are drawn from. + width_bytes: Preferred per-hop path-hash width (bytes) for the emitted spec. + target_id: The pinned target's canonical id, or ``None`` for path mode. target_hash: The target's hex hash used as the symmetric spec's turning point (its full key prefix; truncated to the spec width at commit). target_label: The target's display name for the route preview. - device_label: Our own node's name, opening the route preview. - topology: The evidence graph suggestions are drawn from. - width_bytes: Preferred per-hop path-hash width (bytes) for the emitted spec. + device_hash: Our own public key, so the preview's endpoints carry a hash + exactly like the trace screen's route line. hops: Canonical ids of already-composed hops (reopening the dialog resumes where the user left off). fetch_nodes: Canonical ids whose live neighbour table can be fetched (repeater contacts with a public key); when the path's tail is one of them, the *fetch neighbours* row appears. - symmetric: Whether the return leg is auto-mirrored (the default) or the - user is composing the whole walk themselves (see :meth:`_toggle_mode`). """ super().__init__() - self.title = f"compose path · {target_label}" + self.title = f"compose path · {target_label}" if target_label else "compose path" self._target_id = target_id - self._target_hash = target_hash.lower().removeprefix("0x") + self._target_hash = (target_hash or "").lower().removeprefix("0x") self._target_label = target_label self._device_label = device_label + self._device_hash = (device_hash or "").lower().removeprefix("0x") self._topology = topology self._width_bytes = width_bytes self._hops: list[str] = list(hops or []) self._fetch_nodes = fetch_nodes - self._symmetric = symmetric self._entry = "" self._index = 0 @@ -148,13 +155,13 @@ class PathComposerScreen(Screen): """The composed hops so far (for re-seeding after a fetch).""" return list(self._hops) - @property - def symmetric(self) -> bool: - """Whether the return leg is auto-mirrored (for re-seeding after a fetch).""" - return self._symmetric - # --- state ----------------------------------------------------------------- + @property + def _mirrored(self) -> bool: + """Whether the return leg is the pinned target's mirror (target mode).""" + return self._target_id is not None + def _tail(self) -> str: """The node the path currently ends at (us until a hop is added).""" return self._hops[-1] if self._hops else self._topology.self_id @@ -162,17 +169,16 @@ class PathComposerScreen(Screen): def _suggestions(self) -> list: """The current suggestion rows, filtered by the typed entry. - Symmetric mode excludes the target (the path implicitly turns at it) and every + Target mode excludes the target (the path implicitly turns at it) and every used hop (revisiting one on the outbound leg is never useful — the mirror - already recrosses it). Asymmetric mode only excludes ourselves and the tail: - the target must be routable *through*, and a return leg legitimately reuses - outbound repeaters. + already recrosses it). Path mode only excludes ourselves and the tail: a + return leg legitimately reuses outbound repeaters. Returns: The (possibly filtered) :class:`~meshterm.services.topology.HopSuggestion` list for the path's tail, capped at :data:`_MAX_SUGGESTIONS`. """ - if self._symmetric: + if self._mirrored: exclude = frozenset({self._topology.self_id, self._target_id, *self._hops}) else: exclude = frozenset({self._topology.self_id, self._tail()}) @@ -205,108 +211,82 @@ class PathComposerScreen(Screen): if self._tail() in self._fetch_nodes: rows.append(("fetch", self._tail())) rows.append(("action", _USE)) - rows.append(("action", _MODE)) - rows.append(("action", _AUTO)) + if self._mirrored: # a path walk has no target for the device to route to + rows.append(("action", _AUTO)) rows.append(("action", _CANCEL)) return rows def _spec(self) -> str: """Render the composed route as the spec the trace command will send. - Symmetric mode appends the target and the mirrored return leg (see - :func:`~meshterm.services.topology.render_forced_spec`); asymmetric mode - renders the composed walk verbatim (:func:`~meshterm.services.topology. + Target mode appends the target and the mirrored return leg (see + :func:`~meshterm.services.topology.render_forced_spec`); path mode renders + the composed walk verbatim (:func:`~meshterm.services.topology. render_custom_spec`) — empty until it has at least one hop. """ - if self._symmetric: + if self._mirrored: return render_forced_spec( tuple(self._hops), self._target_hash, self._width_bytes ) return render_custom_spec(tuple(self._hops), self._width_bytes) - def _toggle_mode(self) -> None: - """Flip between the symmetric mirror and hand-composing the whole walk. - - The route itself never changes on a toggle — only who owns its return half: - - * symmetric → asymmetric *materializes* the mirror the preview was showing - (target and reversed hops become real, editable hops), so backspace can now - peel the return leg apart; - * asymmetric → symmetric keeps whatever precedes the target's first appearance - as the outbound leg (everything after it was return-leg territory, which the - mirror now regenerates); with the target nowhere in the walk, every hop is - kept as outbound. - """ - if self._symmetric: - self._hops = [*self._hops, self._target_id, *reversed(self._hops)] - elif self._target_id in self._hops: - self._hops = self._hops[: self._hops.index(self._target_id)] - self._symmetric = not self._symmetric - # The toggle changes the suggestion set (the target's excludability), shifting - # row numbers — re-anchor the cursor onto this same action row. - rows = self._rows() - self._index = next( - i for i, (kind, payload) in enumerate(rows) - if kind == "action" and payload == _MODE - ) - # --- rendering --------------------------------------------------------------- def _node_text(self, node: str, *, dim: bool = False) -> Text: - """A node as ``Name (hash)`` when known, else its bare hash. + """One route node, rendered exactly as the trace screen's route line does. - The hash is shown at the spec's preferred path-hash width — the same width - every other trace-feature view (planned route, completed route, hop table) - truncates to — so a node reads the same length everywhere instead of the - fixed, arbitrary slice a naive ``node[:n]`` would give it. The target keeps - its ok-tinted name wherever it appears (pinned as the symmetric turning point, - or added mid-walk in asymmetric mode), so the route's whole point stays - visible at a glance. + ``Name (hash)`` when known — name in the brand colour, hash muted in + parentheses — a bare brand hash when not, and our own device as its accent + label plus hash. Hashes show at the spec's preferred path-hash width, the + same width every other trace-feature view truncates to, so a node reads the + same length everywhere. Args: node: The node's canonical id (or ``self_id`` for our own device). - dim: Whether to render in the resolved-return-leg's faint style rather - than the outbound leg's normal accent/brand styling. + dim: Whether to render in the resolved-return-leg's uniform faint style + rather than the normal route colours. """ if node == self._topology.self_id: - return Text(self._device_label, style="faint" if dim else "accent") + text = Text(self._device_label, style="faint" if dim else "accent") + if self._device_hash: + shown = _shorten_hash(self._device_hash, self._width_bytes) + text.append(f" ({shown})", style="faint" if dim else "muted") + return text if node == self._target_id: - text = Text(self._target_label, style="faint" if dim else "ok") - text.append( - f" ({_shorten_hash(self._target_hash, self._width_bytes)})", - style="faint", - ) - return text - name = self._topology.display_name(node) - style = "faint" if dim else "brand" - shown = _shorten_hash(node, self._width_bytes) - if name: - text = Text(name, style=style) - text.append(f" ({shown})", style="faint") - return text - return Text(shown, style=style) + name: Optional[str] = self._target_label + hash_source = self._target_hash or node + else: + name = self._topology.display_name(node) + hash_source = node + shown = _shorten_hash(hash_source, self._width_bytes) + if not name: + return Text(shown, style="faint" if dim else "brand") + text = Text(name, style="faint" if dim else "brand") + text.append(f" ({shown})", style="faint" if dim else "muted") + return text def _route_preview(self) -> Text: """The route under construction, endpoints filled in automatically. - Symmetric mode shows the composed outbound in full color, then the pinned + Target mode shows the composed outbound in full colour, then the pinned target and the mirrored return resolved and dimmed right alongside it — the - dimming reads as "this half isn't yours to compose". Asymmetric mode shows - every composed hop in full color (they are all yours), with only the final - landing back on our own node dimmed. + dimming reads as "this half isn't yours to compose". Path mode shows every + composed hop in full colour (they are all yours) between our own node at + both ends, with only the final landing back on us dimmed. """ - text = Text(self._device_label, style="accent") + text = Text() + text.append_text(self._node_text(self._topology.self_id)) for hop in self._hops: text.append(" → ", style="muted") text.append_text(self._node_text(hop)) - if self._symmetric: + if self._mirrored: text.append(" → ", style="muted") text.append_text(self._node_text(self._target_id)) for hop in reversed(self._hops): text.append(" → ", style="faint") text.append_text(self._node_text(hop, dim=True)) text.append(" → ", style="faint") - text.append(self._device_label, style="faint") + text.append_text(self._node_text(self._topology.self_id, dim=True)) return text def _suggestion_text(self, suggestion) -> Text: # noqa: ANN001 @@ -325,16 +305,6 @@ class PathComposerScreen(Screen): text.append(f" · {tags}", style="faint") return text - @staticmethod - def _mode_text(symmetric: bool) -> Text: - """The ⇄ action row's label for a mode (Enter on the row flips to the other).""" - text = Text("⇄ ") - if symmetric: - text.append("Symmetric — return mirrors the outbound") - else: - text.append("Asymmetric — you compose out and back") - return text - def _row_text(self, kind: str, payload: object) -> Text: """The display text for one cursor-addressable row.""" if kind == "custom": @@ -353,25 +323,18 @@ class PathComposerScreen(Screen): if payload == _USE: label = Text.assemble(("✓ ", "ok"), "Use this path") spec = self._spec() - # An asymmetric walk with no hops yet has no spec to commit. + # A path walk with no hops yet has no spec to commit. label.append(f" ({spec})" if spec else " (add a hop first)", style="muted") return label - if payload == _MODE: - return self._mode_text(self._symmetric) if payload == _AUTO: return Text("Auto — let the device route") return Text.assemble(("✗ ", "err"), "Cancel") @property def dialog_width(self) -> int: - """Natural outer width hugging the widest row (compositor still caps it). - - Sized for *both* mode labels, not just the active one, so toggling ⇄ never - resizes the box mid-interaction. - """ + """Natural outer width hugging the widest row (compositor still caps it).""" widths = [cell_len(self.title), cell_len(self.footer_hint)] widths.append(cell_len(self._route_preview().plain)) - widths.append(cell_len(self._mode_text(not self._symmetric).plain) + 2) for kind, payload in self._rows(): widths.append(cell_len(self._row_text(kind, payload).plain) + 2) return max(widths, default=20) + 8 @@ -440,10 +403,8 @@ class PathComposerScreen(Screen): self.resolve(FetchNeighbours(node=str(payload))) elif payload == _USE: spec = self._spec() - if spec: # an empty asymmetric walk would masquerade as AUTO_SPEC + if spec: # an empty path walk would masquerade as AUTO_SPEC self.resolve(spec) - elif payload == _MODE: - self._toggle_mode() elif payload == _AUTO: self.resolve(AUTO_SPEC) else: diff --git a/meshterm/ui/trace_screen.py b/meshterm/ui/trace_screen.py index e5e840b..5b77262 100644 --- a/meshterm/ui/trace_screen.py +++ b/meshterm/ui/trace_screen.py @@ -1,49 +1,59 @@ -"""The live trace screen: compose a route from observed topology, then watch it answer. +"""The live trace screens: compose a route from observed topology, then watch it answer. -This is the interactive face of the ``trace`` tool (the scripted CLI keeps its one-shot -table output). Where the old flow started tracing the instant a target was picked, the -screen now opens *armed but idle*: the target's last-known route shows, the path is +This is the interactive face of the two trace tools (the scripted CLI keeps its one-shot +table output). A trace is one walked path — the protocol has no destination field, so +"target" is purely a UX notion — and the two menu features split along exactly that line: + +* **Trace target** (:func:`open_trace`) answers *can I reach this node?* The route is + symmetric: compose (or explore) the outbound leg, and the return is those hops + mirrored around the pinned target. +* **Trace path** (:func:`open_trace_path`) answers *how far can a route I build carry?* + There is no target at all: the whole walk is composed by hand, out and back by + whatever way you choose, and it only has to end within our own earshot. + +Both open the same screen, *armed but idle*: the last-known route shows, the path is whatever you make it, and nothing transmits until you say so. An action list drives it — -↑/↓ select a row, Enter commits it, and each row keeps its hotkey: +↑/↓ select a row, Enter commits it: -* **Trace** (Enter — the cursor opens here, so plain Enter still just traces) runs a - *single* trace. While it flies, a floating *tracing* dialog (spinner, Abort) sits over - the screen — the reply streams into the log behind it, and Esc in the dialog cancels - the trace without leaving the screen. One transmission per keypress is deliberate: - repeaters penalize chatty nodes (flood detection can blacklist us), so sampling is - left human-paced — commit Trace again and the screen keeps aggregating every trace of - the session into its medians. -* **Path width** (``w``) floats a small dialog picking the per-hop path-hash width (1, - 2, 4, or 8 bytes — the leading key slice forced hops are addressed by). It seeds from - the device's routing width; changing it re-renders a standing forced path at the new - width and shapes every spec the composer and explorer emit after it. -* **Compose path** (``p``) opens the path composer (:mod:`~meshterm.ui.path_composer`): - build the route hop by hop, each step suggested from the links observed in *received* +* **Compose path** opens the path composer (:mod:`~meshterm.ui.path_composer`): build + the route hop by hop, each step suggested from the links observed in *received* traffic — traces, firmware-learned contact routes, and RX-logged packet paths — strongest first, with raw hex entry for nodes the data has never seen. Standing on a repeater you hold admin credentials for, the composer can also *fetch that repeater's neighbour table* over the mesh (login required; firmware ignores guests): second-vantage evidence, persisted and folded straight back into the suggestions. - By default only the outbound leg is composed and the return is those hops mirrored — - the spec sent to the device spells out the whole boomerang, since the trace protocol - has no separate return-path field — but the composer's ⇄ row switches to asymmetric, - where the entire walk (out *and* home) is yours to route. -* **Explore paths** (``x``) explores scenarios: ranked candidate routes to the target - straight from the topology evidence (the device's own learned route, the direct shot, - and the strongest observed alternatives). Adopt one directly — or probe them all, one - measured trace per candidate (the same single-transmission rule), persisted as - ``path_candidates`` rows and ranked reliability-first, with the winner offered for - adoption. Evidence proposes, measurement decides, you dispose. +* **Explore paths** (Trace target only — candidates need a destination) explores + scenarios: ranked candidate routes to the target straight from the topology evidence + (the device's own learned route, the direct shot, and the strongest observed + alternatives). Adopt one directly — or probe them all, one measured trace per + candidate, persisted as ``path_candidates`` rows and ranked reliability-first, with + the winner offered for adoption. Evidence proposes, measurement decides, you dispose. +* **Path width** floats a small dialog picking the per-hop path-hash width (1, 2, 4, + or 8 bytes — the leading key slice forced hops are addressed by). It seeds from the + device's routing width; changing it re-renders a standing forced path at the new + width and shapes every spec the composer and explorer emit after it. +* **Sample count** floats a dialog picking how many traces one Trace action runs + (1, 2, 3, 5, or 8). Multi-trace runs are *paced* — a cooldown sleeps between + transmissions, because repeaters penalize (and can blacklist) nodes that burst + traffic — and the in-flight dialog counts them off as replies stream in. +* **Trace** (the cursor opens here, so plain Enter still just traces) runs the chosen + number of traces. While they fly, a floating *tracing* dialog (spinner, progress, + Abort) sits over the screen — replies stream into the log behind it, and Esc in the + dialog cancels without leaving the screen; traces already recorded are kept. The + screen keeps aggregating every trace of the session into its medians. +* **Back** leaves the screen, exactly like Esc. Layout, top to bottom: the walked route (live when a reply has landed, else the planned -composed path, else the target's last stored trace), the run's robust aggregates, the +composed path, else the most recent stored trace), the run's robust aggregates, the action list, per-hop median SNR with quality bars, and the individual traces newest-first. The body scrolls with PgUp/PgDn/Home/End (↑/↓ belong to the action cursor, which only pins the view while it is actually being moved). Every trace is persisted exactly like a scripted run: one ``runs`` row per trace, recorded under it, so the stored history reads the same no matter which front end -produced it. +produced it. Path walks record under :data:`~meshterm.core.models.PATH_TRACE_TARGET`, +keeping them out of the target picker's history while still seeding the path screen's +previous-route line. """ from __future__ import annotations @@ -56,7 +66,7 @@ from rich.console import Group, RenderableType from rich.table import Table from rich.text import Text -from ..core.models import TraceResult, TraceStats +from ..core.models import PATH_TRACE_TARGET, TraceResult, TraceStats from ..services import trace_runner from .theme import snr_style from .tui.render import render_lines, render_to_ansi @@ -78,14 +88,19 @@ _BAR_WIDTH = 16 _BAR_SNR_MIN = -15.0 _BAR_SNR_MAX = 10.0 +#: The sample counts the Sample count dialog offers: how many traces one Trace action +#: runs, paced between transmissions. +SAMPLE_CHOICES = (1, 2, 3, 5, 8) + #: A single-trace runner: ``(path_spec, on_trace)`` → runs exactly one trace, handing -#: the result to ``on_trace`` when it lands. Provided by :func:`open_trace`, which closes +#: the result to ``on_trace`` when it lands. Provided by the session openers, which close #: over the device, repository, and settings so the screen stays free of persistence #: concerns. TraceOnce = Callable[[str, Callable[[TraceResult], None]], Awaitable[None]] #: A path picker flow: takes the current spec, runs its own dialogs over the screen, and #: resolves to the new spec (``""`` = device-routed) or ``None`` to keep the current one. +#: The sample-count flow shares the shape and simply always resolves ``None``. PathFlow = Callable[[str], Awaitable[Optional[str]]] @@ -113,7 +128,7 @@ def snr_bar(snr: Optional[float], width: int = _BAR_WIDTH) -> Text: class TracingDialog(Screen): """The floating in-flight dialog: a spinner chip, live progress, and Abort. - Pushed over the trace screen while a trace (or scenario probe) transmits, so the + Pushed over the trace screen while traces (or a scenario probe) transmit, so the activity — and the way out — is unmissable while replies keep streaming into the screen behind it. Enter, Esc, or Space aborts via the injected callback; the owner pops the dialog when the work finishes, so it never resolves a value of its own. @@ -186,62 +201,76 @@ class TracingDialog(Screen): class TraceScreen(Screen): - """A full-screen live trace session for one target — armed, but idle until told. + """A full-screen live trace session — armed, but idle until told. - An action list carries the verbs: *Trace* (one transmission per commit, because - repeaters can blacklist nodes that burst traffic; a floating dialog with Abort - rides on top while it flies), *Path width* (``w``), *Compose path* (``p``), and - *Explore paths* (``x``). ↑/↓ move the cursor over the rows and Enter commits the - selected one — the cursor opens on Trace, so plain Enter still just traces. - PgUp/PgDn/Home/End scroll the body, and Esc backs out (cancelling any in-flight - trace; already-recorded traces are kept). + One screen serves both trace features; ``mode`` decides which. ``"target"`` traces + a pinned destination over a symmetric (mirrored) route and offers *Explore paths*; + ``"path"`` walks a hand-composed route with no target at all — no Explore (ranked + candidates need a destination), no device routing, and Trace stays inert until a + path exists to walk. + + ↑/↓ move the cursor over the action rows and Enter commits the selected one — the + cursor opens on Trace, so plain Enter still just traces. PgUp/PgDn/Home/End scroll + the body, and Esc (or the Back row) backs out, cancelling any in-flight trace; + already-recorded traces are kept. """ floating = False - #: The action rows, in display order (each also keeps a hotkey — see :meth:`handle`). - _ACTIONS: tuple[str, ...] = ("trace", "width", "compose", "explore") - def __init__( self, target: str, *, + mode: str = "target", device_label: str, device_hash: Optional[str], resolve: NodeResolver, session: Any, trace: TraceOnce, compose_path: PathFlow, - explore: PathFlow, pick_width: PathFlow, + pick_samples: PathFlow, width_bytes: Callable[[], int], + sample_count: Callable[[], int], + explore: Optional[PathFlow] = None, + pace_s: float = 1.0, previous: Optional[TraceResult] = None, ) -> None: """Create the screen (nothing transmits until the user commits Trace). Args: - target: The trace destination (contact name or key prefix). + target: The label traces persist under — the destination's name in target + mode, :data:`~meshterm.core.models.PATH_TRACE_TARGET` in path mode. + mode: ``"target"`` (symmetric, explorable) or ``"path"`` (hand-composed + walk, no destination). device_label: Our own node's name, labelling the route's endpoints. device_hash: Our own public key, so the endpoints carry a hash like every hop. resolve: Maps a hop's raw hash to a friendly contact name when known. session: The running TUI session (for repaints and the floating dialog). trace: Runs exactly one trace, handing back the result (see - :data:`TraceOnce`). + :data:`TraceOnce`); the screen loops it for multi-sample runs. compose_path: Opens the hop-by-hop path composer over this screen, seeded with the current spec; resolves to the new spec or ``None`` if cancelled. - explore: Opens the scenario browser/probe flow over this screen; resolves to - an adopted spec or ``None`` to keep the current one. pick_width: Floats the path-hash width dialog; resolves to the standing - spec re-rendered at the chosen width, or ``None`` when nothing changes - (same :data:`PathFlow` shape as the other flows). + spec re-rendered at the chosen width, or ``None`` when nothing changes. + pick_samples: Floats the sample-count dialog; always resolves ``None`` + (the count lives with the owner, read back via ``sample_count``). width_bytes: Reads the currently chosen per-hop width, for the action row's label (the owner holds the width, since the flows emit specs at it). - previous: The target's most recent stored trace, if any — its route seeds the - route line so the screen opens knowing the path history last saw. + sample_count: Reads the currently chosen sample count, for the action rows + and for how many traces one Trace commit runs. + explore: Opens the scenario browser/probe flow over this screen (target + mode only); resolves to an adopted spec or ``None`` to keep the current. + pace_s: Cooldown slept between the traces of a multi-sample run, so a + sampling session never reads as a burst to the repeaters. + previous: The most recent stored trace, if any — its route seeds the route + line so the screen opens knowing the path history last saw. """ super().__init__() - self.title = f"trace · {target}" + self.title = f"trace · {target}" if mode == "target" else "trace path" self._target = target + self._mode = mode + self._flight_label = target if mode == "target" else "path" self._device_label = device_label self._device_hash = device_hash self._resolve = resolve @@ -250,30 +279,42 @@ class TraceScreen(Screen): self._compose_path = compose_path self._explore = explore self._pick_width = pick_width + self._pick_samples = pick_samples self._width_bytes = width_bytes + self._sample_count = sample_count + self._pace_s = pace_s self._previous = previous self._path_spec = "" self._traces: list[TraceResult] = [] self._running = False self._dialog_open = False self._status = "" + self._progress: Optional[tuple[int, int]] = None # (current, total) mid-run self._spinner = Spinner() self._worker: Optional[asyncio.Task] = None self._flight: Optional[TracingDialog] = None - self._index = 0 # cursor over the action rows; 0 keeps Enter = trace + # The action rows, in display order. Explore needs a destination to rank + # candidates for, so path mode drops it; the cursor opens on Trace either way. + actions = ["compose"] + (["explore"] if mode == "target" else []) + actions += ["width", "samples", "trace", "back"] + self._actions: tuple[str, ...] = tuple(actions) + self._index = self._actions.index("trace") self._pin_cursor = False # only pin the view while ↑/↓ are actually in use # --- state ----------------------------------------------------------------- @property def footer_hint(self) -> str: # type: ignore[override] - """The footer keys, tracking whether a trace is in flight.""" + """The footer keys, tracking whether traces are in flight.""" if self._running: + if self._progress is not None and self._progress[1] > 1: + done, total = self._progress + return f"tracing {done}/{total}… · PgUp/PgDn scroll · Esc back" return "tracing… · PgUp/PgDn scroll · Esc back" return "↑↓ actions · Enter run · PgUp/PgDn scroll · Esc back" def start_trace(self) -> None: - """Kick off one trace in the background (no-op while one is already flying).""" + """Kick off a trace run in the background (no-op while one is already flying).""" if self._running: return self._running = True @@ -283,21 +324,33 @@ class TraceScreen(Screen): self._session.invalidate() async def _run_trace(self) -> None: - """Drive one trace to completion under the floating tracing dialog. + """Drive one Trace commit — the chosen number of traces — under the dialog. - The dialog is pushed for the duration and popped however the trace ends — + Multi-sample runs are paced: :attr:`_pace_s` sleeps between transmissions so + the repeaters never see a burst, and the dialog counts the run off as replies + land. The dialog is pushed for the duration and popped however the run ends — completion, failure, or abort — and its Abort wires straight to :meth:`cancel`, so the cancellation path is the same whether Esc lands on the dialog or the - screen. + screen. An aborted run keeps every trace already recorded. """ + total = max(1, int(self._sample_count())) dialog = TracingDialog( - f"tracing · {self._target}", spinner=self._spinner, on_abort=self.cancel + f"tracing · {self._flight_label}", spinner=self._spinner, on_abort=self.cancel ) self._flight = dialog self._session.push(dialog) ticker = asyncio.ensure_future(self._animate()) try: - await self._trace_once(self._path_spec, self._on_trace) + for done in range(total): + self._progress = (done + 1, total) + if total > 1: + dialog.status = f"trace {done + 1}/{total} · transmitting…" + self._session.invalidate() + await self._trace_once(self._path_spec, self._on_trace) + if done + 1 < total and self._pace_s > 0: + dialog.status = f"trace {done + 1}/{total} landed · pacing…" + self._session.invalidate() + await asyncio.sleep(self._pace_s) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - report inline, keep the screen alive @@ -313,6 +366,7 @@ class TraceScreen(Screen): self._session.pop(dialog) self._flight = None self._running = False + self._progress = None self._session.invalidate() async def _animate(self) -> None: @@ -330,7 +384,7 @@ class TraceScreen(Screen): self._session.invalidate() def cancel(self) -> None: - """Cancel any in-flight trace (already-recorded traces are kept).""" + """Cancel any in-flight trace run (already-recorded traces are kept).""" if self._worker is not None and not self._worker.done(): self._worker.cancel() @@ -346,17 +400,11 @@ class TraceScreen(Screen): if action == "enter": self._commit_action() elif action == "up": - self._index = (self._index - 1) % len(self._ACTIONS) + self._index = (self._index - 1) % len(self._actions) self._pin_cursor = True elif action == "down": - self._index = (self._index + 1) % len(self._ACTIONS) + self._index = (self._index + 1) % len(self._actions) self._pin_cursor = True - elif action == "text" and data.lower() == "p": - self._open_flow(self._compose_path) - elif action == "text" and data.lower() == "x": - self._open_flow(self._explore) - elif action == "text" and data.lower() == "w": - self._open_flow(self._pick_width) elif action == "pageup": self._pin_cursor = False self.scroll_pages(-1) @@ -375,22 +423,32 @@ class TraceScreen(Screen): def _commit_action(self) -> None: """Run the action row under the cursor (the flows guard against re-entry).""" - key = self._ACTIONS[self._index] + key = self._actions[self._index] if key == "trace": + # A path walk has nothing to transmit until a path exists — with no + # target, an empty spec can't fall back to device routing. + if self._mode == "path" and not self._path_spec: + return self.start_trace() elif key == "width": self._open_flow(self._pick_width) + elif key == "samples": + self._open_flow(self._pick_samples) elif key == "compose": self._open_flow(self._compose_path) - else: + elif key == "explore" and self._explore is not None: self._open_flow(self._explore) + elif key == "back": + self.cancel() + self.resolve(None) def _open_flow(self, flow: PathFlow) -> None: """Float a path-picking flow over the screen (one at a time, not mid-trace). The composer, the scenario explorer, and the width picker all resolve the same way: a new spec to adopt (``""`` returns routing to the device), or - ``None`` to leave the current path untouched. + ``None`` to leave the current path untouched (which is all the sample-count + flow ever resolves). Args: flow: The dialog flow to run with the current spec. @@ -422,11 +480,13 @@ class TraceScreen(Screen): stats = TraceStats.from_traces(self._target, self._traces) current = next((t for t in reversed(self._traces) if t.success), None) lines = render_lines( - Group(self._route_line(current), Text(), self._summary(stats)), width + Group(self._route_line(current), Text(), self._summary(stats, current)), width ) lines.append("") self._cursor: Optional[int] = None - for i, key in enumerate(self._ACTIONS): + for i, key in enumerate(self._actions): + if key == "back": + lines.append("") # Back is its own group, set apart like the menus do selected = i == self._index text = self._action_text(key, selected) text.no_wrap = True @@ -439,8 +499,9 @@ class TraceScreen(Screen): hash_bytes = current.path_hash_bytes if current is not None else None tail += [Text(), Text("Per-hop medians", style="accent")] tail.append(self._hops_table(stats, hash_bytes)) - tail += [Text(), Text("Traces", style="accent")] - tail.append(self._trace_log()) + if self._running or self._status or self._traces: + tail += [Text(), Text("Traces", style="accent")] + tail.append(self._trace_log()) lines.extend(render_lines(Group(*tail), width)) self._scroll_total = max(1, len(lines)) return lines @@ -455,24 +516,34 @@ class TraceScreen(Screen): return getattr(self, "_cursor", None) if self._pin_cursor else None def _action_text(self, key: str, selected: bool) -> Text: - """One action row: pointer, glyph, label, and the row's hotkey, muted.""" + """One action row: pointer, glyph, and label (current value inlined).""" text = Text("❯ " if selected else " ", style="brand" if selected else "") - if key == "trace": - text.append("▶ ", style="ok") - text.append("Trace — one transmission") + if key == "compose": + text.append("✎ ", style="brand") + text.append("Compose path") + elif key == "explore": + text.append("⚡ ", style="warn") + text.append("Explore paths") elif key == "width": w = self._width_bytes() text.append("⚙ ", style="accent") text.append(f"Path width — {w} byte{'s' if w != 1 else ''} per hop") - text.append(" w", style="muted") - elif key == "compose": - text.append("✎ ", style="brand") - text.append("Compose path") - text.append(" p", style="muted") + elif key == "samples": + n = self._sample_count() + text.append("# ", style="accent") + text.append(f"Sample count — {n} trace{'s' if n != 1 else ''}") + elif key == "trace": + text.append("▶ ", style="ok") + if self._mode == "path" and not self._path_spec: + text.append("Trace — compose a path first", style="muted") + else: + n = self._sample_count() + if n > 1: + text.append(f"Trace — {n} paced transmissions") + else: + text.append("Trace — one transmission") else: - text.append("⚡ ", style="warn") - text.append("Explore paths") - text.append(" x", style="muted") + text.append("Back") if selected: text.style = "brand" return text @@ -496,25 +567,30 @@ class TraceScreen(Screen): ) ) stamp = self._previous.timestamp.astimezone().strftime("%b %d %H:%M") - line.append(f" (previous · {stamp})", style="faint") + # On its own line (aligned under the route) so a long route never + # squeezes the stamp off the right edge. + line.append(f"\n (previous · {stamp})", style="faint") return line - line.append("unknown — press Enter to trace", style="muted") + if self._mode == "path": + line.append("none — compose a path to walk", style="muted") + else: + line.append("unknown — press Enter to trace", style="muted") return line def _planned_route(self) -> Optional[Text]: """The composed path as a route preview, or ``None`` without one. ``_path_spec`` is the literal wire spec — the whole walk, since the trace - protocol has no separate return-path field. A spec that reads the same - reversed is a symmetric boomerang (outbound hops, the target, those hops - mirrored): its second half is dimmed, reading as "this part isn't yours to - compose." Anything else was hand-composed hop by hop, so every hop renders - in full colour and only the automatic landing back on us stays faint. + protocol has no separate return-path field. In target mode the spec is the + symmetric boomerang (outbound hops, the target, those hops mirrored): its + second half is dimmed, reading as "this part isn't yours to compose". A path + walk was hand-composed hop by hop, so every hop renders in full colour and + only the automatic landing back on us stays faint. """ tokens = [h.strip() for h in self._path_spec.split(",") if h.strip()] if not tokens: return None - if len(tokens) % 2 and tokens == tokens[::-1]: + if self._mode == "target" and len(tokens) % 2 and tokens == tokens[::-1]: mid = len(tokens) // 2 # outbound hops + target = first half, inclusive outbound, return_leg = tokens[: mid + 1], tokens[mid + 1 :] else: @@ -548,7 +624,24 @@ class TraceScreen(Screen): text.append(hop, style=style) return text - def _summary(self, stats: TraceStats) -> Text: + def _displayed_hop_count(self, current: Optional[TraceResult]) -> Optional[int]: + """How many nodes the displayed route passes through, endpoints excluded. + + Follows the route line's precedence — the live route, else the planned spec, + else the stored previous trace — counting exactly the items drawn between the + two ``us`` endpoints (a walked trace's final hash-less hop *is* us, so it + doesn't count). + """ + if current is not None: + return sum(1 for h in current.hops if h.node) + tokens = [h for h in self._path_spec.split(",") if h.strip()] + if tokens: + return len(tokens) + if self._previous is not None and self._previous.hops: + return sum(1 for h in self._previous.hops if h.node) + return None + + def _summary(self, stats: TraceStats, current: Optional[TraceResult]) -> Text: """The session's aggregates plus the current path, label-aligned.""" snr = stats.median_min_snr snr_text = ( @@ -564,8 +657,13 @@ class TraceScreen(Screen): ) if self._path_spec: summary.append(self._path_spec, style="brand") + elif self._mode == "path": + summary.append("none — compose a path first", style="muted") else: summary.append("auto (device-routed)", style="muted") + hops = self._displayed_hop_count(current) + if hops is not None: + summary.append(f" · {hops} hop{'s' if hops != 1 else ''}", style="muted") return summary def _hops_table(self, stats: TraceStats, hash_bytes: Optional[int]) -> Table: @@ -592,12 +690,14 @@ class TraceScreen(Screen): rows: list[RenderableType] = [] if self._running: spin = self._spinner.text() - spin.append(" tracing…", style="muted") + if self._progress is not None and self._progress[1] > 1: + spin.append(f" tracing {self._progress[0]}/{self._progress[1]}…", + style="muted") + else: + spin.append(" tracing…", style="muted") rows.append(spin) elif self._status: rows.append(Text(self._status, style="err")) - elif not self._traces: - rows.append(Text("press Enter to trace", style="muted")) for number, trace in zip(range(len(self._traces), 0, -1), reversed(self._traces)): rows.append(self._trace_row(number, trace)) return Group(*rows) @@ -637,19 +737,60 @@ def _collapse_trace_width(mode: int) -> int: async def open_trace(ctx: "AppContext", target: str) -> int: - """Open the live trace screen for ``target`` and run it until dismissed. + """Open the live *Trace target* screen for ``target`` and run it until dismissed. + + The symmetric feature: can I reach this node? Routes turn at the target and come + home over the mirrored hops; *Explore paths* ranks candidate outbound legs from + the observed evidence. + + Args: + ctx: The shared application context (must be running the interactive TUI surface). + target: The trace destination (contact name or key prefix). + + Returns: + The number of traces run while the screen was open. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + return await _open_session(ctx, target) + + +async def open_trace_path(ctx: "AppContext") -> int: + """Open the live *Trace path* screen and run it until dismissed. + + The hand-routed feature: how far can a route I build carry? There is no target — + the whole walk is composed hop by hop and only has to end within our earshot — so + there is no target picker, no device routing, and no scenario explorer. Traces + record under :data:`~meshterm.core.models.PATH_TRACE_TARGET`. + + Args: + ctx: The shared application context (must be running the interactive TUI surface). + + Returns: + The number of traces run while the screen was open. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + return await _open_session(ctx, None) + + +async def _open_session(ctx: "AppContext", target: Optional[str]) -> int: + """Wire and run one live trace session (both features share this plumbing). Wires the screen to the radio, the database, and the observed-topology services: each trace opens its own ``runs`` row and is recorded under it (same shape a scripted ``meshterm trace`` writes); the composer and scenario flows build a fresh :class:`~meshterm.services.topology.MeshTopology` from stored evidence on each open, so suggestions always reflect the latest received traffic. Nothing transmits until - the user asks — the screen opens idle, its route line seeded from the target's most - recent stored trace. + the user asks — the screen opens idle, its route line seeded from the most recent + stored trace. Args: ctx: The shared application context (must be running the interactive TUI surface). - target: The trace destination (contact name or key prefix). + target: The trace destination (contact name or key prefix), or ``None`` for a + target-less path walk. Returns: The number of traces run while the screen was open. @@ -675,6 +816,9 @@ async def open_trace(ctx: "AppContext", target: str) -> int: raise RuntimeError("the live trace screen is only available in the menu") session = ctx.ui.session + mode = "target" if target is not None else "path" + record_target = target if target is not None else PATH_TRACE_TARGET + device = await ctx.device() contacts = await device.get_contacts() resolve = trace_runner.make_node_resolver(contacts) @@ -693,26 +837,35 @@ async def open_trace(ctx: "AppContext", target: str) -> int: width_bytes = 1 device_width = width_bytes # remembered so the width dialog can mark the default + # How many traces one Trace commit runs. Session state like the width; the + # screen's *Sample count* action changes it (see pick_samples). + sample_count = 1 + # The target as an addressable hash: a known contact's full key, or the typed hex # prefix itself. A non-hex unknown target can still be traced device-routed, but - # composing/exploring needs a destination hash to pin the path on. - needle = target.casefold() - target_contact = next( - ( - c - for c in contacts - if c.name.casefold() == needle - or ((c.public_key or "").lower().startswith(target.lower())) - ), - None, - ) - raw_hash = ( - (target_contact.public_key or target_contact.key_prefix) - if target_contact - else target - ).lower().removeprefix("0x") - target_hash = raw_hash if _is_hex(raw_hash) else None - target_label = target_contact.name if target_contact else target + # composing/exploring needs a destination hash to pin the path on. A path walk has + # no target at all — every target_* stays None and the composer runs hand-routed. + target_contact: Optional[Contact] = None + target_hash: Optional[str] = None + target_label: Optional[str] = None + if target is not None: + needle = target.casefold() + target_contact = next( + ( + c + for c in contacts + if c.name.casefold() == needle + or ((c.public_key or "").lower().startswith(target.lower())) + ), + None, + ) + raw_hash = ( + (target_contact.public_key or target_contact.key_prefix) + if target_contact + else target + ).lower().removeprefix("0x") + target_hash = raw_hash if _is_hex(raw_hash) else None + target_label = target_contact.name if target_contact else target def fresh_topology() -> MeshTopology: """Build the evidence graph from everything currently stored + on the device.""" @@ -725,7 +878,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int: ) async def unaddressable() -> None: - """Explain why path features need a resolvable target, in a small dialog.""" + """Explain why target-mode path features need a resolvable target.""" await session.message_dialog( Text( f"{target!r} isn't a known contact or a hex key prefix, so a forced " @@ -850,21 +1003,30 @@ async def open_trace(ctx: "AppContext", target: str) -> int: fetch, rebuilds the topology with the new evidence, and reopens the composer exactly where the user stood (same hops, refreshed suggestions). """ - if target_hash is None: + if mode == "target" and target_hash is None: await unaddressable() return None topo = fresh_topology() - target_id = topo.canonical(target_hash) or target_hash[:12] - # Re-seed from the current spec. A spec that reads the same reversed is the - # symmetric boomerang (hops, target, mirror): seed just the outbound hops and - # let the composer regenerate the rest. Anything else was hand-composed, so - # reopen in asymmetric mode with every hop of the walk editable. Tokens no - # contact matches are kept verbatim rather than dropped — an adopted route - # must survive a reopen even where the evidence graph is blind. + target_id = ( + (topo.canonical(target_hash) or target_hash[:12]) + if target_hash is not None + else None + ) + # Re-seed from the current spec. A target-mode spec is the symmetric boomerang + # (hops, target, mirror): seed just the outbound hops and let the composer + # regenerate the rest. A path walk is its spec verbatim. Tokens no contact + # matches are kept as they are rather than dropped — an adopted route must + # survive a reopen even where the evidence graph is blind. tokens = [p.strip() for p in current.split(",") if p.strip()] ids = [topo.canonical(h) or h for h in tokens] - symmetric = not ids or (len(ids) % 2 == 1 and ids == ids[::-1]) - seed = ids[: len(ids) // 2] if symmetric else ids + if mode == "path": + seed = ids + elif ids and len(ids) % 2 == 1 and ids == ids[::-1]: + seed = ids[: len(ids) // 2] + elif target_id in ids: + seed = ids[: ids.index(target_id)] # a stale hand walk: keep the outbound + else: + seed = ids while True: # Nodes whose neighbour table can be asked for: repeater contacts with a # public key to log in against (our own node has nothing new to tell us). @@ -876,22 +1038,21 @@ async def open_trace(ctx: "AppContext", target: str) -> int: if cid is not None and cid != topo.self_id: fetchable[cid] = c screen = PathComposerScreen( + device_label=device_label, + device_hash=device_hash, + topology=topo, + width_bytes=width_bytes, target_id=target_id, target_hash=target_hash, target_label=target_label, - device_label=device_label, - topology=topo, - width_bytes=width_bytes, hops=seed, fetch_nodes=frozenset(fetchable), - symmetric=symmetric, ) result = await session.run_screen(screen) if result is CANCEL: return None if isinstance(result, FetchNeighbours): seed = screen.hops # resume mid-thought after the fetch - symmetric = screen.symmetric repeater = fetchable.get(result.node) if repeater is not None and await fetch_neighbours_via( repeater, result.node @@ -903,12 +1064,12 @@ async def open_trace(ctx: "AppContext", target: str) -> int: async def pick_width(current: str) -> Optional[str]: """Float the path-hash width picker and re-render the standing spec to match. - Each row previews the target's key with the addressed slice lit at that - width, so the choice reads as "this much of every key goes on the air". The - chosen width shapes every spec the composer/explorer emit afterwards; a - standing forced path is re-rendered immediately — hops are widened back - through their canonical hashes where known, then collapsed uniformly (a hop - only ever known narrower keeps the whole spec at what it can honour). + Each row previews a key with the addressed slice lit at that width, so the + choice reads as "this much of every key goes on the air". The chosen width + shapes every spec the composer/explorer emit afterwards; a standing forced + path is re-rendered immediately — hops are widened back through their + canonical hashes where known, then collapsed uniformly (a hop only ever known + narrower keeps the whole spec at what it can honour). Returns: The re-rendered spec, or ``None`` when cancelled, unchanged, or there is @@ -929,7 +1090,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int: SelectScreen( "path width", items, - prompt="Forced hops are addressed by this many leading bytes of each key.", + prompt="Forced hops are addressed by this many leading key bytes.", default=width_bytes, footer_hint="↑↓ · Enter set · Esc keep", filterable=False, @@ -947,16 +1108,54 @@ async def open_trace(ctx: "AppContext", target: str) -> int: width = collapse_width(*full, ceiling=width_bytes) return ",".join(f[: width * 2] for f in full) + async def pick_samples(current: str) -> Optional[str]: + """Float the sample-count picker: how many traces one Trace action runs. + + Multi-trace runs are paced by the configured cooldown between transmissions, + so choosing 8 is a deliberate sampling session, never a burst. Always resolves + ``None`` — the count is session state the screen reads back, not a spec. + """ + nonlocal sample_count + pace = ctx.settings.trace_cooldown_s + items: list = [] + for n in SAMPLE_CHOICES: + title = Text(f"{n} trace{'s' if n > 1 else ' '}") + if n == 1: + title.append(" · a single transmission", style="muted") + items.append(Choice(title=title, value=n)) + picked = await session.run_screen( + SelectScreen( + "sample count", + items, + prompt=f"One Trace action runs this many traces, {pace:g} s apart.", + default=sample_count, + footer_hint="↑↓ · Enter set · Esc keep", + filterable=False, + wrap=False, + ) + ) + if picked is not CANCEL and picked is not None: + sample_count = int(picked) + return None + def scenario_title(scenario: PathScenario, topo: MeshTopology) -> Text: - """One scenario as a select row: source, route, and its observed evidence.""" - styles = {"device": "accent", "direct": "muted", "observed": "brand"} - text = Text(scenario.label, style=styles.get(scenario.source, "")) - if scenario.hops: - text.append(" via ", style="muted") - names = [topo.display_name(h) or h for h in scenario.hops] - text.append(" → ".join(names)) + """One scenario as a select row: the route itself, and its observed evidence. + + Observed candidates *are* their hop sequence, so the row leads with the route + and no label; the device route and the direct shot keep their short labels — + that provenance is the point of offering them. + """ + names = [topo.display_name(h) or h for h in scenario.hops] + if scenario.source == "observed": + text = Text(" → ".join(names), style="brand") else: - text.append(" no repeaters", style="muted") + styles = {"device": "accent", "direct": "muted"} + text = Text(scenario.label, style=styles.get(scenario.source, "")) + if names: + text.append(" via ", style="muted") + text.append(" → ".join(names)) + else: + text.append(" no repeaters", style="muted") if scenario.weakest_snr is not None: text.append(" · weakest ", style="muted") text.append(f"{scenario.weakest_snr:+.1f} dB", style=snr_style(scenario.weakest_snr)) @@ -992,8 +1191,8 @@ async def open_trace(ctx: "AppContext", target: str) -> int: One ``runs`` row spans the sweep; each trace and each candidate aggregate is recorded under it, so an aborted probe still keeps everything it measured. - Each candidate gets exactly one transmission (the screen-wide rule: repeaters - can blacklist nodes that burst traffic); ``None`` when aborted. + Each candidate gets exactly one transmission (probing is comparison, not + sampling); ``None`` when aborted. """ run_id = ctx.repo.start_run( "trace", @@ -1124,7 +1323,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int: SelectScreen( f"reach {target_label} · scenarios", items, - prompt="Choose the outbound leg — the return is always those hops mirrored.", + prompt="Choose the outbound leg — the return mirrors it.", footer_hint="↑↓ move · Enter adopt/probe · Esc back", wrap=False, ) @@ -1166,17 +1365,16 @@ async def open_trace(ctx: "AppContext", target: str) -> int: async def trace_once(path_spec: str, on_trace: Callable[[TraceResult], None]) -> None: """Run one persisted trace: one transmission, its own ``runs`` row. - A single transmission per keypress is deliberate — repeaters penalize (and - can blacklist) nodes that burst traffic — so repeat sampling is left to the - human, and the screen aggregates whatever lands. + The screen calls this once per sample; multi-trace runs are paced there, so + the repeaters never see a burst regardless of the chosen count. """ path = trace_runner.parse_trace_path(path_spec, contacts) if path_spec.strip() else None - params: dict[str, Any] = {"target": target} + params: dict[str, Any] = {"target": record_target} if path: params["path"] = path run_id = ctx.repo.start_run("trace", params, ctx.profile_name) try: - result = await device.run_trace(target, path=path) + result = await device.run_trace(record_target, path=path) except BaseException as exc: # A cancelled or failed trace still closes its run row, so no ``running`` # orphan is left behind. @@ -1188,7 +1386,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int: run_id, "ok", { - "target": target, + "target": record_target, "success": result.success, "min_snr": result.min_snr, "rtt_ms": result.round_trip_ms, @@ -1196,17 +1394,21 @@ async def open_trace(ctx: "AppContext", target: str) -> int: ) screen = TraceScreen( - target, + record_target, + mode=mode, device_label=device_label, device_hash=device_hash, resolve=resolve, session=session, trace=trace_once, compose_path=compose, - explore=explore, + explore=explore if mode == "target" else None, pick_width=pick_width, + pick_samples=pick_samples, width_bytes=lambda: width_bytes, - previous=ctx.repo.latest_trace(target), + sample_count=lambda: sample_count, + pace_s=ctx.settings.trace_cooldown_s, + previous=ctx.repo.latest_trace(record_target), ) try: await session.run_screen(screen) diff --git a/meshterm/ui/widgets.py b/meshterm/ui/widgets.py index bb10a4e..41d0035 100644 --- a/meshterm/ui/widgets.py +++ b/meshterm/ui/widgets.py @@ -70,11 +70,6 @@ def channel_glyph(name: str, secret: Optional[bytes]) -> str: return "🌐" return "🔒" -#: Non-breaking space, used in the route line to keep ``name (hash)`` and a node's -#: trailing arrow on the same line so wraps only ever land *after* an arrow. -_NBSP = " " - - def _identity(label: Optional[str]) -> Optional[str]: """Default node resolver: leave labels untouched.""" return label @@ -300,14 +295,14 @@ def _route_node_text( device_hash: Our own device's key/hash, shown alongside its label when known. Returns: - A styled :class:`Text` for the node. A non-breaking space joins name and hash - so they never split across a line wrap. + A styled :class:`Text` for the node, joined with plain spaces throughout — + never non-breaking ones, which prompt_toolkit renders as underscores. """ if not label or label == device_label: text = Text(device_label, style="accent") shown = _shorten_hash(device_hash, hash_bytes) if device_hash else "" if shown: - text.append(f"{_NBSP}(", style="muted") # nbsp keeps "name (hash)" together + text.append(" (", style="muted") text.append(shown, style="muted") text.append(")", style="muted") return text @@ -316,7 +311,7 @@ def _route_node_text( if not named or named == label: return Text(shown, style="brand") # unknown node: hash only text = Text(named, style="brand") - text.append(f"{_NBSP}(", style="muted") # nbsp keeps "name (hash)" together + text.append(" (", style="muted") text.append(shown, style="muted") text.append(")", style="muted") return text @@ -333,8 +328,7 @@ def _route_text( Shows the path the trace actually walked — the forced path, or the route the device resolved when auto-routing — with each node annotated by its hash at the command's path-hash width, joined by muted arrows, e.g. - ``Me (a1b2) → Alice (3d63) → Bob (f2a1) → Me (a1b2)``. The line wraps after an arrow - when it is too long for the panel, so each continuation line starts on a node. + ``Me (a1b2) → Alice (3d63) → Bob (f2a1) → Me (a1b2)``. Args: result: The trace whose route to display. @@ -353,9 +347,7 @@ def _route_text( text = Text() for i, node in enumerate(nodes): if i: - # A non-breaking space glues the arrow to the preceding node; the trailing - # regular space is the only wrap point, so wrapping lands *after* the arrow. - text.append(f"{_NBSP}→ ", style="muted") + text.append(" → ", style="muted") text.append_text( _route_node_text(node, device_label, resolve, hash_bytes, device_hash) ) diff --git a/tests/test_live_screens.py b/tests/test_live_screens.py index 3be5336..9eea111 100644 --- a/tests/test_live_screens.py +++ b/tests/test_live_screens.py @@ -72,7 +72,8 @@ def test_snr_bar_clamps_out_of_range_readings() -> None: def _trace_screen( - trace=None, compose_path=None, explore=None, pick_width=None, previous=None + trace=None, compose_path=None, explore=None, pick_width=None, pick_samples=None, + previous=None, mode="target", samples=1, ) -> tuple[TraceScreen, _FakeSession]: session = _FakeSession() @@ -83,26 +84,30 @@ def _trace_screen( return current screen = TraceScreen( - "Alice", + "Alice" if mode == "target" else "(path)", + mode=mode, device_label="Us", device_hash="aabb" + "00" * 30, resolve=lambda label: label, session=session, trace=trace or default_trace, compose_path=compose_path or default_flow, - explore=explore or default_flow, + explore=(explore or default_flow) if mode == "target" else None, pick_width=pick_width or default_flow, + pick_samples=pick_samples or default_flow, width_bytes=lambda: 2, + sample_count=lambda: samples, + pace_s=0.0, # tests never sleep; pacing is asserted through the statuses previous=previous, ) return screen, session async def test_trace_screen_one_trace_per_enter_accumulates() -> None: - """Each Enter transmits exactly one trace; the session aggregates what landed. + """At the default sample count, each Enter transmits exactly one trace. - Single-transmission is the screen's blacklist-avoidance rule: repeat sampling is - the human's call, so three keypresses mean three traces and a three-sample median. + Repeat sampling stays a human decision unless a bigger sample count is chosen + explicitly, so three keypresses mean three traces and a three-sample median. """ screen, _ = _trace_screen() for _ in range(3): @@ -115,6 +120,48 @@ async def test_trace_screen_one_trace_per_enter_accumulates() -> None: assert "burst" not in body.lower() # no burst configuration is offered anywhere +async def test_trace_screen_runs_the_chosen_sample_count() -> None: + """One Enter runs the whole chosen sample count, every trace recorded.""" + ran: list[str] = [] + + async def trace(path_spec, on_trace): # noqa: ANN001 + ran.append(path_spec) + on_trace(_trace(5.0)) + + screen, session = _trace_screen(trace=trace, samples=3) + screen.start_trace() + await screen._worker + assert len(ran) == 3 + assert len(screen._traces) == 3 + assert session.stack == [] # the dialog was popped with the run + + +async def test_trace_screen_multi_trace_reports_progress_and_abort_keeps_landed() -> None: + """A multi-trace run counts itself off; aborting keeps what already landed.""" + release = asyncio.Event() + ran = 0 + + async def trace(path_spec, on_trace): # noqa: ANN001 + nonlocal ran + ran += 1 + on_trace(_trace(5.0)) + if ran == 2: + await release.wait() # hold the run mid-flight on the second trace + + screen, session = _trace_screen(trace=trace, samples=5) + screen.start_trace() + await asyncio.sleep(0) + dialog = session.stack[0] + assert "2/5" in dialog.status # the dialog counts the run off + assert "2/5" in screen.footer_hint + assert "2/5" in _plain(screen.render_body(100)) # the log spinner row too + screen.cancel() + with pytest.raises(asyncio.CancelledError): + await screen._worker + assert len(screen._traces) == 2 # already-recorded traces are kept + assert session.stack == [] + + 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) @@ -128,6 +175,15 @@ async def test_trace_screen_seeds_route_from_previous_trace() -> None: assert "(previous" not in body +async def test_previous_stamp_sits_on_its_own_line() -> None: + """The (previous · …) marker renders under the route, never squeezed beside it.""" + old = _trace(4.0) + old.timestamp = utcnow() - timedelta(hours=3) + screen, _ = _trace_screen(previous=old) + stamp_line = next(l for l in screen.render_body(100) if "(previous" in l) + assert "→" not in stamp_line # the route stays on the line above + + async def test_trace_screen_only_one_trace_at_a_time() -> None: """Enter during an in-flight trace is a no-op; the running flag gates re-entry.""" started = 0 @@ -161,7 +217,7 @@ async def test_trace_screen_failure_reads_inline() -> None: async def test_trace_screen_composer_updates_the_spec() -> None: - """`p` opens the injected composer flow; its result becomes the next trace's path.""" + """Committing Compose path runs the flow; its result becomes the next trace's path.""" asked: list[str] = [] async def compose(current: str): # noqa: ANN001 @@ -169,7 +225,9 @@ async def test_trace_screen_composer_updates_the_spec() -> None: return "3d,f2,3d" # one forced hop: outbound, target, then the mirrored return screen, _ = _trace_screen(compose_path=compose) - screen.handle("text", "p") + screen.handle("down") # Trace → Back… + screen.handle("down") # …wrapping onto Compose path, the first row + screen.handle("enter") await asyncio.sleep(0) assert asked == [""] assert screen._path_spec == "3d,f2,3d" @@ -181,13 +239,15 @@ async def test_trace_screen_composer_updates_the_spec() -> None: async def test_trace_screen_explore_adopts_a_scenario_path() -> None: - """`x` runs the injected explore flow; adopting a path sets the spec, None keeps it.""" + """Committing Explore paths runs the flow; adopting sets the spec, None keeps it.""" async def adopt(current: str): # noqa: ANN001 return "3d63,f2c2" screen, _ = _trace_screen(explore=adopt) - screen.handle("text", "x") + for _ in range(3): + screen.handle("up") # Trace → Sample count → Path width → Explore paths + screen.handle("enter") await asyncio.sleep(0) assert screen._path_spec == "3d63,f2c2" @@ -195,7 +255,7 @@ async def test_trace_screen_explore_adopts_a_scenario_path() -> None: return None screen._explore = keep - screen.handle("text", "x") + screen.handle("enter") # the cursor is still on Explore paths await asyncio.sleep(0) assert screen._path_spec == "3d63,f2c2" # None leaves the spec untouched @@ -210,35 +270,64 @@ async def test_trace_screen_action_cursor_commits_the_selected_row() -> None: screen, _ = _trace_screen(pick_width=width_flow) body = _plain(screen.render_body(100)) - assert "Trace — one transmission" in body - assert "Path width — 2 bytes per hop" in body - assert "Compose path" in body and "Explore paths" in body - screen.handle("down") # Trace → Path width + # The menu order the actions read in: build first, tune, then transmit, then out. + labels = ["Compose path", "Explore paths", "Path width — 2 bytes per hop", + "Sample count — 1 trace", "Trace — one transmission", "Back"] + positions = [body.index(label) for label in labels] + assert positions == sorted(positions) + screen.handle("up") # Trace → Sample count + screen.handle("up") # → Path width screen.handle("enter") await asyncio.sleep(0) assert opened == ["width:"] assert screen._path_spec == "3d63,f2c2,3d63" -async def test_trace_screen_w_hotkey_opens_the_width_dialog() -> None: - """`w` floats the width flow directly; a cancelled flow leaves the spec alone.""" +async def test_trace_screen_sample_count_row_opens_its_dialog() -> None: + """Committing Sample count floats the flow; its None resolution keeps the spec.""" opened: list[str] = [] - async def width_flow(current): # noqa: ANN001 + async def samples_flow(current): # noqa: ANN001 opened.append(current) - return None # cancelled: the spec must stay untouched + return None - screen, _ = _trace_screen(pick_width=width_flow) + screen, _ = _trace_screen(pick_samples=samples_flow) screen._path_spec = "3d,f2,3d" - screen.handle("text", "w") + screen.handle("up") # Trace → Sample count + screen.handle("enter") await asyncio.sleep(0) assert opened == ["3d,f2,3d"] - assert screen._path_spec == "3d,f2,3d" - assert screen._index == 0 # the cursor stays on Trace: plain Enter still traces + assert screen._path_spec == "3d,f2,3d" # the count is not a spec: nothing changes + + +async def test_trace_screen_hotkeys_are_retired() -> None: + """The old w/p/x shortcuts are gone: typing must not float any flow.""" + opened: list[str] = [] + + async def flow(current): # noqa: ANN001 + opened.append(current) + return None + + screen, _ = _trace_screen(compose_path=flow, explore=flow, pick_width=flow, + pick_samples=flow) + for key in ("p", "x", "w", "s"): + screen.handle("text", key) + await asyncio.sleep(0) + assert opened == [] + assert not screen._running + + +async def test_trace_screen_back_row_resolves_like_escape() -> None: + """The Back row leaves the screen exactly as Esc does.""" + screen, _ = _trace_screen() + screen.future = asyncio.get_running_loop().create_future() + screen.handle("down") # Trace → Back + screen.handle("enter") + assert screen.future.result() is None def test_planned_route_dims_only_the_mirrored_return_leg() -> None: - """A palindromic (symmetric) spec dims its second half; a hand-composed one doesn't.""" + """A palindromic target-mode spec dims its second half; hand walks never dim.""" screen, _ = _trace_screen() def faint_cells(text) -> int: # noqa: ANN001 @@ -249,11 +338,49 @@ def test_planned_route_dims_only_the_mirrored_return_leg() -> None: screen._path_spec = "3d,f2,3d" # symmetric boomerang: the mirror is dimmed symmetric = screen._planned_route() assert symmetric.plain == "Us → 3d → f2 → 3d → Us" - screen._path_spec = "3d,f2,27" # asymmetric walk: every hop is the user's + screen._path_spec = "3d,f2,27" # a stale hand walk: every hop is the user's custom = screen._planned_route() assert custom.plain == "Us → 3d → f2 → 27 → Us" assert faint_cells(symmetric) > faint_cells(custom) + # In path mode even a there-and-back-the-same-way walk is fully hand-composed, + # so a palindrome must NOT read as "not yours to compose". + walk, _ = _trace_screen(mode="path") + walk._path_spec = "3d,f2,3d" + assert faint_cells(walk._planned_route()) == faint_cells(custom) + + +def test_summary_appends_the_displayed_hop_count() -> None: + """The path row ends with how many nodes the displayed route passes through.""" + screen, _ = _trace_screen() + screen._path_spec = "3d,f2,3d" + assert "· 3 hops" in _plain(screen.render_body(100)) + screen._on_trace(_trace(5.0, 2.0)) # a live 2-hop route now outranks the plan + assert "· 2 hops" in _plain(screen.render_body(100)) + + +def test_trace_log_section_hidden_until_there_is_something_to_log() -> None: + """Idle with no traces, the Traces heading (and its old hint) don't render.""" + screen, _ = _trace_screen() + assert "Traces" not in _plain(screen.render_body(100)) + + +async def test_trace_screen_path_mode_gates_trace_and_drops_explore() -> None: + """Path mode: no Explore row, and Trace stays inert until a path exists.""" + screen, _ = _trace_screen(mode="path") + body = _plain(screen.render_body(100)) + assert "Explore paths" not in body + assert "Trace — compose a path first" in body + assert "none — compose a path" in body + screen.handle("enter") # the cursor opens on Trace, but there is nothing to walk + assert not screen._running and screen._worker is None + screen._path_spec = "3d,f2" + assert "Trace — one transmission" in _plain(screen.render_body(100)) + screen.handle("enter") + assert screen._running + await screen._worker + assert len(screen._traces) == 1 + async def test_trace_screen_opens_idle_until_enter() -> None: """Selecting a target must never transmit by itself: nothing flies until Enter.""" diff --git a/tests/test_repository.py b/tests/test_repository.py index 72d54ea..87d99bb 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -7,7 +7,7 @@ from pathlib import Path import pytest -from meshterm.core.models import Hop, TraceResult +from meshterm.core.models import PATH_TRACE_TARGET, Hop, TraceResult from meshterm.persistence.repository import Repository @@ -167,6 +167,18 @@ def test_traced_targets_empty_db(repo: Repository) -> None: assert repo.traced_targets() == [] +def test_traced_targets_excludes_path_walks(repo: Repository) -> None: + """Path walks record under the ``(path)`` sentinel and never become pickable targets.""" + run = repo.start_run("trace", {}) + repo.record_trace(run, _make_trace("Alice", ("r", 1.0))) + repo.record_trace(run, _make_trace(PATH_TRACE_TARGET, ("r", 2.0))) + + assert repo.traced_targets() == ["Alice"] + # …but the walks still seed the path screen's previous-route line. + latest = repo.latest_trace(PATH_TRACE_TARGET) + assert latest is not None and latest.target == PATH_TRACE_TARGET + + # -- latest_trace edge cases ----------------------------------------------- diff --git a/tests/test_skeleton.py b/tests/test_skeleton.py index b910da0..9c81eba 100644 --- a/tests/test_skeleton.py +++ b/tests/test_skeleton.py @@ -192,6 +192,22 @@ def test_route_text_unknown_node_shows_hash_only() -> None: assert "aa11bb" not in plain +def test_route_text_contains_no_non_breaking_spaces() -> None: + """prompt_toolkit draws U+00A0 as an underscore, so route text must never emit one.""" + from meshterm.core.models import Hop + from meshterm.ui.widgets import _route_text + + result = TraceResult( + target="x", + success=True, + hops=[Hop(0, "3d63", 12.0), Hop(1, None, 12.0)], + path_hash_bytes=2, + ) + plain = _route_text(result, "Me", device_hash="a1b2" + "00" * 30).plain + assert "\xa0" not in plain + assert plain == "Me (a1b2) → 3d63 → Me (a1b2)" + + def test_parse_trace_path_rejects_unknown_token() -> None: """A token that is neither a known contact nor valid hex is an error.""" with pytest.raises(ValueError): diff --git a/tests/test_topology.py b/tests/test_topology.py index 29d6be0..72ac872 100644 --- a/tests/test_topology.py +++ b/tests/test_topology.py @@ -290,17 +290,25 @@ async def test_probe_paths_defaults_to_one_trace_per_candidate() -> None: # --- path composer --------------------------------------------------------------------- -def _composer(topo, hops=None, fetch_nodes=frozenset(), symmetric=True): # noqa: ANN001 +def _composer(topo, hops=None, fetch_nodes=frozenset(), target=True): # noqa: ANN001 + """A composer over ``topo`` — pinned on Far (target mode) or target-less.""" + pinned = ( + dict( + target_id="f2c24f54551e", + target_hash="f2c24f54551e" + "0" * 52, + target_label="Far", + ) + if target + else {} + ) return PathComposerScreen( - target_id="f2c24f54551e", - target_hash="f2c24f54551e" + "0" * 52, - target_label="Far", device_label="Us", + device_hash=US + "0" * 52, topology=topo, width_bytes=1, hops=list(hops or []), fetch_nodes=fetch_nodes, - symmetric=symmetric, + **pinned, ) @@ -363,7 +371,6 @@ async def test_composer_commits_spec_auto_and_cancel() -> None: screen.future = asyncio.get_running_loop().create_future() screen.handle("end") # jump to the last action row (Cancel)… screen.handle("up") # …past Auto… - screen.handle("up") # …and the ⇄ mode toggle… screen.handle("up") # …up to "Use this path" screen.handle("enter") assert screen.future.result() == "3d,f2,3d" @@ -381,32 +388,31 @@ async def test_composer_commits_spec_auto_and_cancel() -> None: assert cancelled.future.result() is CANCEL -def test_composer_mode_toggle_materializes_and_regenerates_the_mirror() -> None: - """⇄ to asymmetric turns the shown mirror into editable hops (route unchanged); - ⇄ back keeps what precedes the target as the outbound leg.""" - walks = [_traced(("3d", 12.0), ("f2", -5.0), ("3d", -5.5), (None, 12.0))] - screen = _composer(_topo(trace_paths=walks), hops=["3d63c6429436"]) - spec_before = screen._spec() - screen.handle("end") # Cancel… - screen.handle("up") # …Auto… - screen.handle("up") # …the ⇄ mode row - screen.handle("enter") - assert not screen.symmetric - assert screen._hops == ["3d63c6429436", "f2c24f54551e", "3d63c6429436"] - assert screen._spec() == spec_before # the wire route is identical - screen.handle("enter") # the cursor re-anchors on ⇄, so Enter toggles back - assert screen.symmetric - assert screen._hops == ["3d63c6429436"] - assert screen._spec() == spec_before +def test_composer_path_mode_opens_us_to_us_without_auto() -> None: + """Target-less, the preview is just ``us → us`` and there is no Auto action. + + Only the final landing back on us dims — the hops (and the opening endpoint) + are all the user's to compose, and with no destination there is nothing for + the device to route to. + """ + screen = _composer(_topo(), target=False) + assert screen.title == "compose path" # no target to name + preview = screen._route_preview() + assert preview.plain == "Us (aa) → Us (aa)" # endpoints carry our hash + faint = sum( + span.end - span.start for span in preview.spans if "faint" in str(span.style) + ) + assert 0 < faint < len(preview.plain) # some dimming (the landing), never all + assert "Auto" not in _rows_plain(screen) -async def test_composer_asymmetric_suggests_the_target_and_commits_the_walk_verbatim() -> None: - """Asymmetric mode routes *through* the target and commits exactly the composed hops.""" +async def test_composer_path_mode_suggests_through_anything_and_commits_verbatim() -> None: + """A path walk routes *through* any node and commits exactly the composed hops.""" import asyncio walks = [_traced(("3d", 12.0), ("f2", -5.0), ("3d", -5.5), (None, 12.0))] - screen = _composer(_topo(trace_paths=walks), hops=["3d63c6429436"], symmetric=False) - # The tail (Hub) can hear the target — asymmetric mode must offer it as a hop. + screen = _composer(_topo(trace_paths=walks), hops=["3d63c6429436"], target=False) + # No pinned target: the tail (Hub) hears Far, so Far is a plain hop suggestion. assert any(s.node == "f2c24f54551e" for s in screen._suggestions()) screen._hops.append("f2c24f54551e") # A return leg may legitimately reuse an outbound repeater (only the tail is barred). @@ -416,19 +422,17 @@ async def test_composer_asymmetric_suggests_the_target_and_commits_the_walk_verb loop = asyncio.get_running_loop() screen.future = loop.create_future() - screen.handle("end") - for _ in range(3): - screen.handle("up") # Cancel → Auto → ⇄ → Use this path + screen.handle("end") # Cancel… + screen.handle("up") # …up to "Use this path" (no Auto row between them) screen.handle("enter") assert screen.future.result() == "3d,f2" # With no hops at all there is nothing to commit: Use must stay inert, because # resolving "" would masquerade as Auto (device-routed). - empty = _composer(_topo(trace_paths=walks), symmetric=False) + empty = _composer(_topo(trace_paths=walks), target=False) empty.future = loop.create_future() assert "(add a hop first)" in _rows_plain(empty) empty.handle("end") - for _ in range(3): - empty.handle("up") + empty.handle("up") empty.handle("enter") assert not empty.future.done()