Fix relayed OTA retries and add guarded legacy recovery

Use a bounded timed OTA forwarding cache while preserving echo, queue, TempRadio and forwarding-policy guards. Add opt-in manifest recovery for directly reachable legacy relays and interactive 125 kHz SF5/SF6/SF7 rehearsals, with regression tests and documentation.

tools/lora_ota/lora_ota.py

@@ -146,6 +146,11 @@ 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
+# The deployed SimpleMeshTables FIFO holds 160 hashes. These short, direct
+# (never forwarded) unknown-OTA frames provide 20% loss headroom. This is an
+# explicit, best-effort legacy workaround, not an acknowledgement of reception.
+LEGACY_OTA_CACHE_FRAMES = 192
+LEGACY_OTA_CACHE_ROUNDS = 2
 TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json"
 SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json"
 MIN_MESHCLI_VERSION = (1, 6, 0)
@@ -190,6 +195,10 @@ class TransmissionError(OtaError):
     """A command may not have reached its destination or returned a reply."""
 
 
+class TempRadioPathUnproven(TransmissionError):
+    """The temporary path failed, but its short leases returned cleanly."""
+
+
 class TransmissionStopped(OtaError):
     """The operator chose to stop an otherwise continuing retry loop."""
 
@@ -2116,6 +2125,19 @@ class Controller:
                 )
         raise OtaError("meshcli did not return the controller radio settings")
 
+    def send_legacy_ota_cache_probe(self, packet: bytes, timeout: float) -> None:
+        marker = f"OTA_CACHE_{secrets.token_hex(8)}"
+        result = self._execute(
+            ["send_raw", packet.hex(), "echo", marker],
+            "legacy OTA cache turnover", timeout=timeout,
+        )
+        # meshcli 1.6.3 prints no JSON on send_raw success, only on failure.
+        # A marker proves the command completed; it is NOT an on-air ACK.
+        output = result.stdout + "\n" + result.stderr
+        if (result.returncode != 0 or marker not in result.stdout
+                or re.search(r"error|unknown command|unsupported|invalid", output, re.I)):
+            raise OtaError("legacy OTA cache probe was not accepted by the companion")
+
     def get_public_key(self, timeout: float | None = None) -> str:
         objects = (
             self._run(["infos"], "read controller identity")
@@ -3042,6 +3064,82 @@ def warn_short_temp_radio_window(
         )
 
 
+def prompt_for_125khz_temp_radio(
+    args: argparse.Namespace, *, after_failed_rehearsal: bool = False,
+) -> bool:
+    """Offer the interactive 125-kHz SF5-to-SF7 path rehearsal ladder."""
+    if (
+        args.temp_values[1] != 250
+        or args.prepare_only
+        or args.yes
+        or not sys.stdin.isatty()
+    ):
+        return False
+    if after_failed_rehearsal:
+        question = (
+            "The 250 kHz rehearsal could not prove the temporary path, but "
+            "all short leases expired and the normal path was verified. "
+            "Try 125 kHz at SF5, then SF6 and SF7 if needed, with a fresh "
+            "three-minute rehearsal and confirmation at each step? [y/N] "
+        )
+    else:
+        question = (
+            "Try TempRadio at 125 kHz SF5, then SF6 and SF7 if needed "
+            "(confirming each setting before rehearsal and transfer)? [y/N] "
+        )
+    try:
+        approved = input(question).strip().lower() in ("y", "yes")
+    except EOFError:
+        return False
+    if not approved:
+        return False
+    freq, _bandwidth, _sf, cr, minutes = args.temp_values
+    args.temp_values = (freq, 125.0, 5, cr, minutes)
+    args.temp_radio = (
+        f"{format_decimal(freq)},125,5,{cr},{minutes}"
+    )
+    print(
+        f"[preflight] selected {args.temp_radio}; the transfer window remains "
+        f"{minutes} minutes"
+    )
+    warn_short_temp_radio_window(args, include_install=not args.no_install)
+    return True
+
+
+def prompt_for_next_125khz_sf(args: argparse.Namespace) -> bool:
+    """Advance one SF only after a cleanly returned failed path rehearsal."""
+    freq, bandwidth, sf, cr, minutes = args.temp_values
+    if (
+        bandwidth != 125
+        or sf not in (5, 6)
+        or args.prepare_only
+        or args.yes
+        or not sys.stdin.isatty()
+    ):
+        return False
+    next_sf = sf + 1
+    try:
+        approved = input(
+            f"125 kHz SF{sf} could not prove the temporary path. All short "
+            f"leases expired and the normal path was verified. Try a fresh "
+            f"three-minute rehearsal at SF{next_sf}? [y/N] "
+        ).strip().lower() in ("y", "yes")
+    except EOFError:
+        return False
+    if not approved:
+        return False
+    args.temp_values = (freq, bandwidth, next_sf, cr, minutes)
+    args.temp_radio = (
+        f"{format_decimal(freq)},125,{next_sf},{cr},{minutes}"
+    )
+    print(
+        f"[preflight] selected {args.temp_radio}; the transfer window remains "
+        f"{minutes} minutes"
+    )
+    warn_short_temp_radio_window(args, include_install=not args.no_install)
+    return True
+
+
 def parse_source_terminal_banner(value: str) -> tuple[str, str]:
     """Read one Full/Companion welcome identity, not an advert or cached name."""
     text = value.replace("\r", "")
@@ -4662,6 +4760,8 @@ def confirm_update(
         print(f"  boot target : {package.target_id:08X}; stage only, explicit install required")
         print("  boot safety : destination rechecks trust, continuity and upgrade-only policy at manual install")
     print(f"  TempRadio   : {args.temp_radio}")
+    if getattr(args, "legacy_relay_dedup_workaround", False):
+        print("  legacy retry: up to 2 x 192 short, non-forwarded TempRadio frames on manifest timeout")
     saved_rxps = getattr(args, "target_rxps_saved", None)
     rxps_profile = getattr(args, "target_rxps_profile", None)
     if isinstance(saved_rxps, RxpsSettings) and saved_rxps.enabled:
@@ -5004,6 +5104,90 @@ def find_and_start_pull(
     )
 
 
+def recover_legacy_manifest_timeout(
+    controller: Controller,
+    args: argparse.Namespace,
+    package: MotaInfo,
+    seeder: SeederProcess | None,
+    status: str,
+    deadline: float,
+    attempt: int,
+) -> bool:
+    """Bounded, opt-in turnover of a directly reachable legacy relay's FIFO.
+
+    Only restart this exact MID after a confirmed manifest timeout with no
+    blocks. Never clear a partial/ready image, reboot a node, change a package,
+    or infer that an unacknowledged LoRa packet actually arrived.
+    """
+    if (not getattr(args, "legacy_relay_dedup_workaround", False)
+            or attempt >= LEGACY_OTA_CACHE_ROUNDS
+            or download_manifest_id(status) != package.manifest_id
+            or not re.search(r"download: failed \(manifest timeout\) 0/0\b", status, re.I)):
+        return False
+    relays = getattr(args, "relay_values", [])
+    if len(relays) != 1 or not getattr(args, "source_shares_controller", False):
+        raise OtaError("legacy cache workaround requires one relay and a shared source/controller")
+    relay, _password = relays[0]
+    contacts = controller._run(["contact_info", relay], "check direct legacy relay route", timeout=5)
+    matching = [c for c in contacts if contact_matches_selector(c, relay)]
+    if (len(matching) != 1 or matching[0].get("out_path_len") != 0
+            or not re.fullmatch(r"[0-9a-fA-F]{64}", relay)):
+        raise OtaError("legacy cache workaround requires the relay's full key and a direct (0-hop) contact route")
+    expected = RadioSettings(*args.temp_values[:4], False)
+    frame_prefix = bytes((0x32, 0, 0))  # OTA (0x0c), DIRECT, no path, unknown opcode 0
+    nonce = secrets.token_bytes(8)
+    spacing = max(0.25, 4 * lora_airtime_seconds(
+        len(frame_prefix) + len(nonce) + 2, expected.bandwidth,
+        expected.spreading_factor, expected.coding_rate, preamble_symbols=32,
+    ))
+    print(f"[legacy] manifest timeout: bounded cache turnover {attempt + 1}/{LEGACY_OTA_CACHE_ROUNDS}; "
+          f"{LEGACY_OTA_CACHE_FRAMES} short direct frames, no relay reboot or package changes")
+    for index in range(LEGACY_OTA_CACHE_FRAMES):
+        ensure_seeder_running(seeder, "during legacy manifest recovery")
+        if index % 16 == 0:
+            # Use the existing binary controller connection, not the serial
+            # source that motatool owns. Fail closed on missing lease telemetry.
+            objects = controller._run(["cli", "tempradio"], "check legacy recovery TempRadio", timeout=5)
+            replies = [o["text"] for o in objects if isinstance(o.get("text"), str)]
+            if len(replies) != 1:
+                raise OtaError("cannot prove local TempRadio lease for legacy recovery")
+            state = parse_source_temp_radio_status(replies[0])
+            remaining = re.search(r"\b(\d+)s left\b", replies[0])
+            needed = 16 * (5 + spacing) + 10
+            if (state is None or state[0] != "active" or not expected.matches(state[1])
+                    or remaining is None or int(remaining[1]) < needed
+                    or deadline - time.monotonic() < needed):
+                raise OtaError("insufficient verified TempRadio/transfer time for legacy recovery")
+        controller.send_legacy_ota_cache_probe(
+            frame_prefix + nonce + struct.pack("<H", index),
+            timeout=min(5.0, max(0.1, deadline - time.monotonic())),
+        )
+        time.sleep(spacing)
+    # Radio admission is not delivery. Re-read the exact session before any
+    # restart; an externally changed or completed session is never discarded.
+    current = remote_command_with_seeder(controller, args.target, "ota status", seeder,
+                                         "after legacy cache turnover")
+    if download_manifest_id(current) != package.manifest_id:
+        raise OtaError("destination session changed during legacy recovery")
+    if not re.search(r"download: failed \(manifest timeout\) 0/0\b", current, re.I):
+        require_package_session(current, package)
+        return True
+    if time.monotonic() >= deadline:
+        raise OtaError("transfer deadline expired during legacy recovery; no pull resent")
+    try:
+        reply = controller.remote_command(args.target,
+                                          f"ota pull {package.manifest_id} flash", retry=False)
+    except TransmissionError:
+        reply = remote_command_with_seeder(controller, args.target, "ota status", seeder,
+                                           "resolving legacy recovery pull")
+        require_package_session(reply, package)
+    else:
+        if not re.match(rf"OK (?:pulling|resuming) mid={package.manifest_id}\b", reply, re.I):
+            raise OtaError(f"legacy recovery pull was not confirmed: {reply}")
+    print("[legacy] retried the same manifest; normal verification and transfer deadline remain in force")
+    return True
+
+
 def monitor_download(
     controller: Controller,
     args: argparse.Namespace,
@@ -5011,6 +5195,7 @@ def monitor_download(
     seeder: SeederProcess | None = None,
 ) -> str:
     deadline = time.monotonic() + args.transfer_timeout_minutes * 60
+    legacy_recoveries = 0
     poll_seconds = float(args.poll_seconds)
     # A status command is ordinary half-duplex LoRa traffic. Asking immediately
     # after `ota pull` can occupy the link for a full reply timeout and was
@@ -5065,6 +5250,12 @@ def monitor_download(
                     controller, args.target, "ota status", seeder,
                     "during transfer",
                 )
+                if recover_legacy_manifest_timeout(
+                    controller, args, package, seeder, status, deadline, legacy_recoveries
+                ):
+                    legacy_recoveries += 1
+                    last_progress = time.monotonic()
+                    continue
                 require_package_session(status, package)
                 if "ready to install" in status.lower():
                     return status
@@ -5088,6 +5279,12 @@ def monitor_download(
             controller, args.target, "ota status", seeder, "during transfer"
         )
         query_seconds = time.monotonic() - query_started
+        if recover_legacy_manifest_timeout(
+            controller, args, package, seeder, status, deadline, legacy_recoveries
+        ):
+            legacy_recoveries += 1
+            time.sleep(min(first_wait, max(0.0, deadline - time.monotonic())))
+            continue
         require_package_session(status, package)
         lowered = status.lower()
         if "ready to install" in lowered:
@@ -5855,6 +6052,7 @@ def run_temp_radio_preflight(
     controller_may_need_restore = False
     natural_expiry_waited = False
     normal_proven = False
+    normal_return_needed_recovery = False
     earliest_lease_expiry: float | None = None
     latest_lease_expiry: float | None = None
     primary_error: BaseException | None = None
@@ -6513,7 +6711,12 @@ def run_temp_radio_preflight(
             )
         prove_controller_identity("on TempRadio", deadline=proof_deadline)
         prove_source_identity("on TempRadio", deadline=proof_deadline)
-        prove_remote_identities("on TempRadio", deadline=proof_deadline)
+        try:
+            prove_remote_identities("on TempRadio", deadline=proof_deadline)
+        except TransmissionError as exc:
+            # The caller may offer a narrower bandwidth only after this
+            # rehearsal's finally block owns expiry and proves normal return.
+            raise TempRadioPathUnproven(str(exc)) from exc
         temp_proof_deadline("temporary path proof completion")
 
         if not shared_controller:
@@ -6587,6 +6790,7 @@ def run_temp_radio_preflight(
                 except (OtaError, OSError) as exc:
                     first_normal_error = exc
                 if first_normal_error is not None:
+                    normal_return_needed_recovery = True
                     recovery_error: BaseException | None = None
                     try:
                         recover_participants_on_temp()
@@ -6633,6 +6837,13 @@ def run_temp_radio_preflight(
                 f"three-minute TempRadio rehearsal failed: {primary_error}; "
                 "cleanup also failed: " + "; ".join(cleanup_errors)
             ) from primary_error
+        if isinstance(primary_error, TempRadioPathUnproven) and (
+            not normal_proven or normal_return_needed_recovery
+        ):
+            raise OtaError(
+                "temporary path proof failed and normal return required "
+                "recovery; no in-run bandwidth retry is safe"
+            ) from primary_error
         raise primary_error
     if cleanup_errors:
         raise OtaError(
@@ -7395,6 +7606,12 @@ def build_parser() -> argparse.ArgumentParser:
         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(
+        "--legacy-relay-dedup-workaround", action="store_true",
+        help=("on exact 0/0 manifest timeout, try up to two 192-frame direct cache turnovers; "
+              "requires one directly reachable relay and --source-shares-controller; "
+              "adds airtime, never reboots, does not fix legacy block retries"),
+    )
     parser.add_argument(
         "--source-contact",
         metavar="NAME_OR_KEY",
@@ -7588,6 +7805,12 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
         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")
+    if args.legacy_relay_dedup_workaround and (
+        len(args.relay) != 1 or not args.source_shares_controller
+        or args.ota_hops not in (None, 1)
+    ):
+        parser.error("--legacy-relay-dedup-workaround requires one --relay, "
+                     "--source-shares-controller, and no additional OTA hops")
     clear_manifest = args.clear_completed_manifest
     clear_body_hash = args.clear_completed_on_body_hash
     if bool(clear_manifest) != bool(clear_body_hash):
@@ -8664,6 +8887,23 @@ def main(
             or not 1 <= target_rxps_saved.level <= 10
         ):
             target_rxps_profile = None
+        # Let an interactive operator choose the narrower tuple before the
+        # update plan and any TempRadio lease are committed.
+        if prompt_for_125khz_temp_radio(args):
+            temp_command = f"tempradio {args.temp_radio}"
+            target_rxps_profile = select_rxps_temp_profile(
+                target.current_version,
+                args.temp_values,
+                all_participants_support_adaptive_preamble=(
+                    all_participants_support_adaptive_preamble
+                ),
+            )
+            if (
+                target_rxps_saved is None
+                or target_rxps_saved.level is None
+                or not 1 <= target_rxps_saved.level <= 10
+            ):
+                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)
@@ -8678,13 +8918,44 @@ def main(
         # long TempRadio window, relay timing, seeding, pull, or install can be
         # mutated. The rehearsal owns only independent three-minute leases and
         # proves their natural return before handing control back to this run.
-        run_temp_radio_preflight(
-            controller,
-            args,
-            target,
-            original_radio,
-            temp_radio,
-        )
+        while True:
+            try:
+                run_temp_radio_preflight(
+                    controller, args, target, original_radio, temp_radio,
+                )
+                break
+            except TempRadioPathUnproven:
+                # The typed failure is emitted only after short-lease expiry
+                # and an unrecovered exact normal-path proof. Never reuse a
+                # live lease or silently change the approved radio plan.
+                if args.temp_values[1] == 250:
+                    retry = prompt_for_125khz_temp_radio(
+                        args, after_failed_rehearsal=True,
+                    )
+                else:
+                    retry = prompt_for_next_125khz_sf(args)
+                if not retry:
+                    raise
+                temp_command = f"tempradio {args.temp_radio}"
+                target_rxps_profile = select_rxps_temp_profile(
+                    target.current_version,
+                    args.temp_values,
+                    all_participants_support_adaptive_preamble=(
+                        all_participants_support_adaptive_preamble
+                    ),
+                )
+                if (
+                    target_rxps_saved is None
+                    or target_rxps_saved.level is None
+                    or not 1 <= target_rxps_saved.level <= 10
+                ):
+                    target_rxps_profile = None
+                args.target_rxps_profile = target_rxps_profile
+                freq, bandwidth, sf, cr, _minutes = args.temp_values
+                temp_radio = RadioSettings(
+                    freq, bandwidth, sf, cr, original_radio.repeat
+                )
+                confirm_update(args, target, package)
         apply_ota_hop_settings(
             controller, args, args.ota_hop_settings, owned_ota_hop_settings, work_dir,
         )

tools/lora_ota/test_lora_ota.py

@@ -3673,6 +3673,12 @@ class TempRadioPreflightTests(unittest.TestCase):
                 return "OK - normal radio restore scheduled"
             if command == "get public.key":
                 key = TempRadioPreflightTests.NODE_KEYS[name]
+                if (
+                    self.failure == "temp_remote_unreachable"
+                    and on_temp
+                    and name == "remote"
+                ):
+                    raise ota.TransmissionError("no matching CLI reply")
                 if self.failure == "temp_remote_identity" and on_temp and name == "far":
                     return "> " + "44" * 32
                 if self.failure == "normal_remote_identity" and not on_temp and self.scheduled_temp and name == "remote":
@@ -4542,6 +4548,98 @@ class TempRadioPreflightTests(unittest.TestCase):
         pull.assert_not_called()
         install.assert_not_called()
 
+    def test_temporary_path_timeout_allows_retry_only_after_clean_return(self) -> None:
+        controller, _calls, clock, source_state, error = self.run_rehearsal(
+            failure="temp_remote_unreachable", capture_error=True,
+        )
+        self.assertIsInstance(error, ota.TempRadioPathUnproven)
+        self.assertGreaterEqual(
+            clock.now,
+            ota.TEMP_RADIO_PREFLIGHT_MINUTES * 60
+            + ota.TEMP_RADIO_PREFLIGHT_MARGIN_SECONDS,
+        )
+        self.assertFalse(bool(source_state["active"]))
+        self.assertTrue(controller.radio.matches(self.NORMAL))
+        self.assertTrue(controller.all_nodes_normal())
+
+    def test_operator_reconfirms_each_125khz_sf_before_rehearsal(self) -> None:
+        image = firmware(b"bandwidth retry" * 300, VERSION_NEW)
+        package = ota.parse_mota(mota_blob(image))
+        controller = mock.Mock()
+        controller.get_radio.return_value = self.NORMAL
+        controller.get_clock.return_value = int(ota.time.time()) + 1
+        saved = ota.RxpsSettings(False, 18205, 20423, 8, 16)
+        with tempfile.TemporaryDirectory() as directory:
+            argv = [
+                "release.mota", "remote",
+                "--controller-serial", "/dev/controller",
+                "--source-serial", "/dev/source",
+                "--password", "secret",
+                "--work-dir", str(Path(directory) / "work"),
+                "--temp-radio", "909.5,250,7,5,120",
+            ]
+            with contextlib.ExitStack() as stack:
+                stack.enter_context(mock.patch.object(ota, "preflight_inputs"))
+                stack.enter_context(mock.patch.object(ota, "bind_contact_selectors"))
+                stack.enter_context(mock.patch.object(ota, "preflight_source_cli"))
+                stack.enter_context(mock.patch.object(
+                    ota, "ensure_source_clock_gate_safe",
+                    return_value=(1_800_000_000, 1_800_000_059),
+                ))
+                stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=saved))
+                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={"destination": VERSION_NEW},
+                ))
+                stack.enter_context(mock.patch.object(
+                    ota, "read_remote_rxps",
+                    return_value=ota.RxpsSettings(False, 18205, 20423, 8, 16),
+                ))
+                confirm = stack.enter_context(mock.patch.object(ota, "confirm_update"))
+                rehearsal = stack.enter_context(mock.patch.object(
+                    ota, "run_temp_radio_preflight",
+                    side_effect=[
+                        ota.TempRadioPathUnproven("temporary path unavailable"),
+                        ota.TempRadioPathUnproven("SF5 path unavailable"),
+                        ota.TempRadioPathUnproven("SF6 path unavailable"),
+                        None,
+                    ],
+                ))
+                disable_source = stack.enter_context(mock.patch.object(ota, "disable_source_rxps"))
+                arm_target = stack.enter_context(mock.patch.object(
+                    ota, "arm_target_temp_radio",
+                    side_effect=ota.OtaError("stop before long lease"),
+                ))
+                stack.enter_context(mock.patch.object(
+                    ota, "switch_controller_to_temp_radio",
+                    side_effect=ota.OtaError("mock cleanup cannot switch"),
+                ))
+                stack.enter_context(mock.patch.object(
+                    ota.sys, "stdin", mock.Mock(isatty=lambda: True),
+                ))
+                prompts = stack.enter_context(mock.patch(
+                    "builtins.input", side_effect=["n", "y", "y", "y"],
+                ))
+                stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+                stack.enter_context(contextlib.redirect_stderr(io.StringIO()))
+                result = ota.main(argv, controller_override=controller)
+
+        self.assertEqual(result, 2)
+        self.assertEqual(prompts.call_count, 4)
+        self.assertEqual(confirm.call_count, 4)
+        self.assertEqual(
+            [(call.args[-1].bandwidth, call.args[-1].spreading_factor)
+             for call in rehearsal.call_args_list],
+            [(250.0, 7), (125.0, 5), (125.0, 6), (125.0, 7)],
+        )
+        self.assertEqual(arm_target.call_args.args[2], "tempradio 909.5,125,7,5,120")
+        disable_source.assert_called_once()
+
     def test_schedule_appearing_during_baseline_blocks_every_mutation(
         self,
     ) -> None:
@@ -4688,6 +4786,106 @@ class TempRadioPreflightTests(unittest.TestCase):
             )
 
 
+class TempRadioBandwidthPromptTests(unittest.TestCase):
+    @staticmethod
+    def args() -> argparse.Namespace:
+        return argparse.Namespace(
+            temp_values=(909.5, 250.0, 7, 5, 120),
+            temp_radio="909.5,250,7,5,120",
+            prepare_only=False,
+            yes=False,
+            no_install=True,
+        )
+
+    def test_operator_can_select_125_before_rehearsal(self) -> None:
+        args = self.args()
+        with (
+            mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)),
+            mock.patch("builtins.input", return_value="y") as prompt,
+            mock.patch.object(ota, "warn_short_temp_radio_window") as warning,
+        ):
+            self.assertTrue(ota.prompt_for_125khz_temp_radio(args))
+        self.assertEqual(args.temp_values, (909.5, 125.0, 5, 5, 120))
+        self.assertEqual(args.temp_radio, "909.5,125,5,5,120")
+        self.assertIn("SF5, then SF6 and SF7", prompt.call_args.args[0])
+        warning.assert_called_once_with(args, include_install=False)
+
+    def test_failed_rehearsal_prompt_requires_fresh_explicit_choice(self) -> None:
+        args = self.args()
+        with (
+            mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)),
+            mock.patch("builtins.input", return_value="yes") as prompt,
+            mock.patch.object(ota, "warn_short_temp_radio_window"),
+        ):
+            self.assertTrue(ota.prompt_for_125khz_temp_radio(
+                args, after_failed_rehearsal=True,
+            ))
+        self.assertIn("all short leases expired", prompt.call_args.args[0])
+        self.assertEqual(args.temp_values[1:3], (125.0, 5))
+        self.assertFalse(ota.prompt_for_125khz_temp_radio(args))
+
+    def test_125khz_ladder_stops_after_sf7(self) -> None:
+        args = self.args()
+        args.temp_values = (909.5, 125.0, 5, 5, 120)
+        args.temp_radio = "909.5,125,5,5,120"
+        with (
+            mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)),
+            mock.patch("builtins.input", side_effect=["y", "yes"]) as prompt,
+            mock.patch.object(ota, "warn_short_temp_radio_window") as warning,
+        ):
+            self.assertTrue(ota.prompt_for_next_125khz_sf(args))
+            self.assertEqual(args.temp_values[2], 6)
+            self.assertEqual(args.temp_radio, "909.5,125,6,5,120")
+            self.assertTrue(ota.prompt_for_next_125khz_sf(args))
+            self.assertEqual(args.temp_values[2], 7)
+            self.assertEqual(args.temp_radio, "909.5,125,7,5,120")
+            self.assertFalse(ota.prompt_for_next_125khz_sf(args))
+        self.assertEqual(prompt.call_count, 2)
+        self.assertEqual(warning.call_count, 2)
+
+    def test_125khz_ladder_requires_interactive_approval(self) -> None:
+        for is_tty, yes, answer in (
+            (False, False, "y"),
+            (True, True, "y"),
+            (True, False, "n"),
+        ):
+            with self.subTest(is_tty=is_tty, yes=yes, answer=answer):
+                args = self.args()
+                args.temp_values = (909.5, 125.0, 5, 5, 120)
+                args.temp_radio = "909.5,125,5,5,120"
+                args.yes = yes
+                with (
+                    mock.patch.object(
+                        ota.sys, "stdin", mock.Mock(isatty=lambda: is_tty),
+                    ),
+                    mock.patch("builtins.input", return_value=answer) as prompt,
+                ):
+                    self.assertFalse(ota.prompt_for_next_125khz_sf(args))
+                self.assertEqual(args.temp_values[2], 5)
+                if not is_tty or yes:
+                    prompt.assert_not_called()
+
+    def test_noninteractive_or_declined_offer_keeps_original_tuple(self) -> None:
+        for is_tty, yes, answer in (
+            (False, False, "y"),
+            (True, True, "y"),
+            (True, False, "n"),
+        ):
+            with self.subTest(is_tty=is_tty, yes=yes, answer=answer):
+                args = self.args()
+                args.yes = yes
+                with (
+                    mock.patch.object(
+                        ota.sys, "stdin", mock.Mock(isatty=lambda: is_tty),
+                    ),
+                    mock.patch("builtins.input", return_value=answer) as prompt,
+                ):
+                    self.assertFalse(ota.prompt_for_125khz_temp_radio(args))
+                self.assertEqual(args.temp_values[1], 250.0)
+                if not is_tty or yes:
+                    prompt.assert_not_called()
+
+
 class ReliabilityTests(unittest.TestCase):
     def test_controller_rejects_meshcli_zero_exit_no_response(self) -> None:
         """Do not trust meshcore-cli's exit status without protocol output."""

tools/lora_ota/test_lora_ota_legacy_dedup.py

@@ -0,0 +1,163 @@
+"""Offline coverage for the bounded, explicit legacy relay workaround."""
+import argparse
+import contextlib
+import io
+import subprocess
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+from test_lora_ota import firmware, mota_blob, VERSION_NEW
+
+
+class LegacyManifestRecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.package = ota.parse_mota(mota_blob(firmware(b"test" * 100, VERSION_NEW)))
+        self.relay = "ab" * 32
+        self.args = argparse.Namespace(
+            legacy_relay_dedup_workaround=True, source_shares_controller=True,
+            relay_values=[(self.relay, "secret")], target="cd" * 32,
+            temp_values=(909.5, 125, 7, 5, 120),
+        )
+        self.status = ("OTA | download: failed (manifest timeout) 0/0 "
+                       f"id={self.package.manifest_id}")
+        self.controller = mock.Mock()
+        self.contact = {"public_key": self.relay, "out_path_len": 0}
+        self.lease = "TempRadio active: 909.500,125.00,7,5 6000s left,preamble=32"
+        self.controller._run.side_effect = lambda cmd, *a, **kw: (
+            [self.contact] if cmd[0] == "contact_info" else [{"text": self.lease}]
+        )
+        self.controller.remote_command.side_effect = [
+            self.status, f"OK pulling mid={self.package.manifest_id} -> flash (primary traffic)"
+        ]
+        self.addCleanup(mock.patch.stopall)
+        mock.patch.object(ota.time, "sleep").start()
+        mock.patch.object(ota.time, "monotonic", return_value=100).start()
+        self.output = contextlib.redirect_stdout(io.StringIO())
+        self.output.__enter__()
+        self.addCleanup(self.output.__exit__, None, None, None)
+
+    def recover(self, status=None, attempt=0, deadline=1000):
+        return ota.recover_legacy_manifest_timeout(
+            self.controller, self.args, self.package, None,
+            self.status if status is None else status, deadline, attempt,
+        )
+
+    def test_unique_small_direct_frames_and_single_exact_pull(self):
+        self.assertTrue(self.recover())
+        frames = [c.args[0] for c in self.controller.send_legacy_ota_cache_probe.call_args_list]
+        self.assertEqual(len(frames), 192)
+        self.assertEqual(len(set(frames)), 192)
+        self.assertEqual({len(f) for f in frames}, {13})
+        self.assertTrue(all(f[:3] == bytes.fromhex("320000") for f in frames))
+        self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list],
+                         ["ota status", f"ota pull {self.package.manifest_id} flash"])
+        self.assertFalse(self.controller.remote_command.call_args_list[-1].kwargs["retry"])
+        self.assertEqual(self.controller._run.call_count, 13)  # route + 12 lease proofs
+
+    def test_off_by_default_and_two_round_limit(self):
+        self.args.legacy_relay_dedup_workaround = False
+        self.assertFalse(self.recover())
+        self.args.legacy_relay_dedup_workaround = True
+        self.assertFalse(self.recover(attempt=2))
+        self.controller._run.assert_not_called()
+
+    def test_wrong_mid_partial_ready_and_other_errors_never_trigger(self):
+        for status in (
+            self.status.replace(self.package.manifest_id, "FFFFFFFF"),
+            self.status.replace("0/0", "1/40"),
+            self.status.replace("manifest timeout", "signature invalid"),
+            f"download: fetching 1/40 id={self.package.manifest_id}",
+            f"download: ready to install 40/40 id={self.package.manifest_id}",
+        ):
+            with self.subTest(status=status):
+                self.assertFalse(self.recover(status))
+        self.controller.send_legacy_ota_cache_probe.assert_not_called()
+
+    def test_not_direct_and_unproven_temp_lease_fail_before_transmit(self):
+        for path in (-1, 1, None):
+            self.contact["out_path_len"] = path
+            with self.assertRaisesRegex(ota.OtaError, "direct"):
+                self.recover()
+        self.contact["out_path_len"] = 0
+        for lease in ("TempRadio inactive", self.lease.replace("6000s", "30s"),
+                      self.lease.replace("125.00", "250.00"),
+                      self.lease.replace(" 6000s left", "")):
+            self.lease = lease
+            with self.assertRaises(ota.OtaError):
+                self.recover()
+        self.controller.send_legacy_ota_cache_probe.assert_not_called()
+
+    def test_global_deadline_is_not_extended(self):
+        with self.assertRaisesRegex(ota.OtaError, "insufficient"):
+            self.recover(deadline=101)
+        self.controller.send_legacy_ota_cache_probe.assert_not_called()
+
+    def test_session_change_never_cancels_or_pulls(self):
+        self.controller.remote_command.side_effect = ["no download"]
+        with self.assertRaisesRegex(ota.OtaError, "session changed"):
+            self.recover()
+        self.assertEqual(self.controller.remote_command.call_count, 1)
+
+    def test_completion_during_turnover_is_preserved(self):
+        self.controller.remote_command.side_effect = [
+            f"download: ready to install 40/40 id={self.package.manifest_id}"
+        ]
+        self.assertTrue(self.recover())
+        self.assertEqual(self.controller.remote_command.call_count, 1)
+
+    def test_lost_pull_ack_reconciles_without_resending(self):
+        self.controller.remote_command.side_effect = [self.status, ota.TransmissionError("lost"),
+            f"download: fetching 0/40 id={self.package.manifest_id}"]
+        self.assertTrue(self.recover())
+        self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list],
+                         ["ota status", f"ota pull {self.package.manifest_id} flash", "ota status"])
+
+    def test_probe_error_aborts_no_pull(self):
+        self.controller.send_legacy_ota_cache_probe.side_effect = ota.OtaError("queue full")
+        with self.assertRaisesRegex(ota.OtaError, "queue full"):
+            self.recover()
+        self.controller.remote_command.assert_not_called()
+
+    def test_raw_send_error_or_missing_marker_is_not_success(self):
+        for failure in ("Error sending raw packet", "Unknown command", ""):
+            controller = object.__new__(ota.Controller)
+            def execute(commands, *a, **kw):
+                output = failure + (commands[-1] if failure else "")
+                return subprocess.CompletedProcess([], 0, stdout=output, stderr="")
+            with mock.patch.object(controller, "_execute", side_effect=execute):
+                with self.assertRaises(ota.OtaError):
+                    controller.send_legacy_ota_cache_probe(b"\x32\0\0", timeout=5)
+
+    def test_parser_requires_explicit_compatible_topology(self):
+        parser = ota.build_parser()
+        common = ["update.mota", "target", "--legacy-relay-dedup-workaround"]
+        for flags in ([], ["--relay", "relay"], ["--source-shares-controller"],
+                      ["--relay", "a", "--relay", "b", "--source-shares-controller"]):
+            with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+                ota.validate_args(parser.parse_args(common + flags), parser)
+
+    def test_monitor_uses_same_deadline_and_bounded_attempt_counter(self):
+        self.args.transfer_timeout_minutes = 20
+        self.args.poll_seconds = 1
+        self.args.reply_timeout = 1
+        ready = f"download: ready to install 1/1 id={self.package.manifest_id}"
+        for passive in (False, True):
+            with self.subTest(passive=passive):
+                self.controller.remote_command.side_effect = [self.status, self.status, ready]
+                seeder = mock.Mock() if passive else None
+                if seeder:
+                    seeder.payload_read_progress.return_value = (0, 1, 0)
+                with mock.patch.object(ota, "initial_status_wait_seconds", return_value=0), \
+                     mock.patch.object(ota, "passive_progress_stall_seconds", return_value=0), \
+                     mock.patch.object(ota, "transfer_tail_guard_seconds", return_value=0), \
+                     mock.patch.object(ota, "recover_legacy_manifest_timeout",
+                                       side_effect=[True, True, False]) as recover:
+                    self.assertEqual(ota.monitor_download(self.controller, self.args,
+                                                         self.package, seeder), ready)
+                self.assertEqual([c.args[-1] for c in recover.call_args_list], [0, 1, 2])
+                self.assertEqual({c.args[-2] for c in recover.call_args_list}, {1300})
+
+
+if __name__ == "__main__":
+    unittest.main()