Manage and restore OTA relay hop limits automatically

tools/lora_ota/lora_ota.py

@@ -144,6 +144,8 @@ INSTALL_RECONCILE_WAIT_SECONDS = 20
 DEFAULT_RELAY_TX_DELAY = 0.3
 RELAY_TIMING_COMMANDS_PER_RELAY = 12
 RELAY_TIMING_RECOVERY_FILE = "relay-timing-settings.json"
+OTA_HOPS_RECOVERY_FILE = "ota-hop-settings.json"
+OTA_MAX_HOPS = 8
 TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json"
 SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json"
 MIN_MESHCLI_VERSION = (1, 6, 0)
@@ -306,6 +308,19 @@ class RelayTimingSettings:
     txdelay: float
 
 
+@dataclass(frozen=True)
+class OtaHopSettings:
+    name: str
+    password: str | None
+    local_source: bool
+    original: int
+    minimum: int
+
+    @property
+    def transfer(self) -> int:
+        return max(self.original, self.minimum)
+
+
 @dataclass(frozen=True)
 class RemoteClockSample:
     """One authenticated remote epoch with complete host-time uncertainty."""
@@ -1614,6 +1629,13 @@ def reply_matches_command(command_text: str, reply: str) -> bool:
         )
     if command == "ota stats":
         return text.startswith("OTA | fw ") or is_unknown or needs_temp
+    if command == "ota config":
+        return (
+            lowered.startswith("ota config:") or is_unknown or is_error
+            or re.search(r"\bota (?:is )?not included\b", lowered) is not None
+        )
+    if command.startswith("ota config hops "):
+        return lowered.startswith("ok ota reach = ") or is_unknown or is_error
     if command == "get bootloader.ver":
         try:
             parse_bootloader_version_reply(text)
@@ -2986,6 +3008,11 @@ def recommended_temp_radio_minutes(
         0 if args.source_already_temp or args.source_shares_controller else 30
     )
     final_reply_count = (4 + relays) if include_install else 1
+    hop_restore_seconds = 0
+    if relays or (getattr(args, "ota_hops", None) or 0) > 0:
+        # A policy restore needs a current read, absolute setter, and readback
+        # on each remote participant plus the local source CLI.
+        hop_restore_seconds = 3 * ((1 + relays) * args.reply_timeout + 30)
     required_seconds = (
         remote_setup_seconds
         + source_setup_seconds
@@ -2996,6 +3023,7 @@ def recommended_temp_radio_minutes(
         + adaptive_poll_ceiling(args.poll_seconds)
         + args.reply_timeout * final_reply_count
         + relays * RELAY_TIMING_COMMANDS_PER_RELAY * args.reply_timeout
+        + hop_restore_seconds
     )
     return required_seconds // 60 + 1
 
@@ -4336,6 +4364,167 @@ def parse_delay_reply(reply: str, label: str) -> float:
     return float(match.group(1))
 
 
+def parse_ota_hops(reply: str, label: str) -> int:
+    # A local serial terminal may include a welcome banner and command echo.
+    # Only the actual config line is authoritative, never another dotted or
+    # numeric value from that surrounding output.
+    lines = re.findall(r"(?mi)^[ \t]*(?:>[ \t]*)?ota config:[^\r\n]*", reply)
+    if len(lines) == 1:
+        values = re.findall(r"(?<!\S)hops=([^\s]+)", lines[0])
+        if len(values) == 1 and re.fullmatch(r"[0-8]", values[0]):
+            return int(values[0])
+    raise OtaError(f"could not read {label} OTA hop limit: {reply}")
+
+
+def ota_hops_command(
+    controller: Controller,
+    args: argparse.Namespace,
+    saved: OtaHopSettings,
+    command: str,
+    *,
+    retry: bool = True,
+) -> str:
+    if saved.local_source:
+        return source_cli_command(args, command, bounded=True, retry=retry)
+
+    def once() -> str:
+        return controller.remote_command(
+            saved.name, command, password=saved.password, retry=False,
+            operation_timeout=TEMP_RADIO_PREFLIGHT_OPERATION_TIMEOUT_SECONDS,
+        )
+
+    if retry:
+        return retry_transmission_bounded(once, f"{command!r} on {saved.name}")
+    return once()
+
+
+def read_ota_hops(
+    controller: Controller, args: argparse.Namespace, saved: OtaHopSettings,
+) -> int:
+    return parse_ota_hops(ota_hops_command(controller, args, saved, "ota config"), saved.name)
+
+
+def prepare_ota_hop_settings(
+    controller: Controller, args: argparse.Namespace,
+) -> list[OtaHopSettings]:
+    """Read every policy before confirmation; never guess an unknown value."""
+    minimum = max(len(args.relay_values), getattr(args, "ota_hops", None) or 0)
+    if minimum == 0:
+        return []  # Preserve direct-run behavior and existing user policy.
+    if minimum > OTA_MAX_HOPS:
+        raise OtaError(f"OTA supports at most {OTA_MAX_HOPS} relay hops")
+    if not has_managed_source_cli(args):
+        raise OtaError(
+            "automatic OTA hop setup requires a managed source CLI; supply "
+            "--source-cli-serial or --source-cli-tcp instead of --source-already-temp"
+        )
+    source_name = (
+        getattr(args, "shared_source_public_key", None)
+        or getattr(args, "source_contact_value", None) or "OTA source"
+    )
+    participants = [
+        OtaHopSettings(source_name, None, True, 0, minimum),
+        OtaHopSettings(args.target, None, False, 0, minimum),
+        *(OtaHopSettings(name, password, False, 0, minimum)
+          for name, password in args.relay_values),
+    ]
+    result = []
+    for index, participant in enumerate(participants):
+        reply = ota_hops_command(controller, args, participant, "ota config")
+        # Non-OTA repeaters can forward OTA opaquely. Skip only an explicit
+        # absence of the build feature. An unknown command alone does not
+        # establish that this firmware has no OTA receive/forward policy.
+        if index >= 2 and re.fullmatch(
+            r"\s*(?:LoRa\s+)?OTA (?:is )?not included in this build"
+            r"(?:[;.:!][^\r\n]*)?\s*",
+            reply, re.IGNORECASE,
+        ):
+            print(f"[hops] {participant.name}: opaque relay; no OTA policy to change")
+            continue
+        result.append(replace(participant, original=parse_ota_hops(reply, participant.name)))
+    return result
+
+
+def set_ota_hops_verified(
+    controller: Controller, args: argparse.Namespace, saved: OtaHopSettings, value: int,
+) -> None:
+    try:
+        reply = ota_hops_command(
+            controller, args, saved, f"ota config hops {value}", retry=False,
+        )
+    except TransmissionError:
+        # A missing acknowledgement is not permission to send another write.
+        # The absolute, idempotent setting is reconciled by an exact readback.
+        reply = None
+    if reply is not None and not re.fullmatch(
+        rf"\s*(?:>[ \t]*)?OK OTA reach = {value} hops? \(saved\)"
+        r"(?: - direct only)?\s*", reply, re.IGNORECASE,
+    ):
+        # Serial replies can carry a banner. Find exactly one complete ACK
+        # line, rather than accepting an OK from an unrelated command.
+        ack = re.findall(
+            rf"(?mi)^[ \t]*(?:>[ \t]*)?OK OTA reach = {value} hops? \(saved\)"
+            r"(?: - direct only)?[ \t]*\r?$", reply,
+        )
+        if len(ack) != 1:
+            raise OtaError(f"{saved.name} refused OTA hop limit {value}: {reply}")
+    if read_ota_hops(controller, args, saved) != value:
+        raise OtaError(f"{saved.name} OTA hop limit did not read back as {value}")
+
+
+def apply_ota_hop_settings(
+    controller: Controller, args: argparse.Namespace, settings: list[OtaHopSettings],
+    owned: list[OtaHopSettings], work_dir: Path,
+) -> None:
+    if not settings:
+        return
+    # Rehearsal/preparation can take minutes. Validate ALL snapshots again
+    # before the first write, so another operator's change is not overwritten.
+    for saved in settings:
+        if read_ota_hops(controller, args, saved) != saved.original:
+            raise OtaError(
+                f"{saved.name} OTA hop limit changed during preparation; "
+                "leaving the external setting unchanged"
+            )
+    changes = [saved for saved in settings if saved.transfer != saved.original]
+    if changes:
+        path = write_private_recovery_file(work_dir / OTA_HOPS_RECOVERY_FILE, json.dumps({
+            "nodes": [{
+                "name": saved.name, "local_source": saved.local_source,
+                "original_hops": saved.original, "transfer_hops": saved.transfer,
+                "restore_command": f"ota config hops {saved.original}",
+            } for saved in changes],
+            "source_endpoint": {
+                "serial": getattr(args, "source_cli_serial", None) or getattr(args, "source_serial", None),
+                "tcp": getattr(args, "source_cli_tcp", None),
+            },
+        }, indent=2) + "\n")
+        print(f"[hops] recovery settings: {path}")
+    for saved in changes:
+        owned.append(saved)  # Own cleanup BEFORE a possibly lost mutation.
+        set_ota_hops_verified(controller, args, saved, saved.transfer)
+        print(f"[hops] {saved.name}: temporarily {saved.original} -> {saved.transfer} (verified)")
+
+
+def restore_ota_hop_settings(
+    controller: Controller, args: argparse.Namespace, owned: list[OtaHopSettings],
+) -> None:
+    errors = []
+    for saved in reversed(owned.copy()):
+        try:
+            current = read_ota_hops(controller, args, saved)
+            if current != saved.original:
+                if current != saved.transfer:
+                    raise OtaError("OTA hop limit changed externally; leaving it unchanged")
+                set_ota_hops_verified(controller, args, saved, saved.original)
+            owned.remove(saved)  # Retire ownership only after exact readback.
+            print(f"[hops] {saved.name}: original {saved.original} restored")
+        except (OtaError, OSError) as exc:
+            errors.append(f"{saved.name}: {exc}")
+    if errors:
+        raise OtaError("could not restore OTA hop settings: " + "; ".join(errors))
+
+
 def read_relay_timing(
     controller: Controller,
     relay_name: str,
@@ -4494,6 +4683,9 @@ def confirm_update(
             f"  relay timing: rxdelay 0, "
             f"txdelay {format_decimal(args.relay_txdelay)} (saved/restored)"
         )
+    for saved in getattr(args, "ota_hop_settings", []):
+        action = "unchanged" if saved.original == saved.transfer else "temporary; saved/restored"
+        print(f"  OTA hops    : {saved.name}: {saved.original} -> {saved.transfer} ({action})")
     print(f"  action      : {'stage only' if args.no_install else 'install and reboot'}")
     current_version = (
         parse_version(target.current_version) if target.current_version else None
@@ -7198,6 +7390,11 @@ def build_parser() -> argparse.ArgumentParser:
         "--relay", action="append", default=[], metavar="NAME_OR_KEY[=PASSWORD]",
         help="optional relay name/key/unique key prefix, farthest-to-nearest; repeat as needed",
     )
+    parser.add_argument(
+        "--ota-hops", type=int, metavar="MINIMUM",
+        help=("minimum OTA reach for undeclared additional hops (0..8); otherwise "
+              "infer from --relay count; only raise lower limits and restore afterward"),
+    )
     parser.add_argument(
         "--source-contact",
         metavar="NAME_OR_KEY",
@@ -7387,6 +7584,10 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
         parser.error(f"--temp-radio: {exc}")
     if not math.isfinite(args.relay_txdelay) or not 0.0 <= args.relay_txdelay <= 2.0:
         parser.error("--relay-txdelay must be between 0 and 2")
+    if len(args.relay) > OTA_MAX_HOPS:
+        parser.error(f"at most {OTA_MAX_HOPS} --relay nodes are supported")
+    if args.ota_hops is not None and not len(args.relay) <= args.ota_hops <= OTA_MAX_HOPS:
+        parser.error("--ota-hops must be 0..8 and at least the declared --relay count")
     clear_manifest = args.clear_completed_manifest
     clear_body_hash = args.clear_completed_on_body_hash
     if bool(clear_manifest) != bool(clear_body_hash):
@@ -8269,6 +8470,7 @@ def main(
     armed_relay_values: list[tuple[str, str]] = []
     relay_public_keys: dict[str, str] = {}
     relay_timing_settings: list[RelayTimingSettings] = []
+    owned_ota_hop_settings: list[OtaHopSettings] = []
     password = args.password or os.environ.get("MESHCORE_ADMIN_PASSWORD", "")
     temp_command = f"tempradio {args.temp_radio}"
 
@@ -8464,6 +8666,7 @@ def main(
             target_rxps_profile = None
         args.target_rxps_saved = target_rxps_saved
         args.target_rxps_profile = target_rxps_profile
+        args.ota_hop_settings = prepare_ota_hop_settings(controller, args)
         confirm_update(args, target, package)
 
         freq, bandwidth, sf, cr, _minutes = args.temp_values
@@ -8482,6 +8685,9 @@ def main(
             original_radio,
             temp_radio,
         )
+        apply_ota_hop_settings(
+            controller, args, args.ota_hop_settings, owned_ota_hop_settings, work_dir,
+        )
 
         # Package generation above belongs on a workstation/build VM; a
         # Pi-class radio host should receive only the completed hash-verified
@@ -8638,6 +8844,13 @@ def main(
         seeder.start()
         find_and_start_pull(controller, args, package, seeder)
         monitor_download(controller, args, package, seeder)
+        # A serial source is exclusively owned by motatool. Release that
+        # ownership before trying to restore its policy through the text CLI.
+        seeder.stop()
+        seeder = None
+        # Transfer is complete; restore while every participant is still on
+        # the same tuple, before installation can reboot the destination.
+        restore_ota_hop_settings(controller, args, owned_ota_hop_settings)
 
         if args.no_install:
             report_staged_update(controller, args, package)
@@ -8677,8 +8890,6 @@ def main(
                         controller, args.target, target_rxps_saved
                     )
                 target_rxps_changed = False
-            seeder.stop()
-            seeder = None
             if source_temp_owned and (
                 not args.leave_controller_radio
                 or not args.source_shares_controller
@@ -8721,9 +8932,7 @@ def main(
             relay_public_keys,
         )
 
-        # Stop seeding before returning the controller to its ordinary channel.
-        seeder.stop()
-        seeder = None
+        # Seeding already stopped before policy restoration and installation.
         if source_temp_owned and (
             not args.leave_controller_radio
             or not args.source_shares_controller
@@ -8808,6 +9017,11 @@ def main(
                             file=sys.stderr,
                         )
                 if controller_changed:
+                    if owned_ota_hop_settings:
+                        try:
+                            restore_ota_hop_settings(controller, args, owned_ota_hop_settings)
+                        except (OtaError, OSError) as exc:
+                            print(f"WARNING: {exc}; retrying after radio restore", file=sys.stderr)
                     if relay_timing_settings:
                         try:
                             restore_relay_timings(controller, relay_timing_settings)
@@ -8931,6 +9145,15 @@ def main(
                         f"{recovery_hint}: {exc}",
                         file=sys.stderr,
                     )
+            if owned_ota_hop_settings and controller is not None:
+                try:
+                    restore_ota_hop_settings(controller, args, owned_ota_hop_settings)
+                except (OtaError, OSError) as exc:
+                    print(
+                        f"CRITICAL: {exc}; use {work_dir / OTA_HOPS_RECOVERY_FILE} "
+                        "after checking the current policy and radio channel",
+                        file=sys.stderr,
+                    )
             if not source_rxps_changed:
                 try:
                     retire_local_source_rxps_recovery()

tools/lora_ota/rak3401_mota_chain.py

@@ -2408,6 +2408,11 @@ def run_step(
         ])
     if args.debug:
         command.append("--debug")
+    # The outer runner's destination policy alone cannot make a direct-only
+    # source/relay transport an indirect route. Carry the reach into the
+    # generic runner's temporary, saved/restored participant policy as well.
+    if getattr(args, "ota_hops", None) is not None:
+        command.extend(["--ota-hops", str(max(args.ota_hops, len(relays)))])
     for relay_key, relay_password in relays:
         command.extend(["--relay", f"{relay_key}={relay_password}"])
     result = ota.main(command, controller_override=controller)

tools/lora_ota/test_lora_ota.py

@@ -7252,6 +7252,7 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
             reboot_wait=90,
             relay=[],
             debug=False,
+            ota_hops=3,
         )
         step = mock.Mock(
             number=1,
@@ -7278,6 +7279,7 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
                     )
                 command = nested.call_args.args[0]
                 self.assertEqual(command.count("--debug"), int(enabled))
+                self.assertEqual(command[command.index("--ota-hops") + 1], "3")
                 self.assertEqual(command[1], target_key)
                 self.assertEqual(command[command.index("--relay") + 1], f"{relay_key}=relay-password")
 

tools/lora_ota/test_lora_ota_hops.py

@@ -0,0 +1,438 @@
+"""Offline policy and lifecycle tests; no radio access or firmware writes."""
+import argparse
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+from test_lora_ota import firmware, mota_blob, target, VERSION_NEW
+
+
+class Stations:
+    def __init__(self, relays=1, source_hops=0, target_hops=0):
+        self.hops = {"source": source_hops, "remote": target_hops,
+                     **{f"relay{i}": 0 for i in range(relays)}}
+        self.calls = []
+        self.events = []
+        self.lost = set()
+        self.fail_reads = set()
+        self.overrides = {}
+        self.seeding = False
+        self.normal = ota.RadioSettings(910.525, 62.5, 7, 5, False)
+
+    def command(self, name, command, **options):
+        self.calls.append((name, command, options))
+        if name in self.fail_reads and command == "ota config":
+            raise ota.TransmissionError("lost config read")
+        if (name, command) in self.overrides:
+            return self.overrides[name, command]
+        if command == "ota folder off":
+            return "OK"
+        if command == "ota config":
+            return f"ota config: autofetch=off autoinstall=off hops={self.hops[name]} keys=0"
+        if command.startswith("ota config hops "):
+            if name == "source" and self.seeding:
+                raise AssertionError("source text CLI used while seeder owns serial")
+            value = int(command.rsplit(" ", 1)[1])
+            self.hops[name] = value
+            self.events.append(f"set:{name}:{value}")
+            if (name, value) in self.lost:
+                self.lost.remove((name, value))
+                raise ota.TransmissionError("lost setting acknowledgement")
+            return f"OK OTA reach = {value} hops (saved)" + (" - direct only" if value == 0 else "")
+        raise AssertionError((name, command))
+
+    def remote_command(self, name, command, **options):
+        return self.command(name, command, **options)
+
+    def source_cli(self, _args, command, **options):
+        return self.command("source", command, **options)
+
+    def get_radio(self):
+        return self.normal
+
+    def set_radio(self, _radio, _label):
+        self.events.append("controller-restore")
+
+    def close(self):
+        pass
+
+
+def arguments(relays=1, minimum=None):
+    return argparse.Namespace(
+        target="remote", relay_values=[(f"relay{i}", "relay-secret") for i in range(relays)],
+        ota_hops=minimum, source_serial="/dev/source", source_cli_serial=None,
+        source_cli_tcp=None, source_contact_value="source",
+    )
+
+
+class OtaHopTests(unittest.TestCase):
+    def setUp(self):
+        self.output = contextlib.redirect_stdout(io.StringIO())
+        self.output.__enter__()
+        self.addCleanup(self.output.__exit__, None, None, None)
+        self.sleep = mock.patch.object(ota.time, "sleep")
+        self.sleep.start()
+        self.addCleanup(self.sleep.stop)
+
+    def test_config_parser_accepts_legacy_and_terminal_banners(self):
+        for hops in range(9):
+            reply = f"ota config: cache=off/0 autofetch=off advert=1440min hops={hops} keys=0"
+            self.assertEqual(ota.parse_ota_hops(reply, "station"), hops)
+            self.assertEqual(ota.parse_ota_hops(f"WELCOME\n> ota config\n> {reply}\r\n> ", "station"), hops)
+
+    def test_parser_rejects_missing_ambiguous_malformed_and_error_values(self):
+        for reply in ("", "ERR permission denied", "Unknown command", "OTA | hops=0",
+                      "ota config: hops=9", "ota config: hops=-1", "ota config: hops=1foo",
+                      "ota config: hops=1 hops=1", "ota config: advert=1440",
+                      "ota config: hops=1\nota config: hops=2"):
+            with self.subTest(reply=reply), self.assertRaises(ota.OtaError):
+                ota.parse_ota_hops(reply, "station")
+
+    def test_command_reply_matching_does_not_accept_unrelated_ok(self):
+        self.assertTrue(ota.reply_matches_command("ota config", "ota config: hops=0"))
+        self.assertTrue(ota.reply_matches_command("ota config", "LoRa OTA not included in this build"))
+        self.assertTrue(ota.reply_matches_command("ota config", "LoRa OTA is not included in this build; tempradio cannot enable it."))
+        self.assertTrue(ota.reply_matches_command("ota config hops 1", "OK OTA reach = 1 hop (saved)"))
+        self.assertFalse(ota.reply_matches_command("ota config hops 1", "OK - temp params for 1 mins"))
+
+    def test_direct_run_has_no_policy_queries_or_writes(self):
+        stations = Stations(0)
+        with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+            self.assertEqual(ota.prepare_ota_hop_settings(stations, arguments(0)), [])
+        self.assertEqual(stations.calls, [])
+
+    def test_infers_route_count_preserves_higher_limits_and_restores(self):
+        for count in (1, 2, 3, 8):
+            with self.subTest(relays=count), tempfile.TemporaryDirectory() as directory:
+                stations = Stations(count)
+                stations.hops["relay0"] = 8
+                original = stations.hops.copy()
+                args = arguments(count)
+                owned = []
+                with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                    settings = ota.prepare_ota_hop_settings(stations, args)
+                    self.assertTrue(all(s.minimum == count for s in settings))
+                    ota.apply_ota_hop_settings(stations, args, settings, owned, Path(directory))
+                    self.assertEqual(stations.hops["remote"], count)
+                    self.assertEqual(stations.hops["source"], count)
+                    self.assertEqual(stations.hops["relay0"], 8)
+                    ota.restore_ota_hop_settings(stations, args, owned)
+                self.assertEqual(stations.hops, original)
+                self.assertEqual(owned, [])
+                self.assertFalse(any(name == "relay0" and cmd.startswith("ota config hops ")
+                                     for name, cmd, _ in stations.calls))
+                remote_writes = [options for name, cmd, options in stations.calls
+                                 if name != "source" and cmd.startswith("ota config hops ")]
+                self.assertTrue(all(options["retry"] is False for options in remote_writes))
+                relay_options = [options for name, _cmd, options in stations.calls if name.startswith("relay")]
+                self.assertTrue(all(options["password"] == "relay-secret" for options in relay_options))
+
+    def test_explicit_minimum_handles_additional_undeclared_hops(self):
+        stations = Stations(0)
+        with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+            settings = ota.prepare_ota_hop_settings(stations, arguments(0, 3))
+        self.assertEqual([s.transfer for s in settings], [3, 3])
+
+    def test_higher_limits_produce_no_recovery_file_or_changes(self):
+        stations = Stations(source_hops=4, target_hops=3)
+        stations.hops["relay0"] = 2
+        with tempfile.TemporaryDirectory() as directory:
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                ota.apply_ota_hop_settings(stations, arguments(), settings, [], Path(directory))
+            self.assertFalse((Path(directory) / ota.OTA_HOPS_RECOVERY_FILE).exists())
+        self.assertFalse(stations.events)
+
+    def test_unknown_source_or_relay_stops_before_any_setting_change(self):
+        for name in ("source", "remote", "relay0"):
+            for reply in ("Unknown command", "ERR permission denied", "ota config: hops=9", ""):
+                with self.subTest(name=name, reply=reply):
+                    stations = Stations()
+                    stations.overrides[name, "ota config"] = reply
+                    with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                        with self.assertRaises(ota.OtaError):
+                            ota.prepare_ota_hop_settings(stations, arguments())
+                    self.assertEqual(stations.events, [])
+
+    def test_explicit_non_ota_relay_is_left_opaque(self):
+        stations = Stations()
+        stations.overrides["relay0", "ota config"] = "LoRa OTA is not included in this build; tempradio cannot enable it. Use an OTA-enabled repeater firmware"
+        with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+            settings = ota.prepare_ota_hop_settings(stations, arguments())
+        self.assertEqual([s.name for s in settings], ["source", "remote"])
+
+    def test_setter_reply_lost_without_delivery_stops_without_replaying(self):
+        stations = Stations()
+        saved = ota.OtaHopSettings("remote", None, False, 0, 1)
+        with mock.patch.object(stations, "remote_command", side_effect=[
+            ota.TransmissionError("not delivered"), "ota config: hops=0",
+        ]) as command:
+            with self.assertRaisesRegex(ota.OtaError, "did not read back"):
+                ota.set_ota_hops_verified(stations, arguments(), saved, 1)
+        self.assertEqual([c.args[1] for c in command.call_args_list], ["ota config hops 1", "ota config"])
+
+    def test_recovery_write_failure_prevents_any_mutation(self):
+        stations = Stations()
+        owned = []
+        with tempfile.TemporaryDirectory() as directory:
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                with mock.patch.object(ota, "write_private_recovery_file", side_effect=OSError("disk full")):
+                    with self.assertRaises(OSError):
+                        ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory))
+        self.assertFalse(stations.events)
+        self.assertEqual(owned, [])
+
+    def test_source_serial_ack_can_include_a_welcome_banner(self):
+        stations = Stations()
+        saved = ota.OtaHopSettings("source", None, True, 0, 1)
+        with mock.patch.object(ota, "source_cli_command", side_effect=[
+            "WELCOME\n> ota config hops 1\nOK OTA reach = 1 hop (saved)\n> ",
+            "WELCOME\nota config: hops=1\n> ",
+        ]):
+            ota.set_ota_hops_verified(stations, arguments(), saved, 1)
+
+    def test_unmanaged_source_and_excessive_route_are_rejected(self):
+        for args in (arguments(9), arguments()):
+            if len(args.relay_values) == 1:
+                args.source_serial = None
+            stations = Stations()
+            with self.assertRaises(ota.OtaError):
+                ota.prepare_ota_hop_settings(stations, args)
+            self.assertEqual(stations.calls, [])
+
+    def test_timeout_has_finite_read_retries_and_never_means_zero(self):
+        stations = Stations()
+        stations.fail_reads.add("remote")
+        with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+            with self.assertRaises(ota.TransmissionError):
+                ota.prepare_ota_hop_settings(stations, arguments())
+        self.assertEqual(len([c for c in stations.calls if c[0] == "remote"]), ota.TRANSMISSION_RETRY_LIMIT + 1)
+        self.assertEqual(stations.events, [])
+
+    def test_external_change_during_rehearsal_aborts_all_writes(self):
+        stations = Stations()
+        with tempfile.TemporaryDirectory() as directory:
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                stations.hops["relay0"] = 4
+                with self.assertRaisesRegex(ota.OtaError, "changed during preparation"):
+                    ota.apply_ota_hop_settings(stations, arguments(), settings, [], Path(directory))
+            self.assertFalse((Path(directory) / ota.OTA_HOPS_RECOVERY_FILE).exists())
+        self.assertEqual(stations.events, [])
+
+    def test_recovery_is_private_password_free_and_durable_before_writes(self):
+        stations = Stations()
+        args = arguments()
+        owned = []
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            original_command = stations.command
+
+            def command(name, text, **options):
+                if text.startswith("ota config hops "):
+                    path = root / ota.OTA_HOPS_RECOVERY_FILE
+                    self.assertTrue(path.is_file())
+                    self.assertTrue(any(saved.name == name for saved in owned))
+                    raw = path.read_text()
+                    self.assertNotIn("relay-secret", raw)
+                    self.assertIn("restore_command", raw)
+                    if os.name != "nt":
+                        self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+                return original_command(name, text, **options)
+
+            with mock.patch.object(stations, "command", side_effect=command):
+                with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                    settings = ota.prepare_ota_hop_settings(stations, args)
+                    ota.apply_ota_hop_settings(stations, args, settings, owned, root)
+            saved = json.loads((root / ota.OTA_HOPS_RECOVERY_FILE).read_text())
+            self.assertEqual(saved["source_endpoint"]["serial"], "/dev/source")
+
+    def test_lost_set_and_restore_acknowledgements_use_readback_without_replay(self):
+        stations = Stations()
+        stations.lost = {("source", 1), ("remote", 1), ("relay0", 1),
+                         ("source", 0), ("remote", 0), ("relay0", 0)}
+        with tempfile.TemporaryDirectory() as directory:
+            owned = []
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory))
+                ota.restore_ota_hop_settings(stations, arguments(), owned)
+        self.assertEqual(owned, [])
+        self.assertEqual(len(stations.events), 6)
+        self.assertTrue(all(value == 0 for value in stations.hops.values()))
+
+    def test_bad_readback_is_not_success_and_partial_changes_are_restored(self):
+        stations = Stations()
+        stations.overrides["remote", "ota config hops 1"] = "OK OTA reach = 1 hop (saved)"
+        with tempfile.TemporaryDirectory() as directory:
+            owned = []
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                with self.assertRaisesRegex(ota.OtaError, "did not read back"):
+                    ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory))
+                self.assertEqual([s.name for s in owned], ["source", "remote"])
+                ota.restore_ota_hop_settings(stations, arguments(), owned)
+        self.assertEqual(owned, [])
+        self.assertEqual(stations.hops["source"], 0)
+
+    def test_restore_attempts_all_nodes_and_preserves_external_changes(self):
+        stations = Stations()
+        with tempfile.TemporaryDirectory() as directory:
+            owned = []
+            with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli):
+                settings = ota.prepare_ota_hop_settings(stations, arguments())
+                ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory))
+                stations.hops["relay0"] = 4
+                with self.assertRaisesRegex(ota.OtaError, "changed externally"):
+                    ota.restore_ota_hop_settings(stations, arguments(), owned)
+                self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 4})
+                self.assertEqual([s.name for s in owned], ["relay0"])
+
+    def test_cli_minimum_must_cover_declared_route_and_stay_in_range(self):
+        parser = ota.build_parser()
+        for extra in (["--ota-hops", "-1"], ["--ota-hops", "9"],
+                      ["--relay", "relay", "--ota-hops", "0"],
+                      ["--relay", "relay"] * 9):
+            with self.subTest(extra=extra), contextlib.redirect_stderr(io.StringIO()):
+                args = parser.parse_args(["release.mota", "remote", *extra])
+                with self.assertRaises(SystemExit):
+                    ota.validate_args(args, parser)
+
+
+class OtaHopLifecycleTests(unittest.TestCase):
+    def run_main(self, directory, *, failure=None, leave=False, no_install=True):
+        stations = Stations()
+        image = firmware(b"ota hop lifecycle" * 300, VERSION_NEW)
+        package = ota.parse_mota(mota_blob(image))
+        seeder = mock.Mock()
+
+        def start():
+            stations.seeding = True
+            stations.events.append("seeder-start")
+
+        def stop():
+            stations.seeding = False
+            stations.events.append("seeder-stop")
+
+        seeder.start.side_effect = start
+        seeder.stop.side_effect = stop
+
+        def phase(name):
+            def action(*_args, **_kwargs):
+                stations.events.append(name)
+                if failure == name:
+                    if name == "interrupt":
+                        raise KeyboardInterrupt
+                    raise ota.OtaError("synthetic " + name)
+            return action
+
+        def pull(*_args):
+            stations.events.append("pull")
+            if failure == "interrupt":
+                raise KeyboardInterrupt
+            if failure == "pull":
+                raise ota.OtaError("synthetic discovery failure")
+            self.assertTrue(all(hops == 1 for hops in stations.hops.values()))
+
+        def install(*_args):
+            stations.events.append("install")
+            self.assertTrue(all(hops == 0 for hops in stations.hops.values()))
+            if failure == "install":
+                raise ota.OtaError("synthetic install failure")
+            return True
+
+        argv = ["release.mota", "remote", "--controller-serial", "/dev/controller",
+                "--source-serial", "/dev/source", "--password", "secret",
+                "--relay", "relay0=relay-secret", "--work-dir", str(Path(directory) / "work"),
+                "--yes"]
+        if no_install:
+            argv.append("--no-install")
+        if leave:
+            argv.append("--leave-controller-radio")
+        with contextlib.ExitStack() as stack:
+            for name in ("preflight_inputs", "preflight_source_cli", "bind_contact_selectors",
+                         "verify_shared_source_identity", "ensure_controller_clock_safe",
+                         "restore_relay_timings", "shorten_relay_temp_windows",
+                         "shorten_target_temp_window", "enforce_relay_timing", "report_staged_update",
+                         "wait_for_post_install_identity", "verify_installed"):
+                stack.enter_context(mock.patch.object(ota, name))
+            stack.enter_context(mock.patch.object(ota, "ensure_source_clock_gate_safe", return_value=(1800000000, 1800000059)))
+            stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=ota.RxpsSettings(False, 100, 100)))
+            stack.enter_context(mock.patch.object(ota, "disable_source_rxps", return_value=False))
+            stack.enter_context(mock.patch.object(ota, "query_target", return_value=target()))
+            stack.enter_context(mock.patch.object(ota, "prepare_package", return_value=(Path("release.mota"), package, None)))
+            stack.enter_context(mock.patch.object(ota, "read_lora_ota_participant_versions", return_value={}))
+            stack.enter_context(mock.patch.object(ota, "read_remote_rxps", return_value=ota.RxpsSettings(False, 100, 100)))
+            stack.enter_context(mock.patch.object(ota, "confirm_update", side_effect=phase("confirm")))
+            stack.enter_context(mock.patch.object(ota, "run_temp_radio_preflight", side_effect=phase("rehearse")))
+            stack.enter_context(mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli))
+            stack.enter_context(mock.patch.object(ota, "arm_target_temp_radio", side_effect=phase("arm-target")))
+            stack.enter_context(mock.patch.object(ota, "arm_relay_temp_radio_once", return_value=True))
+            stack.enter_context(mock.patch.object(ota, "arm_source_temp_radio_once", return_value=True))
+            stack.enter_context(mock.patch.object(ota, "switch_controller_to_temp_radio", side_effect=phase("switch")))
+            stack.enter_context(mock.patch.object(ota, "read_relay_timing", return_value=ota.RelayTimingSettings("relay0", "relay-secret", 0, 0.3)))
+            stack.enter_context(mock.patch.object(ota, "read_remote_public_key_bounded", return_value="a" * 64))
+            stack.enter_context(mock.patch.object(ota, "shorten_source_temp_window", return_value=True))
+            stack.enter_context(mock.patch.object(ota, "SeederProcess", return_value=seeder))
+            stack.enter_context(mock.patch.object(ota, "find_and_start_pull", side_effect=pull))
+            stack.enter_context(mock.patch.object(ota, "monitor_download", side_effect=phase("monitor")))
+            stack.enter_context(mock.patch.object(ota, "request_install", side_effect=install))
+            stack.enter_context(mock.patch.object(ota.time, "sleep"))
+            stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+            stack.enter_context(contextlib.redirect_stderr(io.StringIO()))
+            # The production source name is a bound public key; this test uses
+            # the descriptive fallback while the source CLI fixture is local.
+            original_prepare = ota.prepare_ota_hop_settings
+
+            def prepare(controller, args):
+                args.source_contact_value = "source"
+                return original_prepare(controller, args)
+
+            stack.enter_context(mock.patch.object(ota, "prepare_ota_hop_settings", side_effect=prepare))
+            result = ota.main(argv, controller_override=stations)
+        return result, stations
+
+    def test_success_restores_after_seeder_stop_even_when_temp_radio_is_preserved(self):
+        for leave in (False, True):
+            with self.subTest(leave=leave), tempfile.TemporaryDirectory() as directory:
+                result, stations = self.run_main(directory, leave=leave)
+                self.assertEqual(result, 0)
+                self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0})
+                events = stations.events
+                self.assertLess(events.index("confirm"), events.index("rehearse"))
+                self.assertLess(events.index("rehearse"), events.index("set:source:1"))
+                self.assertLess(events.index("set:remote:1"), events.index("arm-target"))
+                self.assertLess(events.index("seeder-stop"), events.index("set:source:0"))
+
+    def test_rehearsal_and_confirmation_failures_never_write_hop_settings(self):
+        for failure in ("confirm", "rehearse"):
+            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory:
+                result, stations = self.run_main(directory, failure=failure)
+                self.assertEqual(result, 2)
+                self.assertFalse(any(event.startswith("set:") for event in stations.events))
+
+    def test_discovery_failure_and_interrupt_restore_all_original_settings(self):
+        for failure in ("pull", "interrupt", "arm-target"):
+            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory:
+                result, stations = self.run_main(directory, failure=failure)
+                self.assertEqual(result, 130 if failure == "interrupt" else 2)
+                self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0})
+
+    def test_install_runs_only_after_policies_are_restored(self):
+        for failure in (None, "install"):
+            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory:
+                result, stations = self.run_main(directory, failure=failure, no_install=False)
+                self.assertEqual(result, 2 if failure else 0)
+                self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0})
+                self.assertLess(stations.events.index("set:source:0"), stations.events.index("install"))
+
+
+if __name__ == "__main__":
+    unittest.main()