Add link-health regression detection against a rolling baseline

Splits a target's persisted trace history into a recent window and an older
baseline and flags end-to-end (bottleneck SNR, success rate) and per-hop link
SNR metrics that dropped beyond a threshold, with warn/critical severity.

Adds Repository.recent_traces / traced_targets, a pure link_quality service,
and a `link-health` tool that can analyze stored history or take fresh traces
first. Also tidies the repository import ordering.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.claude/settings.json

@@ -20,7 +20,8 @@
       "PowerShell(foreach \\($p in 'COM26','COM16','COM17'\\) { Write-Output \"=== $p ===\"; .venv\\\\Scripts\\\\python.exe -m meshtools --port $p --quiet info 2>&1 | Select-Object -First 4 })",
       "PowerShell(Get-CimInstance Win32_SerialPort | Where-Object { $_.DeviceID -eq 'COM11' } | Select-Object DeviceID, Description, Name | Format-List; Write-Output \"---\"; Get-Process | Where-Object { $_.Name -match 'meshcore|arduino|putty|python|node|serial|cu' } | Select-Object Id, ProcessName, MainWindowTitle | Format-Table -AutoSize)",
       "PowerShell(foreach \\($id in 25324,30408\\) { try { Stop-Process -Id $id -Force -ErrorAction Stop; Write-Output \"killed $id\" } catch { Write-Output \"could not kill ${id}: $\\($_.Exception.Message\\)\" } }; Start-Sleep -Milliseconds 800; Get-CimInstance Win32_Process -Filter \"Name='python.exe'\" | Select-Object ProcessId | Format-Table -AutoSize)",
-      "PowerShell(.venv\\\\Scripts\\\\python.exe -m meshtools --quiet info 2>&1)"
+      "PowerShell(.venv\\\\Scripts\\\\python.exe -m meshtools --quiet info 2>&1)",
+      "Bash(git commit -q -m ' *)"
     ]
   }
 }

README.md

@@ -19,6 +19,7 @@ turned into interactive HTML visualizations.
 | **TX-power optimization** | ✅ working | Sweep transmit power (coarse + refine), converge on the best signal, optionally apply, render an interactive chart. |
 | **Link budget** | ✅ working | Predict time-on-air, receiver sensitivity, range, and duty-cycle/dwell headroom for a radio config — offline, no radio needed. |
 | **Passive monitor** | ✅ working | Listen on the air and log every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history. |
+| **Link health** | ✅ working | Compare a target's recent traces to its rolling baseline and flag SNR/reliability regressions, end-to-end and per hop. |
 | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. |
 | **Visualizations** | 🟡 partial | Plotly TX-sweep chart (HTML) done; pyvis mesh graph + time-series planned. |
 

meshtools/persistence/repository.py

@@ -9,11 +9,10 @@ from __future__ import annotations
 import json
 import sqlite3
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from typing import Any, Optional
 
-from datetime import datetime
-
 from ..core.models import (
     HeardNode,
     Hop,
@@ -214,6 +213,56 @@ class Repository:
             timestamp=datetime.fromisoformat(row["created_at"]),
         )
 
+    def recent_traces(self, target: str, *, limit: int = 200) -> list[TraceResult]:
+        """Return recent traces to ``target``, newest first, rehydrated with hops.
+
+        This is the history the link-quality baseline is computed over, so both timed-out
+        and successful traces are included (a rise in failures is itself a regression).
+
+        Args:
+            target: The trace destination to load.
+            limit: Maximum number of traces to return.
+
+        Returns:
+            The matching :class:`TraceResult` objects, newest first.
+        """
+        rows = self._conn.execute(
+            "SELECT * FROM traces WHERE target = ? ORDER BY id DESC LIMIT ?",
+            (target, limit),
+        ).fetchall()
+        traces: list[TraceResult] = []
+        for row in rows:
+            hop_rows = self._conn.execute(
+                "SELECT hop_index, node, snr FROM trace_hops WHERE trace_id = ? "
+                "ORDER BY hop_index",
+                (row["id"],),
+            ).fetchall()
+            hops = [
+                Hop(index=h["hop_index"], node=h["node"], snr=h["snr"]) for h in hop_rows
+            ]
+            traces.append(
+                TraceResult(
+                    target=row["target"],
+                    success=bool(row["success"]),
+                    hops=hops,
+                    round_trip_ms=row["round_trip_ms"],
+                    tx_power=row["tx_power"],
+                    timestamp=datetime.fromisoformat(row["created_at"]),
+                )
+            )
+        return traces
+
+    def traced_targets(self) -> list[str]:
+        """Return the distinct trace destinations recorded, most-traced first.
+
+        Returns:
+            Target names ordered by descending trace count.
+        """
+        rows = self._conn.execute(
+            "SELECT target, COUNT(*) AS n FROM traces GROUP BY target ORDER BY n DESC"
+        ).fetchall()
+        return [row["target"] for row in rows]
+
     def record_tx_sample(self, run_id: int, level: TxLevelResult) -> None:
         """Persist one robust TX-power level from an optimization sweep.
 

meshtools/services/link_quality.py

@@ -0,0 +1,297 @@
+"""Link-quality regression detection against a rolling baseline.
+
+The trace and TX tools already persist every measurement, so a link's *history* is on
+disk. This service turns that history into an answer to "is this link worse than it
+usually is?" — it splits a target's traces into a recent window and an older baseline,
+then flags end-to-end and per-hop metrics that have degraded beyond a threshold (e.g. the
+yagi→local hop sitting 6 dB below its 30-day median, or the success rate falling off).
+
+It is pure logic over :class:`~meshtools.core.models.TraceResult` lists, so it runs
+identically on live data, replayed history, or the simulator.
+"""
+
+from __future__ import annotations
+
+import statistics
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+
+from ..core.models import LOCAL_DEVICE_LABEL, TraceResult, TraceStats
+
+#: Default SNR drop (dB) from baseline that counts as a regression.
+DEFAULT_SNR_DROP_DB = 3.0
+#: Default success-rate drop (fraction) from baseline that counts as a regression.
+DEFAULT_SUCCESS_DROP = 0.2
+#: Traces taken as the "recent" window (the rest form the baseline).
+DEFAULT_RECENT_COUNT = 3
+#: Minimum baseline traces required before a verdict is trustworthy.
+DEFAULT_MIN_BASELINE = 3
+
+NodeResolver = Callable[[Optional[str]], Optional[str]]
+
+
+@dataclass(slots=True)
+class MetricChange:
+    """A single metric compared between the baseline and recent windows.
+
+    Attributes:
+        subject: What the metric describes (the target, or a ``a → b`` link).
+        metric: Human-readable metric name (e.g. ``min SNR``).
+        kind: ``target`` for an end-to-end metric, ``link`` for one hop.
+        baseline: The baseline-window value.
+        recent: The recent-window value.
+        baseline_n: Number of baseline samples behind ``baseline``.
+        recent_n: Number of recent samples behind ``recent``.
+        threshold: The drop magnitude that marks a regression.
+        regressed: Whether ``recent`` is worse than ``baseline`` past ``threshold``.
+    """
+
+    subject: str
+    metric: str
+    kind: str
+    baseline: float
+    recent: float
+    baseline_n: int
+    recent_n: int
+    threshold: float
+    regressed: bool
+
+    @property
+    def delta(self) -> float:
+        """Signed change ``recent - baseline`` (negative means degraded)."""
+        return self.recent - self.baseline
+
+    @property
+    def severity(self) -> str:
+        """``critical`` when the drop is at least twice the threshold, else ``warn``/``ok``."""
+        if not self.regressed:
+            return "ok"
+        return "critical" if -self.delta >= 2 * self.threshold else "warn"
+
+
+@dataclass(slots=True)
+class HealthReport:
+    """The link-quality verdict for one target.
+
+    Attributes:
+        target: The trace destination analyzed.
+        baseline_n: Traces in the baseline window.
+        recent_n: Traces in the recent window.
+        changes: Every metric evaluated (target-level and per-link).
+    """
+
+    target: str
+    baseline_n: int
+    recent_n: int
+    changes: list[MetricChange] = field(default_factory=list)
+
+    @property
+    def regressions(self) -> list[MetricChange]:
+        """The subset of ``changes`` that actually regressed, worst drop first."""
+        return sorted(
+            (c for c in self.changes if c.regressed), key=lambda c: c.delta
+        )
+
+    @property
+    def status(self) -> str:
+        """``insufficient-data``, ``ok``, ``degraded`` (warn), or ``critical``."""
+        if not self.changes:
+            return "insufficient-data"
+        regs = self.regressions
+        if not regs:
+            return "ok"
+        if any(c.severity == "critical" for c in regs):
+            return "critical"
+        return "degraded"
+
+
+def analyze_target(
+    target: str,
+    traces: list[TraceResult],
+    *,
+    recent_count: int = DEFAULT_RECENT_COUNT,
+    min_baseline: int = DEFAULT_MIN_BASELINE,
+    snr_drop_db: float = DEFAULT_SNR_DROP_DB,
+    success_drop: float = DEFAULT_SUCCESS_DROP,
+    resolve: Optional[NodeResolver] = None,
+    device_label: str = LOCAL_DEVICE_LABEL,
+) -> HealthReport:
+    """Detect link-quality regressions for ``target`` from its trace history.
+
+    The traces are split by recency: the newest ``recent_count`` form the recent window,
+    the remainder the baseline. End-to-end bottleneck SNR and success rate are compared,
+    plus the median SNR of each per-hop link that appears in both windows.
+
+    Args:
+        target: The trace destination.
+        traces: Its trace history (any order; partitioned here by timestamp).
+        recent_count: Number of newest traces treated as the recent window.
+        min_baseline: Minimum baseline traces required for a verdict.
+        snr_drop_db: SNR drop (dB) from baseline that flags a regression.
+        success_drop: Success-rate drop (fraction) that flags a regression.
+        resolve: Optional resolver mapping a hop hash to a friendly node name.
+        device_label: Label for our own device at the path endpoints.
+
+    Returns:
+        A :class:`HealthReport`. With too little history its ``changes`` is empty and
+        ``status`` is ``insufficient-data``.
+    """
+    ordered = sorted(traces, key=lambda t: t.timestamp, reverse=True)
+    recent = ordered[:recent_count]
+    baseline = ordered[recent_count:]
+    if len(recent) < 2 or len(baseline) < min_baseline:
+        return HealthReport(target=target, baseline_n=len(baseline), recent_n=len(recent))
+
+    base_stats = TraceStats.from_traces(target, baseline)
+    recent_stats = TraceStats.from_traces(target, recent)
+    changes: list[MetricChange] = []
+
+    if base_stats.median_min_snr is not None and recent_stats.median_min_snr is not None:
+        changes.append(
+            _change(
+                subject=target,
+                metric="min SNR (dB)",
+                kind="target",
+                baseline=base_stats.median_min_snr,
+                recent=recent_stats.median_min_snr,
+                baseline_n=base_stats.successes,
+                recent_n=recent_stats.successes,
+                threshold=snr_drop_db,
+            )
+        )
+
+    changes.append(
+        _change(
+            subject=target,
+            metric="success rate",
+            kind="target",
+            baseline=base_stats.success_rate,
+            recent=recent_stats.success_rate,
+            baseline_n=base_stats.samples,
+            recent_n=recent_stats.samples,
+            threshold=success_drop,
+        )
+    )
+
+    changes.extend(
+        _link_changes(baseline, recent, snr_drop_db, resolve, device_label)
+    )
+    return HealthReport(
+        target=target, baseline_n=len(baseline), recent_n=len(recent), changes=changes
+    )
+
+
+def _change(
+    *,
+    subject: str,
+    metric: str,
+    kind: str,
+    baseline: float,
+    recent: float,
+    baseline_n: int,
+    recent_n: int,
+    threshold: float,
+) -> MetricChange:
+    """Build a :class:`MetricChange`, flagging it regressed on a drop past ``threshold``.
+
+    Both metrics here are "higher is better" (SNR, success rate), so a regression is a
+    decrease of at least ``threshold``.
+
+    Args:
+        subject: What the metric describes.
+        metric: Metric name.
+        kind: ``target`` or ``link``.
+        baseline: Baseline value.
+        recent: Recent value.
+        baseline_n: Baseline sample count.
+        recent_n: Recent sample count.
+        threshold: Drop magnitude that marks a regression.
+
+    Returns:
+        The populated :class:`MetricChange`.
+    """
+    regressed = (baseline - recent) >= threshold
+    return MetricChange(
+        subject=subject,
+        metric=metric,
+        kind=kind,
+        baseline=baseline,
+        recent=recent,
+        baseline_n=baseline_n,
+        recent_n=recent_n,
+        threshold=threshold,
+        regressed=regressed,
+    )
+
+
+def _link_changes(
+    baseline: list[TraceResult],
+    recent: list[TraceResult],
+    snr_drop_db: float,
+    resolve: Optional[NodeResolver],
+    device_label: str,
+) -> list[MetricChange]:
+    """Compare per-hop link SNR between the two windows.
+
+    Only links present in *both* windows are compared, so a transient path that appears
+    in just one window doesn't masquerade as a regression.
+
+    Args:
+        baseline: Baseline-window traces.
+        recent: Recent-window traces.
+        snr_drop_db: SNR drop (dB) that flags a link regression.
+        resolve: Optional hop-hash → name resolver for the link label.
+        device_label: Label for our own device.
+
+    Returns:
+        One :class:`MetricChange` per shared link, worst (most negative) drop first.
+    """
+    base_links = _link_snrs(baseline, resolve, device_label)
+    recent_links = _link_snrs(recent, resolve, device_label)
+    changes: list[MetricChange] = []
+    for link in base_links.keys() & recent_links.keys():
+        base_snrs = base_links[link]
+        recent_snrs = recent_links[link]
+        changes.append(
+            _change(
+                subject=link,
+                metric="link SNR (dB)",
+                kind="link",
+                baseline=statistics.median(base_snrs),
+                recent=statistics.median(recent_snrs),
+                baseline_n=len(base_snrs),
+                recent_n=len(recent_snrs),
+                threshold=snr_drop_db,
+            )
+        )
+    return sorted(changes, key=lambda c: c.delta)
+
+
+def _link_snrs(
+    traces: list[TraceResult], resolve: Optional[NodeResolver], device_label: str
+) -> dict[str, list[float]]:
+    """Collect per-link SNR samples keyed by an ``origin → destination`` label.
+
+    Args:
+        traces: The traces to read edges from.
+        resolve: Optional hop-hash → name resolver.
+        device_label: Label for our own device at the endpoints.
+
+    Returns:
+        A mapping of link label to the SNR readings seen on it.
+    """
+    out: dict[str, list[float]] = {}
+    for trace in traces:
+        if not trace.success:
+            continue
+        for edge in trace.edges(device_label):
+            label = f"{_name(edge.origin, resolve)} → {_name(edge.destination, resolve)}"
+            out.setdefault(label, []).append(edge.snr)
+    return out
+
+
+def _name(label: str, resolve: Optional[NodeResolver]) -> str:
+    """Resolve a hop label to a friendly name when a resolver is supplied."""
+    if resolve is None:
+        return label
+    return resolve(label) or label

meshtools/tools/link_health.py

@@ -0,0 +1,281 @@
+"""The ``link-health`` tool: flag links that have degraded against their own baseline.
+
+Every trace MeshTools runs is persisted, so each link has a history. This tool splits a
+target's trace history into a recent window and an older baseline and reports any
+end-to-end or per-hop metric that has dropped beyond a threshold — turning the live SNR
+number the app shows into "this hop is 6 dB below its usual median". Optionally it takes a
+few fresh traces first, so it can be run standalone rather than only over stored data.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import questionary
+import typer
+from rich.panel import Panel
+from rich.table import Table
+from rich.text import Text
+
+from ..context import AppContext
+from ..services import link_quality, trace_runner
+from ..services.link_quality import HealthReport, MetricChange
+from ..ui.widgets import make_progress
+from .base import Tool, ToolResult, register
+
+_STATUS_STYLE = {
+    "ok": "ok",
+    "degraded": "warn",
+    "critical": "err",
+    "insufficient-data": "muted",
+}
+_SEVERITY_STYLE = {"critical": "err", "warn": "warn", "ok": "ok"}
+
+
+@register
+class LinkHealthTool(Tool):
+    """Detect link-quality regressions for a target against its rolling baseline."""
+
+    name = "link-health"
+    help = "Flag links whose SNR or reliability dropped below their historical baseline."
+    category = "Diagnostics"
+    order = 30
+
+    async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]:
+        """Choose a target (from history or contacts) and an optional fresh-sample count.
+
+        Args:
+            ctx: Shared application context.
+
+        Returns:
+            A parameter dict, or ``None`` if the user cancelled.
+        """
+        # Offer targets we already have history for first, then known contacts.
+        history = ctx.repo.traced_targets()
+        try:
+            contacts = [c.name for c in await (await ctx.device()).get_contacts()]
+        except Exception:  # noqa: BLE001 - history-only analysis needs no device
+            contacts = []
+        choices = list(dict.fromkeys(history + contacts))
+        prompt = "Target to check:"
+        if choices:
+            target = await questionary.autocomplete(
+                prompt, choices=choices, ignore_case=True
+            ).ask_async()
+        else:
+            target = await questionary.text(prompt).ask_async()
+        if not target:
+            return None
+
+        samples = await questionary.text(
+            "Take how many fresh traces first? (0 = use stored history only)",
+            default="0",
+            validate=_is_nonneg_int,
+        ).ask_async()
+        if samples is None:
+            return None
+        return {"target": target.strip(), "samples": int(samples)}
+
+    async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Optionally trace, load history, analyze regressions, and render the report.
+
+        Args:
+            ctx: Shared application context.
+            params: ``target``, optional ``samples``/``path``/``recent``/``snr_drop``/
+                ``success_drop``/``min_baseline``, and the injected ``_run_id``.
+
+        Returns:
+            A :class:`ToolResult` carrying the verdict and any regressions.
+        """
+        target = params["target"]
+        run_id = params["_run_id"]
+        samples = int(params.get("samples", 0))
+        resolve = None
+
+        if samples > 0:
+            device = await ctx.device()
+            contacts = await device.get_contacts()
+            resolve = trace_runner.make_node_resolver(contacts)
+            path = None
+            if params.get("path"):
+                path = trace_runner.parse_trace_path(params["path"], contacts)
+            with make_progress(ctx.console) as progress:
+                task = progress.add_task(f"tracing {target}", total=samples)
+                await trace_runner.run_traces(
+                    device,
+                    target,
+                    samples=samples,
+                    path=path,
+                    cooldown_s=ctx.settings.trace_cooldown_s,
+                    on_result=lambda *_: progress.advance(task),
+                    persist=lambda t: ctx.repo.record_trace(run_id, t),
+                )
+
+        traces = ctx.repo.recent_traces(target)
+        if resolve is None:
+            # No fresh run: still try to name hops, but tolerate having no device.
+            try:
+                contacts = await (await ctx.device()).get_contacts()
+                resolve = trace_runner.make_node_resolver(contacts)
+            except Exception:  # noqa: BLE001 - analysis works on raw hashes too
+                resolve = None
+
+        report = link_quality.analyze_target(
+            target,
+            traces,
+            recent_count=int(params.get("recent", link_quality.DEFAULT_RECENT_COUNT)),
+            min_baseline=int(params.get("min_baseline", link_quality.DEFAULT_MIN_BASELINE)),
+            snr_drop_db=float(params.get("snr_drop", link_quality.DEFAULT_SNR_DROP_DB)),
+            success_drop=float(params.get("success_drop", link_quality.DEFAULT_SUCCESS_DROP)),
+            resolve=resolve,
+        )
+
+        ctx.console.print(_report_panel(report))
+        if report.status == "insufficient-data":
+            message = (
+                f"[muted]not enough history for[/muted] [brand]{target}[/brand] "
+                f"({report.baseline_n} baseline + {report.recent_n} recent traces); "
+                "run more traces or the monitor first."
+            )
+        elif report.status == "ok":
+            message = f"[ok]✓[/ok] [brand]{target}[/brand] is healthy vs its baseline"
+        else:
+            n = len(report.regressions)
+            style = _STATUS_STYLE[report.status]
+            message = (
+                f"[{style}]![/{style}] [brand]{target}[/brand] shows "
+                f"[{style}]{n} regression{'s' if n != 1 else ''}[/{style}] "
+                f"({report.status})"
+            )
+
+        return ToolResult(
+            summary={
+                "target": report.target,
+                "status": report.status,
+                "baseline_n": report.baseline_n,
+                "recent_n": report.recent_n,
+                "regressions": [
+                    {
+                        "subject": c.subject,
+                        "metric": c.metric,
+                        "baseline": round(c.baseline, 2),
+                        "recent": round(c.recent, 2),
+                        "delta": round(c.delta, 2),
+                        "severity": c.severity,
+                    }
+                    for c in report.regressions
+                ],
+            },
+            message=message,
+        )
+
+    def register_cli(self, app: typer.Typer) -> None:
+        """Register the ``link-health`` subcommand.
+
+        Args:
+            app: The Typer application.
+        """
+        from ..cli import run_tool_command
+
+        @app.command(name=self.name, help=self.help)
+        def _link_health(
+            target: str = typer.Option(..., "--target", "-t", help="Target node name/prefix."),
+            samples: int = typer.Option(
+                0, "--samples", "-n", help="Fresh traces to take before analyzing."
+            ),
+            path: Optional[str] = typer.Option(
+                None, "--path", "-p", help="Force a path for the fresh traces."
+            ),
+            recent: int = typer.Option(
+                link_quality.DEFAULT_RECENT_COUNT, "--recent", help="Newest traces as 'recent'."
+            ),
+            snr_drop: float = typer.Option(
+                link_quality.DEFAULT_SNR_DROP_DB, "--snr-drop", help="SNR drop (dB) to flag."
+            ),
+            success_drop: float = typer.Option(
+                link_quality.DEFAULT_SUCCESS_DROP,
+                "--success-drop",
+                help="Success-rate drop (fraction) to flag.",
+            ),
+        ) -> None:
+            tool_params: dict[str, Any] = {
+                "target": target, "samples": samples, "recent": recent,
+                "snr_drop": snr_drop, "success_drop": success_drop,
+            }
+            if path:
+                tool_params["path"] = path
+            run_tool_command(self, tool_params)
+
+
+def _report_panel(report: HealthReport) -> Panel:
+    """Render the health report as a status header plus a table of changes.
+
+    Args:
+        report: The analyzed report.
+
+    Returns:
+        A Rich :class:`Panel` summarizing the verdict and metric changes.
+    """
+    style = _STATUS_STYLE.get(report.status, "muted")
+    header = Text.assemble(
+        ("target   ", "muted"), (f"{report.target}\n", "brand"),
+        ("status   ", "muted"), (report.status, style), ("\n", ""),
+        ("window   ", "muted"),
+        (f"{report.recent_n} recent vs {report.baseline_n} baseline traces", ""),
+    )
+    if not report.changes:
+        return Panel(
+            header, title="[accent]link health[/accent]", border_style=style, expand=False
+        )
+
+    table = Table(box=None, padding=(0, 2, 0, 0), expand=False)
+    table.add_column("metric", style="muted")
+    table.add_column("subject")
+    table.add_column("baseline", justify="right")
+    table.add_column("recent", justify="right")
+    table.add_column("Δ", justify="right")
+    # Worst regressions first, then the still-healthy metrics for context.
+    for change in sorted(report.changes, key=lambda c: (not c.regressed, c.delta)):
+        table.add_row(*_change_row(change))
+
+    from rich.console import Group
+
+    body = Group(header, Text(), table)
+    return Panel(body, title="[accent]link health[/accent]", border_style=style, expand=False)
+
+
+def _change_row(change: MetricChange) -> tuple[str, Text, str, str, Text]:
+    """Format one metric change as a styled table row.
+
+    Args:
+        change: The metric change to render.
+
+    Returns:
+        The five cell values for the changes table.
+    """
+    sev_style = _SEVERITY_STYLE.get(change.severity, "muted")
+    marker = "● " if change.regressed else "  "
+    delta_cell = Text(f"{marker}{change.delta:+.2f}", style=sev_style)
+    subject = Text(change.subject, style="brand" if change.regressed else None)
+    return (
+        change.metric,
+        subject,
+        f"{change.baseline:.2f}",
+        f"{change.recent:.2f}",
+        delta_cell,
+    )
+
+
+def _is_nonneg_int(value: str) -> bool | str:
+    """Validate a non-negative integer for a questionary text answer.
+
+    Args:
+        value: Raw input.
+
+    Returns:
+        ``True`` if valid, otherwise an error message string.
+    """
+    try:
+        return True if int(value) >= 0 else "Enter zero or a positive number."
+    except ValueError:
+        return "Enter a whole number."

tests/test_link_quality.py

@@ -0,0 +1,93 @@
+"""Tests for link-quality regression detection against a rolling baseline."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from meshtools.core.models import Hop, TraceResult, utcnow
+from meshtools.services import link_quality
+
+_NOW = utcnow()
+
+
+def _trace(snr: float, age_s: int, *, success: bool = True, hop: str = "3d") -> TraceResult:
+    """Build a two-hop trace; bottleneck SNR is ``snr`` and it is ``age_s`` seconds old."""
+    return TraceResult(
+        target="Alice",
+        success=success,
+        hops=[Hop(0, hop, snr), Hop(1, None, snr + 1)] if success else [],
+        timestamp=_NOW - timedelta(seconds=age_s),
+    )
+
+
+def test_detects_snr_regression() -> None:
+    """A recent SNR drop past the threshold is flagged, with critical severity."""
+    # Baseline (older) ~+8 dB, recent (newer) ~+2 dB: a 6 dB drop.
+    baseline = [_trace(8.0, age_s=100 + i) for i in range(5)]
+    recent = [_trace(2.0, age_s=i) for i in range(3)]
+    report = link_quality.analyze_target("Alice", baseline + recent, recent_count=3)
+
+    assert report.status == "critical"  # 6 dB drop ≥ 2× the 3 dB threshold
+    snr_reg = next(c for c in report.regressions if c.metric == "min SNR (dB)")
+    assert snr_reg.baseline == 8.0
+    assert snr_reg.recent == 2.0
+    assert snr_reg.delta == -6.0
+    # The per-hop link also regressed.
+    assert any(c.kind == "link" and c.regressed for c in report.regressions)
+
+
+def test_healthy_link_reports_ok() -> None:
+    """A steady link shows no regressions and an ok status."""
+    traces = [_trace(7.5, age_s=i) for i in range(8)]
+    report = link_quality.analyze_target("Alice", traces, recent_count=3)
+    assert report.status == "ok"
+    assert report.regressions == []
+    assert report.changes  # metrics were still evaluated
+
+
+def test_success_rate_regression() -> None:
+    """A jump in recent failures is flagged even when surviving traces look fine."""
+    baseline = [_trace(7.0, age_s=100 + i) for i in range(5)]
+    recent = [_trace(7.0, age_s=i, success=False) for i in range(3)]
+    report = link_quality.analyze_target("Alice", baseline + recent, recent_count=3)
+
+    success_reg = next(c for c in report.regressions if c.metric == "success rate")
+    assert success_reg.baseline == 1.0
+    assert success_reg.recent == 0.0
+    assert report.status == "critical"
+
+
+def test_insufficient_history() -> None:
+    """Too little baseline data yields an insufficient-data verdict, not a false alarm."""
+    traces = [_trace(8.0, age_s=i) for i in range(3)]  # only 3 total, < min baseline + recent
+    report = link_quality.analyze_target("Alice", traces, recent_count=3, min_baseline=3)
+    assert report.status == "insufficient-data"
+    assert report.changes == []
+
+
+def test_links_only_compared_when_present_in_both_windows() -> None:
+    """A path that appears in only one window is not reported as a link regression."""
+    # Baseline goes via hop 'aa'; recent via a different hop 'bb' — no shared link.
+    baseline = [_trace(8.0, age_s=100 + i, hop="aa") for i in range(5)]
+    recent = [_trace(8.0, age_s=i, hop="bb") for i in range(3)]
+    report = link_quality.analyze_target("Alice", baseline + recent, recent_count=3)
+    link_changes = [c for c in report.changes if c.kind == "link"]
+    # Only the shared 'X → us' return links overlap; the 'us → aa'/'us → bb' legs don't.
+    assert all("aa" not in c.subject or "bb" not in c.subject for c in link_changes)
+    assert report.status == "ok"
+
+
+def test_resolver_names_link_subjects() -> None:
+    """A resolver renames hop hashes in link subjects for readable output."""
+    baseline = [_trace(8.0, age_s=100 + i) for i in range(5)]
+    recent = [_trace(2.0, age_s=i) for i in range(3)]
+    report = link_quality.analyze_target(
+        "Alice",
+        baseline + recent,
+        recent_count=3,
+        resolve=lambda h: "Repeater" if h == "3d" else h,
+        device_label="Me",
+    )
+    link_regs = [c for c in report.regressions if c.kind == "link"]
+    assert any("Repeater" in c.subject for c in link_regs)
+    assert any("Me" in c.subject for c in link_regs)