Route out one way, home another: asymmetric paths, width picker, actions

Four trace-feature upgrades, all UX-preserving:

- Stored traces recover their path-hash width from the (uniform) hop
  hashes at rehydration, so a target's previous route renders short
  hashes instead of the device's full 64-char public key.
- The trace screen's verbs are selectable action rows now — Trace,
  Path width, Compose path, Explore paths. The cursor opens on Trace so
  plain Enter still just traces; p/x/w hotkeys stay on their rows, and
  the cursor only pins the view while it's actually being moved, so
  PgUp/PgDn still read a long trace log freely.
- A floating path-width dialog picks the 1/2/4/8-byte hop addressing
  (each row previews the target key with the addressed slice lit, the
  device's routing width marked). Changing it re-renders a standing
  forced path — hops widened back through their canonical hashes — and
  shapes every spec composed or explored afterwards.
- The composer's new ⇄ row toggles symmetric (return mirrors the
  outbound, the current behaviour) against asymmetric, where the whole
  walk is composed by hand — target routed *through*, outbound
  repeaters reusable, only the landing back on us implicit. Toggling
  materializes the mirror into editable hops (or regenerates it from
  what precedes the target), so the wire route never changes under the
  toggle; the planned-route preview dims only palindromic return legs.

Also: the target now carries its hash in the composer preview like
every other node, and the composer's footer hint reads briefer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Fable 5 committed 2026-07-10 23:07:40 -04:00
1 parent fbcf13baf5
commit 22b182000b
7 files changed
+569 -132

No files matched your search

+25
View File
@@ -37,6 +37,29 @@ ACTIVITY_WINDOW = timedelta(hours=2)
ACTIVITY_BUCKETS = 24
def _hops_hash_bytes(hops: list[Hop]) -> Optional[int]:
"""Recover a stored trace's per-hop path-hash width from its hop hashes.
The ``traces`` table predates :attr:`~meshterm.core.models.TraceResult.
path_hash_bytes`, so the width isn't a column — but it doesn't need to be: a trace
reply names every hop at exactly the command's width, so the (uniform) length of
the stored hashes *is* the width. Without it, rehydrated traces would render node
hashes — and our own device's full public key — at absurd lengths.
Args:
hops: The rehydrated hops (the final hash-less hop, our own device, is skipped).
Returns:
The width in bytes, or ``None`` when there are no hashes or they disagree
(which stored trace replies never do; ``None`` falls back to full-width display).
"""
widths = {len(h.node) for h in hops if h.node}
if len(widths) != 1:
return None
width = widths.pop()
return width // 2 if width and width % 2 == 0 else None
@dataclass(slots=True)
class ChannelStats:
"""Aggregated message history for one channel conversation.
@@ -303,6 +326,7 @@ class Repository:
hops=hops,
round_trip_ms=row["round_trip_ms"],
tx_power=row["tx_power"],
path_hash_bytes=_hops_hash_bytes(hops),
timestamp=datetime.fromisoformat(row["created_at"]),
)
@@ -345,6 +369,7 @@ class Repository:
hops=hops,
round_trip_ms=row["round_trip_ms"],
tx_power=row["tx_power"],
path_hash_bytes=_hops_hash_bytes(hops),
timestamp=datetime.fromisoformat(row["created_at"]),
)
+30 -4
View File
@@ -72,14 +72,15 @@ def _is_hex(value: str) -> bool:
def render_forced_spec(hops: tuple[str, ...], target_hash: str, width_bytes: int) -> str:
"""Render a forced-path spec: outbound hops, the target, then the return leg.
"""Render a symmetric forced-path spec: outbound hops, the target, the mirror back.
The MeshCore trace protocol has no separate "return path" field — the whole
boomerang (out to the target, back to us) is one path the repeaters walk in
order, so the spec must spell out the return hops explicitly, not just the
outbound ones. The return leg is always the outbound hops in reverse (the
trace protocol doesn't support asymmetric routing), so it's appended here
rather than asked of the caller.
outbound ones. This renders the *symmetric* round trip — the return leg is the
outbound hops in reverse, appended here rather than asked of the caller. A
hand-composed route that comes home a different way is rendered verbatim by
:func:`render_custom_spec` instead.
Args:
hops: Intermediate repeaters, in order from us outward, excluding both
@@ -97,6 +98,31 @@ def render_forced_spec(hops: tuple[str, ...], target_hash: str, width_bytes: int
return ",".join(outbound)
def render_custom_spec(hops: tuple[str, ...], width_bytes: int) -> str:
"""Render a hand-composed (asymmetric) path spec exactly as given.
The counterpart of :func:`render_forced_spec` for a route the user spelled out in
full — out to the target and back home by whatever way they chose. Nothing is
appended or mirrored: the hops *are* the path, and the trace protocol happily
walks any sequence that ends within earshot of us. Only the shared width rules
apply.
Args:
hops: Every hop of the route in walk order, the target among them, excluding
our own device at both ends (the reply lands on us off the final hop).
width_bytes: Preferred per-hop path-hash width in bytes (1, 2, 4, or 8);
shrunk via :func:`collapse_width` when a hop's known hash is narrower.
Returns:
A comma-separated hex spec, e.g. ``"3d,f2,27"`` — or ``""`` with no hops
(there is no meaningful zero-hop custom route).
"""
if not hops:
return ""
width = collapse_width(*hops, ceiling=width_bytes)
return ",".join(h[: width * 2] for h in hops)
def collapse_width(*hashes: str, ceiling: int) -> int:
"""The widest representable per-hop hash width every given hash can honour.
+139 -44
View File
@@ -1,12 +1,11 @@
"""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`` with the target pinned as the final hop, since a
trace only replies when the destination's own hash ends the path — 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 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.
Interaction, following the reorder screen's cursor-over-rows-and-actions pattern:
@@ -19,15 +18,22 @@ 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 preserved).
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.
The return leg is never composed — it's always the outbound hops in reverse, so
there's nothing to choose there — but it isn't optional either: the trace protocol has
no separate return-path field, so the spec sent to the device must spell out the whole
boomerang, out and back. The preview resolves and shows that return leg dimmed,
right alongside the hops you're actually choosing, rather than leaving it to a footnote.
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.
"""
from __future__ import annotations
@@ -38,7 +44,7 @@ from typing import Optional
from rich.cells import cell_len
from rich.text import Text
from ..services.topology import MeshTopology, _is_hex, render_forced_spec
from ..services.topology import MeshTopology, _is_hex, render_custom_spec, render_forced_spec
from .theme import snr_style
from .tui.render import render_lines, render_to_ansi
from .tui.screen import Screen
@@ -54,6 +60,7 @@ _MAX_SUGGESTIONS = 12
#: Action-row sentinels (kept distinct from suggestion rows, which carry node ids).
_USE = "use"
_MODE = "mode"
_AUTO = "auto"
_CANCEL = "cancel"
@@ -79,16 +86,15 @@ class FetchNeighbours:
class PathComposerScreen(Screen):
"""Compose a forced outbound trace path step by step.
"""Compose a forced trace path step by step, symmetric or hand-routed.
Resolves with the finished spec (comma-separated hex: outbound hops, the target,
then the return leg), :data:`AUTO_SPEC` for device routing, or
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
:data:`~meshterm.ui.tui.screen.CANCEL`.
"""
footer_hint = (
"↑↓ move · Enter add/select · type to filter or enter hex · ⌫ remove · Esc cancel"
)
footer_hint = "↑↓ · Enter add · type to filter · ⌫ remove · Esc"
def __init__(
self,
@@ -101,23 +107,27 @@ class PathComposerScreen(Screen):
width_bytes: int,
hops: Optional[list[str]] = None,
fetch_nodes: frozenset[str] = frozenset(),
symmetric: bool = True,
) -> None:
"""Build the composer.
Args:
target_id: The target's canonical id (excluded from suggestions; the path
implicitly ends at it).
target_hash: The target's hex hash used as the spec's final hop (its full
key prefix; truncated to the spec width at commit).
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.
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.
hops: Canonical ids of already-composed intermediate hops (reopening the
dialog resumes where the user left off).
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}"
@@ -129,14 +139,20 @@ class PathComposerScreen(Screen):
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
@property
def hops(self) -> list[str]:
"""The composed intermediate hops so far (for re-seeding after a fetch)."""
"""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 -----------------------------------------------------------------
def _tail(self) -> str:
@@ -146,13 +162,20 @@ 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
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.
Returns:
The (possibly filtered) :class:`~meshterm.services.topology.HopSuggestion`
list for the path's tail, capped at :data:`_MAX_SUGGESTIONS`.
"""
exclude = frozenset(
{self._topology.self_id, self._target_id, *self._hops}
)
if self._symmetric:
exclude = frozenset({self._topology.self_id, self._target_id, *self._hops})
else:
exclude = frozenset({self._topology.self_id, self._tail()})
suggestions = self._topology.next_hops(self._tail(), exclude=exclude)
needle = self._entry.lower()
if needle:
@@ -182,17 +205,50 @@ 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))
rows.append(("action", _CANCEL))
return rows
def _spec(self) -> str:
"""Render the composed route as the forced-path spec, return leg included.
"""Render the composed route as the spec the trace command will send.
See :func:`~meshterm.services.topology.render_forced_spec` for the
width-collapsing and return-leg rules.
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.
render_custom_spec`) — empty until it has at least one hop.
"""
return render_forced_spec(tuple(self._hops), self._target_hash, self._width_bytes)
if self._symmetric:
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 ---------------------------------------------------------------
@@ -202,7 +258,10 @@ class PathComposerScreen(Screen):
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.
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.
Args:
node: The node's canonical id (or ``self_id`` for our own device).
@@ -211,6 +270,13 @@ class PathComposerScreen(Screen):
"""
if node == self._topology.self_id:
return Text(self._device_label, style="faint" if dim else "accent")
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)
@@ -221,18 +287,24 @@ class PathComposerScreen(Screen):
return Text(shown, style=style)
def _route_preview(self) -> Text:
"""The route under construction: outbound in full color, the mirrored
return leg resolved and dimmed right alongside it.
"""The route under construction, endpoints filled in automatically.
Symmetric mode shows the composed outbound in full color, 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.
"""
text = Text(self._device_label, style="accent")
for hop in self._hops:
text.append(" → ", style="muted")
text.append_text(self._node_text(hop))
text.append(" → ", style="muted")
text.append(self._target_label, style="ok")
for hop in reversed(self._hops):
text.append(" → ", style="faint")
text.append_text(self._node_text(hop, dim=True))
if self._symmetric:
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")
return text
@@ -253,6 +325,16 @@ 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":
@@ -270,17 +352,26 @@ class PathComposerScreen(Screen):
return text
if payload == _USE:
label = Text.assemble(("✓ ", "ok"), "Use this path")
label.append(f" ({self._spec()})", style="muted")
spec = self._spec()
# An asymmetric 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)."""
"""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.
"""
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
@@ -348,7 +439,11 @@ class PathComposerScreen(Screen):
elif kind == "fetch":
self.resolve(FetchNeighbours(node=str(payload)))
elif payload == _USE:
self.resolve(self._spec())
spec = self._spec()
if spec: # an empty asymmetric walk would masquerade as AUTO_SPEC
self.resolve(spec)
elif payload == _MODE:
self._toggle_mode()
elif payload == _AUTO:
self.resolve(AUTO_SPEC)
else:
+234 -79
View File
@@ -3,34 +3,43 @@
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
whatever you make it, and nothing transmits until you say so. Three verbs drive it:
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:
* **Enter** 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 — press Enter again and the screen keeps aggregating
every trace of the session into its medians.
* **p** opens the path composer (:mod:`~meshterm.ui.path_composer`): build the outbound
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. Only the outbound leg is
composed — the return is always those hops in reverse — but the spec sent to the
device spells out the whole boomerang, since the trace protocol has no separate
return-path field.
* **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.
* **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*
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.
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, per-hop
median SNR with quality bars, and the individual traces newest-first.
composed path, else the target's last 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
@@ -53,7 +62,7 @@ from .theme import snr_style
from .tui.render import render_lines, render_to_ansi
from .tui.screen import Screen
from .tui.spinner import Spinner
from .widgets import NodeResolver, _link_text, _route_text
from .widgets import NodeResolver, _link_text, _route_text, highlighted_hash
if TYPE_CHECKING:
from ..context import AppContext
@@ -179,16 +188,20 @@ class TracingDialog(Screen):
class TraceScreen(Screen):
"""A full-screen live trace session for one target — armed, but idle until told.
Keys: Enter runs one trace (a floating dialog with Abort rides on top while it
flies — one transmission per keypress, because repeaters can blacklist nodes that
burst traffic), ``p`` composes the forced path hop by hop from observed topology,
``x`` explores and probes ranked path scenarios, the usual scroll keys move the
view, and Esc backs out (cancelling any in-flight trace; already-recorded traces
are kept).
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).
"""
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,
@@ -200,9 +213,11 @@ class TraceScreen(Screen):
trace: TraceOnce,
compose_path: PathFlow,
explore: PathFlow,
pick_width: PathFlow,
width_bytes: Callable[[], int],
previous: Optional[TraceResult] = None,
) -> None:
"""Create the screen (nothing transmits until the user presses Enter).
"""Create the screen (nothing transmits until the user commits Trace).
Args:
target: The trace destination (contact name or key prefix).
@@ -216,6 +231,11 @@ class TraceScreen(Screen):
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).
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.
"""
@@ -229,6 +249,8 @@ class TraceScreen(Screen):
self._trace_once = trace
self._compose_path = compose_path
self._explore = explore
self._pick_width = pick_width
self._width_bytes = width_bytes
self._previous = previous
self._path_spec = ""
self._traces: list[TraceResult] = []
@@ -238,6 +260,8 @@ class TraceScreen(Screen):
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
self._pin_cursor = False # only pin the view while ↑/↓ are actually in use
# --- state -----------------------------------------------------------------
@@ -245,8 +269,8 @@ class TraceScreen(Screen):
def footer_hint(self) -> str: # type: ignore[override]
"""The footer keys, tracking whether a trace is in flight."""
if self._running:
return "tracing… · ↑↓ PgUp/PgDn scroll · Esc back"
return "Enter trace · p compose path · x explore paths · ↑↓ 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)."""
@@ -313,35 +337,60 @@ class TraceScreen(Screen):
# --- input -------------------------------------------------------------------
def handle(self, action: str, data: str = "") -> None:
"""Run a trace, compose/explore paths, scroll, or dismiss."""
"""Move the action cursor, commit the selected action, scroll, or dismiss.
↑/↓ belong to the action cursor (and pin the view to it); the body scrolls
with PgUp/PgDn/Home/End, each of which releases the pin so a long trace log
can be read without the cursor yanking the view back.
"""
if action == "enter":
self.start_trace()
self._commit_action()
elif action == "up":
self._index = (self._index - 1) % len(self._ACTIONS)
self._pin_cursor = True
elif action == "down":
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 == "up":
self.scroll_lines(-1)
elif action == "down":
self.scroll_lines(1)
elif action == "text" and data.lower() == "w":
self._open_flow(self._pick_width)
elif action == "pageup":
self._pin_cursor = False
self.scroll_pages(-1)
elif action in ("pagedown", "space"):
self._pin_cursor = False
self.scroll_pages(1)
elif action in ("home", "ctrl_home"):
self._pin_cursor = False
self.scroll_to_top()
elif action in ("end", "ctrl_end"):
self._pin_cursor = False
self.scroll_to_bottom()
elif action == "escape":
self.cancel()
self.resolve(None)
def _commit_action(self) -> None:
"""Run the action row under the cursor (the flows guard against re-entry)."""
key = self._ACTIONS[self._index]
if key == "trace":
self.start_trace()
elif key == "width":
self._open_flow(self._pick_width)
elif key == "compose":
self._open_flow(self._compose_path)
else:
self._open_flow(self._explore)
def _open_flow(self, flow: PathFlow) -> None:
"""Float a path-picking flow over the screen (one at a time, not mid-trace).
Both the composer and the scenario explorer resolve the same way: a new spec to
adopt (``""`` returns routing to the device), or ``None`` to leave the current
path untouched.
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.
Args:
flow: The dialog flow to run with the current spec.
@@ -364,27 +413,69 @@ class TraceScreen(Screen):
# --- rendering -----------------------------------------------------------------
def render_body(self, width: int) -> list[str]:
"""Render the route, aggregates, per-hop medians, and the trace log."""
lines = render_lines(Group(*self._sections()), width)
"""Render the route, aggregates, action list, per-hop medians, and the log.
The action rows are rendered line by line (not through one Rich group) so the
highlighted row's body line is known exactly — that is what :meth:`cursor_line`
pins while the user is navigating.
"""
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
)
lines.append("")
self._cursor: Optional[int] = None
for i, key in enumerate(self._ACTIONS):
selected = i == self._index
text = self._action_text(key, selected)
text.no_wrap = True
text.truncate(width, overflow="ellipsis")
if selected:
self._cursor = len(lines)
lines.append(render_to_ansi(text, width))
tail: list[RenderableType] = []
if stats.hop_snrs:
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())
lines.extend(render_lines(Group(*tail), width))
self._scroll_total = max(1, len(lines))
return lines
def _sections(self) -> list[RenderableType]:
"""Assemble the screen's stacked sections for the current state."""
stats = TraceStats.from_traces(self._target, self._traces)
current = next((t for t in reversed(self._traces) if t.success), None)
sections: list[RenderableType] = [
self._route_line(current),
Text(),
self._summary(stats),
]
if stats.hop_snrs:
hash_bytes = current.path_hash_bytes if current is not None else None
sections += [Text(), Text("Per-hop medians", style="accent")]
sections.append(self._hops_table(stats, hash_bytes))
sections += [Text(), Text("Traces", style="accent")]
sections.append(self._trace_log())
return sections
def cursor_line(self) -> Optional[int]:
"""The highlighted action row while ↑/↓ are in use; free scrolling otherwise.
Returning ``None`` between navigations matters: the frame force-keeps a
cursor line visible, which would otherwise stop PgDn ever scrolling the
action list off screen to read a long trace log.
"""
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."""
text = Text("❯ " if selected else " ", style="brand" if selected else "")
if key == "trace":
text.append("▶ ", style="ok")
text.append("Trace — one transmission")
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")
else:
text.append("⚡ ", style="warn")
text.append("Explore paths")
text.append(" x", style="muted")
if selected:
text.style = "brand"
return text
def _route_line(self, current: Optional[TraceResult]) -> Text:
"""The route: live when a reply has landed, else planned, else the stored one."""
@@ -413,17 +504,21 @@ class TraceScreen(Screen):
def _planned_route(self) -> Optional[Text]:
"""The composed path as a route preview, or ``None`` without one.
``_path_spec`` is the literal wire spec: outbound hops, the target, then the
return leg (those hops mirrored) — the trace protocol has no separate
return-path field, so the composer already baked the return leg in. The
preview splits it back at its midpoint (the target) to dim the return half,
reading as "this part isn't yours to compose."
``_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.
"""
tokens = [h.strip() for h in self._path_spec.split(",") if h.strip()]
if not tokens:
return None
mid = len(tokens) // 2 # outbound hops + target = first half, inclusive
outbound, return_leg = tokens[: mid + 1], tokens[mid + 1 :]
if 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:
outbound, return_leg = tokens, []
text = Text(self._device_label, style="accent")
for hop in outbound:
text.append(" → ", style="muted")
@@ -565,7 +660,13 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
from ..core.connection import DeviceAuthenticationError
from ..core.models import LOCAL_DEVICE_LABEL, Contact, NeighbourInfo
from ..services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths
from ..services.topology import MeshTopology, PathScenario, _is_hex, build_topology
from ..services.topology import (
MeshTopology,
PathScenario,
_is_hex,
build_topology,
collapse_width,
)
from .path_composer import FetchNeighbours, PathComposerScreen
from .surface import TuiUi
from .tui import CANCEL, Choice, SelectScreen, Separator
@@ -584,11 +685,13 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
# The per-hop width composed/scenario specs are emitted at: the region's routing
# width, collapsed to what a trace can encode. Unknown (old firmware) → 1 byte, the
# protocol default and what all stored evidence uses anyway.
# protocol default and what all stored evidence uses anyway. The user can override
# it for the session through the screen's *Path width* action (see pick_width).
try:
width_bytes = _collapse_trace_width(int(await device.get_path_hash_mode()))
except Exception: # noqa: BLE001 - optional read; the 1-byte default always works
width_bytes = 1
device_width = width_bytes # remembered so the width dialog can mark the default
# 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
@@ -752,15 +855,16 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
return None
topo = fresh_topology()
target_id = topo.canonical(target_hash) or target_hash[:12]
# Re-seed from the current spec's outbound half only: it reads as hops, the
# target, then the target's own hops mirrored back — drop the target and
# that whole return leg, keeping just the intermediate hops the user chose.
# 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.
tokens = [p.strip() for p in current.split(",") if p.strip()]
seed = [
cid
for h in tokens[: len(tokens) // 2]
if (cid := topo.canonical(h)) is not None
]
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
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).
@@ -780,12 +884,14 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
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
@@ -794,6 +900,53 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
continue
return result
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).
Returns:
The re-rendered spec, or ``None`` when cancelled, unchanged, or there is
no forced path to re-render (the width itself still sticks).
"""
nonlocal width_bytes
sample = (target_hash or device_hash or "").lower().removeprefix("0x")
items: list = []
for w in (1, 2, 4, 8):
title = Text(f"{w} byte{'s' if w > 1 else ' '}")
if sample:
title.append(" ")
title.append_text(highlighted_hash(sample[:16], w))
if w == device_width:
title.append(" · device default", style="muted")
items.append(Choice(title=title, value=w))
picked = await session.run_screen(
SelectScreen(
"path width",
items,
prompt="Forced hops are addressed by this many leading bytes of each key.",
default=width_bytes,
footer_hint="↑↓ · Enter set · Esc keep",
filterable=False,
wrap=False,
)
)
if picked is CANCEL or picked is None or int(picked) == width_bytes:
return None
width_bytes = int(picked)
tokens = [p.strip() for p in current.split(",") if p.strip()]
if not tokens:
return None
topo = fresh_topology()
full = [topo.canonical(t) or t for t in tokens]
width = collapse_width(*full, ceiling=width_bytes)
return ",".join(f[: width * 2] for f in full)
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"}
@@ -1051,6 +1204,8 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
trace=trace_once,
compose_path=compose,
explore=explore,
pick_width=pick_width,
width_bytes=lambda: width_bytes,
previous=ctx.repo.latest_trace(target),
)
try:
+58 -1
View File
@@ -72,7 +72,7 @@ def test_snr_bar_clamps_out_of_range_readings() -> None:
def _trace_screen(
trace=None, compose_path=None, explore=None, previous=None
trace=None, compose_path=None, explore=None, pick_width=None, previous=None
) -> tuple[TraceScreen, _FakeSession]:
session = _FakeSession()
@@ -91,6 +91,8 @@ def _trace_screen(
trace=trace or default_trace,
compose_path=compose_path or default_flow,
explore=explore or default_flow,
pick_width=pick_width or default_flow,
width_bytes=lambda: 2,
previous=previous,
)
return screen, session
@@ -198,6 +200,61 @@ async def test_trace_screen_explore_adopts_a_scenario_path() -> None:
assert screen._path_spec == "3d63,f2c2" # None leaves the spec untouched
async def test_trace_screen_action_cursor_commits_the_selected_row() -> None:
"""↑↓ move over the action rows; Enter commits the one under the cursor."""
opened: list[str] = []
async def width_flow(current): # noqa: ANN001
opened.append(f"width:{current}")
return "3d63,f2c2,3d63"
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
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."""
opened: list[str] = []
async def width_flow(current): # noqa: ANN001
opened.append(current)
return None # cancelled: the spec must stay untouched
screen, _ = _trace_screen(pick_width=width_flow)
screen._path_spec = "3d,f2,3d"
screen.handle("text", "w")
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
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."""
screen, _ = _trace_screen()
def faint_cells(text) -> int: # noqa: ANN001
return sum(
span.end - span.start for span in text.spans if "faint" in str(span.style)
)
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
custom = screen._planned_route()
assert custom.plain == "Us → 3d → f2 → 27 → Us"
assert faint_cells(symmetric) > faint_cells(custom)
async def test_trace_screen_opens_idle_until_enter() -> None:
"""Selecting a target must never transmit by itself: nothing flies until Enter."""
screen, session = _trace_screen()
+15
View File
@@ -72,6 +72,21 @@ def test_failed_trace_round_trips(repo: Repository) -> None:
assert got.hops == []
assert got.round_trip_ms is None
assert got.tx_power is None
assert got.path_hash_bytes is None # no hashes to derive a width from
def test_hydrated_traces_recover_the_path_hash_width(repo: Repository) -> None:
"""The stored hop hashes carry the trace's width, so rehydration recovers it.
Without this, a previous trace's route line renders every hash — including our
own device's full 64-char public key — at absurd lengths.
"""
run = repo.start_run("trace", {})
trace = _make_trace("Alice", ("3d63", 3.5), ("f2c2", -2.0), (None, 1.0))
repo.record_trace(run, trace)
assert repo.latest_trace("Alice").path_hash_bytes == 2
assert repo.recent_traces("Alice")[0].path_hash_bytes == 2
# -- recent_traces ---------------------------------------------------------
+68 -4
View File
@@ -21,7 +21,7 @@ from meshterm.persistence.repository import (
TracedPath,
)
from meshterm.services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths
from meshterm.services.topology import build_topology, collapse_width
from meshterm.services.topology import build_topology, collapse_width, render_custom_spec
from meshterm.ui.path_composer import AUTO_SPEC, FetchNeighbours, PathComposerScreen
from meshterm.ui.tui.screen import CANCEL
@@ -150,6 +150,15 @@ def test_scenario_spec_ends_at_target_and_collapses_width() -> None:
assert collapse_width("3d", "f2c24f54551e", ceiling=2) == 1 # narrowest hash rules
def test_render_custom_spec_is_verbatim_at_a_uniform_width() -> None:
"""An asymmetric walk renders exactly its hops — no target append, no mirror."""
hops = ("3d63c6429436", "f2c24f54551e", "27d4396a2967")
assert render_custom_spec(hops, 2) == "3d63,f2c2,27d4"
# A hop only known narrowly drags the whole spec down to what it can honour.
assert render_custom_spec(("3d63c6429436", "be"), 2) == "3d,be"
assert render_custom_spec((), 2) == "" # no hops, no spec (never a fake "auto")
# --- repository round-trip ---------------------------------------------------------
@@ -281,7 +290,7 @@ async def test_probe_paths_defaults_to_one_trace_per_candidate() -> None:
# --- path composer ---------------------------------------------------------------------
def _composer(topo, hops=None, fetch_nodes=frozenset()): # noqa: ANN001
def _composer(topo, hops=None, fetch_nodes=frozenset(), symmetric=True): # noqa: ANN001
return PathComposerScreen(
target_id="f2c24f54551e",
target_hash="f2c24f54551e" + "0" * 52,
@@ -291,6 +300,7 @@ def _composer(topo, hops=None, fetch_nodes=frozenset()): # noqa: ANN001
width_bytes=1,
hops=list(hops or []),
fetch_nodes=fetch_nodes,
symmetric=symmetric,
)
@@ -352,8 +362,9 @@ async def test_composer_commits_spec_auto_and_cancel() -> None:
screen = _composer(_topo(trace_paths=walks), hops=["3d63c6429436"])
screen.future = asyncio.get_running_loop().create_future()
screen.handle("end") # jump to the last action row (Cancel)…
screen.handle("up")
screen.handle("up") # …then back up to "Use this path"
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"
@@ -368,3 +379,56 @@ async def test_composer_commits_spec_auto_and_cancel() -> None:
cancelled.future = asyncio.get_running_loop().create_future()
cancelled.handle("escape")
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
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."""
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.
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).
assert any(s.node == "3d63c6429436" for s in screen._suggestions())
# Out via Hub, home directly off Far: nothing is appended or mirrored.
assert screen._spec() == "3d,f2"
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("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.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("enter")
assert not empty.future.done()