Trace splits in two: Trace target vs Trace path, with paced sampling

A trace is one walked path and "target" is pure UX (hardware-verified),
so the feature now splits along that line: Trace target keeps the
symmetric mirror-and-return flow (picker, Explore paths), while the new
Trace path skips the picker entirely and walks a hand-composed route —
no Auto, no Explore, Trace gated until a path exists, history recorded
under the "(path)" sentinel so composed walks never pollute the target
picker. The composer loses its mode toggle: a pinned target means
mirrored, no target means verbatim, and its preview opens as us → us.

Sampling returns, deliberately: a Sample count action (1/2/3/5/8) runs
that many traces per commit, paced by trace_cooldown_s between sends and
counted off on the in-flight dialog — never a burst. Actions reorder to
compose/explore/width/samples/trace with a Back row; the w/p/x hotkeys
retire.

Display: route nodes drop the underscore (prompt_toolkit draws U+00A0 as
"_", so name (hash) now joins with plain spaces), the composer renders
nodes exactly like the route line, the (previous · …) stamp gets its own
line, the path row ends with the displayed hop count, Explore rows lead
with the route instead of an "observed path" label, and the empty Traces
placeholder is gone.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Fable 5 committed 2026-07-10 23:33:12 -04:00
1 parent 22b182000b
commit 0ded99883f
10 files changed
+897 -459

No files matched your search

+8
View File
@@ -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.
+5 -2
View File
@@ -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]
+192 -79
View File
@@ -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.
+99 -138
View File
@@ -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:
+369 -167
View File
@@ -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)
+6 -14
View File
@@ -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)
)
+152 -25
View File
@@ -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."""
+13 -1
View File
@@ -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 -----------------------------------------------
+16
View File
@@ -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):
+37 -33
View File
@@ -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()