Add menus for GitHub firmware and OTA preparation

docs/ota_wizard.md

@@ -48,123 +48,118 @@ Close other apps using the USB port before starting.
 
 ## Start
 
-From the MeshCore checkout:
+From the MeshCore checkout, run:
 
 ```bash
 python3 tools/lora_ota/ota_wizard.py
 ```
 
-On Windows, use `py` in place of `python3`. The first run installs USB and
-verification dependencies in a private Python environment. On Raspberry Pi OS,
-install `python3-venv` if Python reports that it is missing.
+On Windows, use `py` in place of `python3`. No command-line options are needed.
+The menu provides these actions:
 
-The wizard selects the USB radio and destination automatically when there is
-only one of each. Otherwise, choose from the displayed list. Enter the
-destination admin password once, choose firmware, then review the plan and
-press Enter to start. Password entry is hidden and passwords are never saved.
+1. **Update a nearby radio** — the default.
+2. **Check radios / save a preparation request** — inspect without installing.
+3. **Prepare an update bundle on this computer** — no radios needed.
+4. **Recover an interrupted update** — restore saved settings.
+5. **Settings** — change transfer, bootloader, connection, or build preferences.
+6. **Exit**.
 
-Firmware choices are:
+The first live connection installs USB and verification dependencies in a
+private Python environment if needed. On Raspberry Pi OS, install
+`python3-venv` if Python reports that it is missing.
+Preparing a signed input file can also install signature-verification support
+on first use, without installing the USB software.
 
-- **Published GitHub firmware:** the default, matched to the exact board and
-  role. A running version that is already current is left in place.
-- **Custom firmware file:** an application DFU ZIP, BIN, HEX, full mOTA, or a
-  prepared update bundle.
-- **Build custom Full firmware from source:** offered on a computer. This
-  needs the firmware build tools and Bash; it is not offered by default on
-  a Pi. The generated version is newer than the detected installed version.
+For an update, the wizard selects the USB radio and destination automatically
+when there is only one of each. Otherwise, choose from the displayed lists.
+Enter the destination admin password once, choose firmware, then review the
+plan and press Enter to start. Password entry is hidden and passwords are
+never saved. After an action, the menu is shown again.
 
-The displayed plan includes package sizes, firmware version, and any
-bootloader or trusted update signer being installed. After a reboot, the
-wizard checks the running firmware hash and version, or the bootloader's
-identity and version, before reporting success.
+## Choose firmware from GitHub releases
+
+The firmware menu offers:
+
+- **Latest compatible GitHub release (recommended):** downloads the published
+  stable file for the detected board and role.
+- **Choose another GitHub release:** lists compatible releases, including
+  clearly labelled previews, so you can choose a version.
+- **Local firmware file or prepared bundle:** accepts an application DFU ZIP,
+  BIN, HEX, full mOTA, or a bundle prepared on another computer.
+- **Build custom Full firmware from source:** offered on a computer with the
+  firmware build tools and Bash. It is not offered by default on a Pi.
+
+You do not need to copy release URLs or manually download published files.
+The wizard checks the board and role, searches release files for an exact
+running base when preparing a differential update, and checks for a compatible
+signed bootloader on supported nRF52 destinations. Firmware that is already
+installed or older is left in place.
+
+The final plan shows the package sizes, firmware version, and any bootloader
+or trusted update signer being installed. After reboot, the wizard checks the
+running firmware hash and version, or the bootloader identity and version,
+before reporting success.
 
 ## Differential or full updates
 
-The wizard looks for the **exact running firmware** in its cache and published
-GitHub assets. It uses a differential package when it has a compatible builder,
-can match that exact base, and the resulting package is smaller. A file with
-the same version label but a different body hash is not a usable base.
+**Automatic** is the default transfer setting. It uses a differential package
+when a compatible builder and the **exact running firmware** are available
+and the package is smaller. A file with the same version label but a different
+body hash is not a usable base. Published bases are found through GitHub
+releases; a local base can be selected when the running image was custom.
 
 On a computer, the differential builder can be prepared automatically with a
-compatible Rust toolchain. On a Pi without a builder, the default is a full
-update made by Python without compiling. This requires destination storage
-that supports full updates, such as the Ikoka Stick's QSPI flash.
+compatible Rust toolchain. On a Pi without a builder, a full update is made by
+Python without compiling. The destination must support full updates, such as
+an Ikoka Stick with working QSPI storage.
+
+To force a full application transfer, open **Settings**, choose **Full** for
+the transfer method, then start the update. Settings also lets you turn the
+bootloader check off for an application-only update, choose a USB port or
+source type, or set a builder path, custom source folder/version, or cache folder.
 
 An internal-flash nRF52 destination that requires differential installation
-cannot use that full fallback. Prepare its bundle on a computer instead:
+cannot use the full fallback. Use the computer-to-Pi workflow below.
 
-1. On the Pi, inspect the connected radios:
+## Prepare on a computer, send from a Pi
 
-   ```bash
-   python3 tools/lora_ota/ota_wizard.py --check
-   ```
+All steps use the same menu; no command-line options are needed.
 
-   Copy the JSON file shown after **Preparation request:** to the computer.
-   It contains public board information and the exact running firmware hash,
-   without the admin password.
+1. On the Pi, choose **Check radios / save a preparation request**. Connect to
+   the source and select the destination as prompted. Copy the JSON file
+   shown after **Preparation request:** to the computer. It contains public
+   board information and the exact running hash, without the password.
+2. On the computer, choose **Prepare an update bundle on this computer**.
+   Select the saved request or enter its file path. Choose firmware from
+   GitHub releases, or select a local custom image. The wizard searches for
+   the exact base and offers a local base file if needed. Accept the proposed
+   output filename or choose another one.
+3. Copy the resulting `.ota.zip` file to the Pi. Choose **Update a nearby
+   radio**, then **Local firmware file or prepared bundle**, and select that
+   ZIP.
 
-2. On the computer, prepare the update using the request, new firmware, and
-   the exact firmware currently running on the destination:
-
-   ```bash
-   python3 tools/lora_ota/ota_wizard.py --prepare-only \
-     --request DEVICE.json --firmware NEW_FIRMWARE.zip \
-     --base RUNNING_FIRMWARE.zip --output update.ota.zip
-   ```
-
-   No radio is opened. If the compatible differential builder and all input
-   files are already present, this also works without network access. Add
-   `--motatool /path/to/motatool` to use an existing builder.
-
-3. Copy `update.ota.zip` back to the Pi and run:
-
-   ```bash
-   python3 tools/lora_ota/ota_wizard.py --firmware update.ota.zip
-   ```
-
-   The live destination is checked again before installation. A changed base
-   hash or a different board stops the differential update.
-
-For an offline full bundle on a destination that supports it, add `--full`
-to the preparation command. This path needs neither the differential builder
+The preparation step does not open a radio. When the builder and input files
+are already present, local preparation also works without network access.
+For a full bundle on a destination that supports it, choose **Full** in
+**Settings** before preparation; this needs neither a differential compiler
 nor USB dependencies.
 
-## Useful options
-
-| Option | Purpose |
-| --- | --- |
-| `--check` | Authenticate and inspect OTA support without downloading or installing firmware |
-| `--port /dev/ttyUSB0` | Choose a USB port; Windows example: `--port COM5` |
-| `--mode repeater` or `--mode companion` | Select the source type if automatic detection is unsuitable |
-| `--target PUBLIC_KEY` | Choose a destination by its full 64-character key; it must still answer direct discovery |
-| `--firmware FILE` | Use a local firmware image or prepared bundle |
-| `--base FILE` | Supply the exact running firmware for differential preparation |
-| `--full` | Choose a full application transfer when the destination supports it |
-| `--build` | Build custom Full firmware from this checkout |
-| `--version 1.18.0.3-custom` | Set the custom build version; it must be newer than the running version |
-| `--no-bootloader` | Update application firmware only |
-| `--cache DIRECTORY` | Choose where packages, public requests, and recovery records are saved |
-| `--yes` | Accept the displayed plan; required choices and credentials must still be supplied |
-
-For unattended use, provide `MESHCORE_ADMIN_PASSWORD` in the process environment
-and pass the port, source type, and target key. The wizard does not write that
-password into its preferences, recovery records, or custom build environment.
+The live destination is checked again before installation. A changed base
+hash or a different board stops a differential update.
 
 ## If the connection is interrupted
 
-Keep the same source radio connected and run the wizard again. It checks the
-saved source and destination public keys before restoring settings from an
-unfinished update. To restore settings and exit:
+Keep the same source radio connected and run the wizard again. Choose
+**Recover an interrupted update**, then select the saved session by its
+public-key prefix and progress state. The wizard checks the complete source
+and destination keys before restoring settings. A normal update also restores an unfinished
+session for the selected destination before proceeding.
 
-```bash
-python3 tools/lora_ota/ota_wizard.py --recover --target PUBLIC_KEY
-```
-
-Keep the recovery record until the wizard reports that restoration is
-complete. A lost install reply is followed by a running-image check; the
-wizard does not blindly repeat an install command. Temporary radio windows
-use the existing main-channel settings, so an interrupted process does not
-leave the destination on a separate update frequency.
+Keep the recovery record until restoration is complete. A lost install reply
+is followed by a running-image check; the wizard does not blindly repeat an
+install command. Temporary radio windows use the existing main-channel
+settings. Existing temporary schedules block an update before settings are
+changed, including on older firmware.
 
 The cache is normally `~/.cache/meshcore/ota` on Linux,
 `~/Library/Caches/MeshCore/ota` on macOS, or `%LOCALAPPDATA%\MeshCore\ota` on
@@ -172,7 +167,33 @@ Windows. It contains firmware, public preparation records, selected board
 keys, and restoration settings. Trusted signer keys added for an update remain
 installed on the destination.
 
-For the individual commands and a manual board-by-board example, see
-[Station G2 to Ikoka Stick](ota_station_g2_to_ikoka_stick.md). For routed
-updates and explicit multi-node setup, use the
-[scripted LoRa OTA guide](lora_ota_automation.md).
+<details markdown="1">
+<summary>Optional command-line use for automation</summary>
+
+Interactive users can use the menus for every step. Existing command-line
+options remain available for scripts:
+
+| Option | Purpose |
+| --- | --- |
+| `--check` | Inspect without downloading or installing firmware |
+| `--port /dev/ttyUSB0` | Choose a port; Windows example: `--port COM5` |
+| `--mode repeater` or `--mode companion` | Select the source type |
+| `--target PUBLIC_KEY` | Choose the full destination key; direct discovery is still required |
+| `--firmware FILE` | Use a local firmware image or prepared bundle |
+| `--base FILE` | Supply the exact running image for differential preparation |
+| `--full` | Force a full transfer where supported |
+| `--build` | Build custom Full firmware from source |
+| `--no-bootloader` | Update application firmware only |
+| `--prepare-only --request DEVICE.json --firmware FILE --output update.ota.zip` | Prepare without radios |
+| `--recover --target PUBLIC_KEY` | Restore settings and exit |
+| `--yes` | Accept the displayed plan; credentials and required choices must still be supplied |
+
+For unattended use, provide `MESHCORE_ADMIN_PASSWORD` in the process environment
+and pass the port, source type, and target key. The password is excluded from
+preferences, recovery records, and the custom build environment.
+
+</details>
+
+For individual commands and a manual board example, see
+[Station G2 to Ikoka Stick](ota_station_g2_to_ikoka_stick.md). For routed updates
+and explicit multi-node setup, use the [scripted LoRa OTA guide](lora_ota_automation.md).

tools/lora_ota/ota_wizard.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import asyncio
+from copy import copy
 import getpass
 import importlib.util
 import importlib.metadata
@@ -73,6 +74,8 @@ def arguments(argv=None):
     parser.add_argument("--recover", action="store_true", help="restore settings from an interrupted update, then exit")
     parser.add_argument("--yes", action="store_true", help="accept the displayed update plan; does not guess missing credentials")
     parser.add_argument("--runtime-ready", action="store_true", help=argparse.SUPPRESS)
+    parser.add_argument("--menu-action", choices=("update", "check", "prepare", "recover"),
+                        help=argparse.SUPPRESS)
     parser.add_argument("--check", action="store_true", help="inspect source and destination without preparing or installing an update")
     parser.add_argument("--prepare-only", action="store_true", help="prepare a portable update bundle on a computer without opening radios")
     parser.add_argument("--request", type=Path, help="destination preparation request saved by a previous live inspection")
@@ -80,6 +83,131 @@ def arguments(argv=None):
     return parser.parse_args(argv)
 
 
+class MenuCancelled(Exception):
+    """Return to the main menu without changing a radio."""
+
+
+def file_prompt(prompt: str, default: str | None = None) -> Path:
+    if default and Path(default).expanduser().is_file():
+        selected = select(prompt, ["Use " + default, "Choose another file", "Back"])
+        if selected == 0:
+            return Path(default).expanduser()
+        if selected == 2:
+            raise MenuCancelled
+    while True:
+        value = ask(prompt + " (blank to go back)")
+        if not value:
+            raise MenuCancelled
+        path = Path(value.strip().strip('"').strip("'")).expanduser()
+        if path.is_file():
+            return path
+        print("That file was not found. Enter its path, or leave it blank to go back.")
+
+
+def menu_settings_path() -> Path:
+    return cache_directory() / "menu-settings.json"
+
+
+def load_menu_settings(args) -> None:
+    try:
+        values = json.loads(ota.read_bounded_file(menu_settings_path(), 8192, "menu settings"))
+    except (OSError, ValueError, ota.OtaError):
+        return
+    if not isinstance(values, dict):
+        return
+    for field in ("full", "no_bootloader"):
+        if type(values.get(field)) is bool:
+            setattr(args, field, values[field])
+    if values.get("mode") in ("auto", "repeater", "companion"):
+        args.mode = values["mode"]
+    for field in ("motatool", "version", "port"):
+        value = values.get(field)
+        if value is None or isinstance(value, str) and len(value) <= 4096:
+            setattr(args, field, value)
+    for field in ("source_tree", "cache"):
+        value = values.get(field)
+        if isinstance(value, str) and value and len(value) <= 4096:
+            setattr(args, field, Path(value).expanduser())
+
+
+def settings_menu(args) -> None:
+    while True:
+        print("\nSettings")
+        options = [
+            "Transfer: " + ("Full" if args.full else "Automatic (prefer differential)"),
+            "Check bootloader: " + ("Off" if args.no_bootloader else "On"),
+            "USB source type: " + args.mode.capitalize(),
+            "USB port: " + (args.port or "Automatic"),
+            "Differential builder: " + (args.motatool or "Automatic"),
+            "Custom source folder: " + str(args.source_tree),
+            "Custom version: " + (args.version or "Automatic"),
+            "Download/cache folder: " + str(args.cache),
+            "Back",
+        ]
+        choice = select("Setting", options, len(options)-1)
+        if choice == 8:
+            break
+        if choice == 0:
+            args.full = bool(select("Transfer", ["Automatic (prefer differential)", "Full"], int(args.full)))
+        elif choice == 1:
+            args.no_bootloader = bool(select("Check bootloader", ["On", "Off"], int(args.no_bootloader)))
+        elif choice == 2:
+            modes = ("auto", "repeater", "companion")
+            args.mode = modes[select("USB source type", ["Automatic", "Repeater", "Companion"], modes.index(args.mode))]
+        elif choice == 3:
+            args.port = ask("USB port (blank for automatic)") or None
+        elif choice == 4:
+            args.motatool = ask("Differential builder path (blank for automatic)") or None
+        elif choice == 5:
+            args.source_tree = Path(ask("MeshCore source folder", str(args.source_tree))).expanduser()
+        elif choice == 6:
+            args.version = ask("Custom version (blank for automatic)") or None
+        elif choice == 7:
+            args.cache = Path(ask("Download/cache folder", str(args.cache))).expanduser()
+    save_preferences(menu_settings_path(), **{
+        field: str(getattr(args, field)) if field in ("cache", "source_tree") else getattr(args, field)
+        for field in ("full", "no_bootloader", "mode", "port", "motatool", "source_tree", "version", "cache")})
+
+
+def choose_request(args):
+    records = []
+    for path in sorted((args.cache / "requests").glob("*.json"), key=lambda path: path.stat().st_mtime, reverse=True)[:50]:
+        try:
+            records.append((path, load_request(path)))
+        except (ota.OtaError, OSError):
+            continue
+    if not records:
+        return load_request(file_prompt("Preparation request JSON"))
+    choices = [f"{record.hw_id} — {record.public_key[:12].upper()} ({record.version_text})"
+               for _, record in records] + ["Open another preparation request file", "Back"]
+    chosen = select("Preparation request", choices)
+    if chosen == len(records) + 1:
+        raise MenuCancelled
+    return (records[chosen][1] if chosen < len(records)
+            else load_request(file_prompt("Preparation request JSON")))
+
+
+def choose_recovery(args) -> str:
+    records = []
+    for path in sorted((args.cache / "sessions").glob("*.json"), key=lambda path: path.stat().st_mtime, reverse=True)[:50]:
+        try:
+            record = json.loads(ota.read_bounded_file(path, 64 * 1024, "recovery record"))
+            key = record.get("target_key") if isinstance(record, dict) else None
+            if (record.get("format") == 1 and isinstance(key, str)
+                    and re.fullmatch(r"[0-9A-Fa-f]{64}", key)
+                    and record.get("phase") not in ("finished", "prepared", "inspecting")):
+                records.append((key, record.get("phase", "unfinished")))
+        except (ota.OtaError, OSError, ValueError, AttributeError):
+            continue
+    if not records:
+        print("No interrupted updates were found in this cache.")
+        raise MenuCancelled
+    chosen = select("Interrupted update", [f"{key[:12].upper()} — {phase}" for key, phase in records] + ["Back"])
+    if chosen == len(records):
+        raise MenuCancelled
+    return records[chosen][0]
+
+
 def ensure_dependencies(args, argv) -> None:
     required = ("serial", "meshcore", "cryptography")
     missing = [name for name in required if importlib.util.find_spec(name) is None]
@@ -231,6 +359,218 @@ def custom_build(args, destination, target_env: str) -> Path:
     raise ota.OtaError("The custom build did not produce a matching application image.")
 
 
+async def latest_asset(catalog, destination):
+    print("Checking GitHub releases for this exact board and role…")
+    try:
+        return await asyncio.to_thread(catalog.choose_asset, target_id=destination.target_id)
+    except ota.OtaError as exc:
+        print(str(exc))
+        return None
+
+
+async def choose_release(catalog, destination):
+    pages = 3
+    while True:
+        choices = await asyncio.to_thread(catalog.release_choices, target_id=destination.target_id,
+                                          include_prerelease=True, max_pages=pages)
+        if not choices:
+            print("No compatible firmware was found on these release pages.")
+        labels = [choice.label for choice in choices]
+        more = pages < 10 and catalog.has_more_releases
+        if more:
+            labels.append("Load older GitHub releases")
+        labels.append("Back")
+        selected = select("GitHub release", labels)
+        if selected < len(choices):
+            return choices[selected].asset
+        if more and selected == len(choices):
+            pages = min(10, pages + 3)
+            continue
+        raise MenuCancelled
+
+
+async def choose_firmware(args, destination, catalog, asset, remembered):
+    if remembered.get("target") != destination.public_key.lower():
+        remembered = {}
+    if args.firmware:
+        return args.firmware, "local"
+    preferred = remembered.get("firmware_source", "latest")
+    while True:
+        source = "build" if args.build else "latest"
+        if not args.build and not args.yes:
+            sources = ["latest", "release", "local"]
+            labels = ["Latest compatible GitHub release (recommended)",
+                      "Choose another GitHub release", "Local firmware file or prepared bundle"]
+            if not small_host():
+                sources.append("build")
+                labels.append("Build custom Full firmware from source")
+            default = sources.index(preferred) if preferred in sources else 0
+            selected = select("Firmware", labels + ["Back"], default)
+            if selected == len(sources):
+                raise MenuCancelled
+            source = sources[selected]
+        if source == "local":
+            try:
+                return file_prompt("Firmware ZIP, BIN, HEX, full mOTA, or prepared bundle",
+                                   remembered.get("firmware_file")), source
+            except MenuCancelled:
+                continue
+        if source == "release":
+            try:
+                selected_asset = await choose_release(catalog, destination)
+            except MenuCancelled:
+                continue
+        else:
+            if asset is None:
+                asset = await latest_asset(catalog, destination)
+            if asset is None:
+                if args.yes or args.build:
+                    raise ota.OtaError("No GitHub application matched this board. Choose another release or a local firmware file.")
+                preferred = "local"
+                continue
+            selected_asset = asset
+        if source in ("latest", "release"):
+            print(f"GitHub release: {selected_asset.release_tag}\n  {selected_asset.name}")
+            return await asyncio.to_thread(download_asset, selected_asset, args.cache / "downloads"), source
+        if not args.version:
+            args.version = ota.format_version(max(destination.version, selected_asset.version) + 1) + "-custom"
+        return await asyncio.to_thread(custom_build, args, destination, selected_asset.target_env), source
+
+
+async def choose_missing_base(args, destination, catalog):
+    """Offer the useful fallbacks only after the automatic exact-base search."""
+    pages = 3
+    while True:
+        choices, actions = [], []
+        if pages < 10:
+            choices.append("Search older GitHub releases for the running firmware")
+            actions.append("github")
+        choices.append("Choose a local copy of the exact running firmware")
+        actions.append("local")
+        if destination.platform == "esp32" or destination.external_storage:
+            choices.append("Use a full update instead")
+            actions.append("full")
+        choices.append("Back")
+        actions.append("back")
+        action = actions[select("The exact running image was not found", choices)]
+        if action == "back":
+            raise MenuCancelled
+        if action == "full":
+            return None
+        if action == "local":
+            supplied = file_prompt("Exact running firmware ZIP, BIN, HEX, or full mOTA")
+            try:
+                return await asyncio.to_thread(find_running_image, destination, args.cache, catalog, supplied=supplied)
+            except ota.OtaError as exc:
+                print(str(exc))
+                continue
+        pages = min(10, pages + 3)
+        try:
+            await asyncio.to_thread(catalog.release_choices, target_id=destination.target_id,
+                                    include_prerelease=True, max_pages=pages)
+            base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog,
+                                           candidate_limit=16 if pages < 10 else 32)
+            if base:
+                return base
+            if not catalog.has_more_releases:
+                pages = 10
+            print("No downloaded GitHub candidate matched the exact running image hash.")
+        except ota.OtaError as exc:
+            print(str(exc))
+
+
+async def prepare_selected_firmware(args, destination, firmware, catalog):
+    prepared = await asyncio.to_thread(load_bundle, firmware, destination)
+    if prepared:
+        image, application = prepared
+        if args.full and not application.info.is_full:
+            application = await asyncio.to_thread(prepare_application, image, destination, args.cache, full=True)
+    else:
+        image = await asyncio.to_thread(load_application, firmware,
+                                        target_id=destination.target_id, hw_id=destination.hw_id)
+        application = await asyncio.to_thread(load_ready_application, firmware, image, destination)
+    ImageCache(args.cache).remember(image)
+    if image.fw_version <= destination.version:
+        print("The selected firmware is already installed or older; keeping the running firmware.")
+        return image, None
+    if application:
+        return image, application
+    builder = None if args.full else find_builder(args.cache, args.motatool)
+    if not args.full and not builder and not small_host():
+        builder = await asyncio.to_thread(install_builder, args.cache)
+    base = None
+    if builder:
+        base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog, supplied=args.base)
+        if base is None and getattr(args, "interactive_menu", False):
+            base = await choose_missing_base(args, destination, catalog)
+    if not builder and not args.full:
+        print("Using a full update: no differential compiler setup is needed on this host.")
+    if destination.platform == "nrf52" and not destination.external_storage and not (base and builder):
+        raise ota.OtaError(
+            "This radio needs an exact-base differential update. On a computer, choose "
+            "Prepare an update bundle and select its saved preparation request. "
+            "A compatible differential builder must be available; its path can be set under Settings.")
+    application = await asyncio.to_thread(prepare_application, image, destination, args.cache,
+                                          base=base, builder=builder, full=args.full)
+    return image, application
+
+
+async def prepare_interactively(args) -> int:
+    destination = choose_request(args)
+    print(f"Preparing for {destination.hw_id} — {destination.public_key[:12].upper()} ({destination.version_text})")
+    catalog = GitHubCatalog()
+    args.interactive_menu = True
+    firmware, _ = await choose_firmware(args, destination, catalog, None, {})
+    verification_cache = ota.cryptography_repair_root()
+    if (verification_cache / "cryptography").is_dir():
+        try:
+            ota.activate_cached_cryptography(verification_cache)
+        except ota.OtaError:
+            pass  # A working system verifier may already be loaded.
+    try:
+        image, package = await prepare_selected_firmware(args, destination, firmware, catalog)
+    except ImportError:
+        # Unsigned full packaging stays dependency-free. Signed inputs request
+        # their verifier only when it is actually needed; no USB SDK is loaded.
+        await asyncio.to_thread(ensure_signature_support)
+        image, package = await prepare_selected_firmware(args, destination, firmware, catalog)
+    if package is None:
+        print("Choose a newer firmware release to prepare an update.")
+        return 0
+    folder = args.cache / "bundles"
+    default = folder / f"{destination.hw_id}-{ota.format_version(image.fw_version)}.ota.zip"
+    index = 1
+    while default.exists():
+        default = folder / f"{destination.hw_id}-{ota.format_version(image.fw_version)}-{index}.ota.zip"
+        index += 1
+    output = Path(ask("Save update bundle", str(default)).strip().strip('"').strip("'")).expanduser()
+    await asyncio.to_thread(save_bundle, output, image, package)
+    print(f"Prepared {package.info.kind}: {output} ({len(package.blob):,} transfer bytes)")
+    print("Copy this ZIP to the Pi. Choose Update a nearby radio, then Local firmware file or prepared bundle.")
+    return 0
+
+
+def ensure_signature_support() -> None:
+    root = ota.cryptography_repair_root()
+    if root.is_dir():
+        try:
+            ota.activate_cached_cryptography(root)
+            return
+        except ota.OtaError:
+            pass
+    print("Preparing signature verification support. This is a one-time download.")
+    root.mkdir(parents=True, exist_ok=True)
+    log = root / "setup.log"
+    environment = {key: value for key, value in os.environ.items()
+                   if key not in ("MESHCORE_ADMIN_PASSWORD", "MESHCORE_OTA_PASSWORD")}
+    with log.open("w", encoding="utf-8") as output:
+        completed = subprocess.run(ota.cryptography_repair_command(root), env=environment,
+                                   stdout=output, stderr=subprocess.STDOUT, timeout=600)
+    if completed.returncode:
+        raise ota.OtaError(f"Signature verification support could not be installed. Details: {log}")
+    ota.activate_cached_cryptography(root)
+
+
 async def run(args) -> int:
     args.cache.mkdir(parents=True, exist_ok=True, mode=0o700)
     if args.target and not re.fullmatch(r"[0-9A-Fa-f]{64}", args.target):
@@ -314,65 +654,8 @@ async def run(args) -> int:
             return 0
         await session.snapshot()
         catalog = GitHubCatalog()
-        print("Checking published firmware for this exact board and role…")
-        try:
-            asset = await asyncio.to_thread(catalog.choose_asset, target_id=destination.target_id)
-        except ota.OtaError as exc:
-            asset = None
-            if not args.firmware and not args.build:
-                print(str(exc))
-        choice = 2 if args.build else 1 if args.firmware else 0
-        if not args.yes and not args.firmware and not args.build:
-            options = ["Published GitHub firmware (recommended)", "Custom firmware file"]
-            if not small_host():
-                options.append("Build custom Full firmware from source")
-            default_choice = remembered.get("firmware_choice", 0 if asset else 1)
-            if remembered.get("target") != key.lower() or default_choice not in range(len(options)):
-                default_choice = 0 if asset else 1
-            choice = select("Firmware", options, default_choice)
-        if choice == 0:
-            if not asset:
-                raise ota.OtaError("No published firmware matched this board; supply --firmware or choose a custom build.")
-            firmware = await asyncio.to_thread(download_asset, asset, args.cache / "downloads")
-        elif choice == 1:
-            default_file = remembered.get("firmware_file") if remembered.get("target") == key.lower() else None
-            firmware = args.firmware or Path(ask("Custom application ZIP, BIN, HEX, full mOTA, or prepared bundle", default_file))
-        else:
-            if not asset:
-                raise ota.OtaError("A custom build needs the exact target environment; use a local firmware file for this board.")
-            if not args.version:
-                args.version = ota.format_version(max(destination.version, asset.version) + 1) + "-custom"
-            firmware = await asyncio.to_thread(custom_build, args, destination, asset.target_env)
-        prepared = await asyncio.to_thread(load_bundle, firmware, destination)
-        if prepared:
-            image, application = prepared
-            if args.full and not application.info.is_full:
-                application = await asyncio.to_thread(prepare_application, image, destination, args.cache, full=True)
-        else:
-            image = await asyncio.to_thread(load_application, firmware,
-                                            target_id=destination.target_id, hw_id=destination.hw_id)
-            application = await asyncio.to_thread(load_ready_application, firmware, image, destination)
-        ImageCache(args.cache).remember(image)
-        if image.fw_version > destination.version and application is None:
-            builder = find_builder(args.cache, args.motatool)
-            if not args.full and not builder and not small_host():
-                builder = await asyncio.to_thread(install_builder, args.cache)
-            base = None
-            if not args.full and builder:
-                base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog,
-                                               supplied=args.base)
-            if not builder:
-                print("Using a full update: no differential compiler setup is needed on this host.")
-            if not builder and destination.platform == "nrf52" and not destination.external_storage:
-                raise ota.OtaError(
-                    "This internal-flash radio requires a differential package. Prepare it on a computer with "
-                    f"--prepare-only --request {request_file} --firmware NEW_FIRMWARE --base RUNNING_FIRMWARE "
-                    "--output update.ota.zip, then select that ZIP on the Pi.")
-            application = await asyncio.to_thread(prepare_application, image, destination, args.cache,
-                                                  base=base, builder=builder, full=args.full)
-        elif image.fw_version <= destination.version:
-            application = None
-            print("The selected published/application version is already installed or older; keeping the running firmware.")
+        firmware, choice = await choose_firmware(args, destination, catalog, None, remembered)
+        image, application = await prepare_selected_firmware(args, destination, firmware, catalog)
         bootloader = None
         if not args.no_bootloader and destination.platform == "nrf52":
             print("Checking the exact-board bootloader update…")
@@ -398,8 +681,8 @@ async def run(args) -> int:
                 destination = await session.install(package, destination)
                 print(f"Verified {package.info.kind} installation.")
         save_preferences(args.cache / "choices.json", port=port, mode=mode,
-                         target=key.lower(), hw=destination.hw_id, firmware_choice=choice,
-                         firmware_file=str(firmware) if choice == 1 else None)
+                         target=key.lower(), hw=destination.hw_id, firmware_source=choice,
+                         firmware_file=str(firmware) if choice == "local" else None)
         print("Update verified successfully.")
         return 0
     finally:
@@ -437,18 +720,59 @@ def prepare_without_radios(args) -> int:
     return 0
 
 
+def menu_main(args) -> int:
+    load_menu_settings(args)
+    action = args.menu_action
+    while True:
+        if action is None:
+            print("\nMeshCore LoRa updater")
+            actions = ("update", "check", "prepare", "recover", "settings", "exit")
+            action = actions[select("Choose", [
+                "Update a nearby radio", "Check radios / save a preparation request",
+                "Prepare an update bundle on this computer", "Recover an interrupted update",
+                "Settings", "Exit"])]
+        current = copy(args)
+        current.interactive_menu = True
+        current.check, current.recover = action == "check", action == "recover"
+        try:
+            if action == "exit":
+                return 0
+            if action == "settings":
+                settings_menu(args)
+            elif action == "prepare":
+                asyncio.run(prepare_interactively(current))
+            else:
+                # Preserve the chosen menu action if first-run dependency
+                # setup reexecutes inside the private Python environment.
+                ensure_dependencies(current, ["--menu-action", action, "--cache", str(args.cache)])
+                if action == "recover":
+                    current.target = choose_recovery(current)
+                asyncio.run(run(current))
+        except MenuCancelled:
+            pass
+        except (ota.OtaError, RuntimeError, OSError, ValueError, ImportError, subprocess.TimeoutExpired) as exc:
+            print(f"Update stopped: {exc}", file=sys.stderr)
+        action = None
+
+
 def main(argv=None) -> int:
     argv = sys.argv[1:] if argv is None else argv
     args = arguments(argv)
     try:
+        if args.menu_action or not any(value != "--runtime-ready" for value in argv):
+            return menu_main(args)
         if args.prepare_only:
             return prepare_without_radios(args)
         ensure_dependencies(args, argv)
         return asyncio.run(run(args))
+    except MenuCancelled:
+        return 0
+    except EOFError:
+        return 0
     except KeyboardInterrupt:
         print("\nStopped. Check the recovery record if the destination was unreachable during cleanup.", file=sys.stderr)
         return 130
-    except (ota.OtaError, RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc:
+    except (ota.OtaError, RuntimeError, OSError, ValueError, ImportError, subprocess.TimeoutExpired) as exc:
         print(f"Update stopped: {exc}", file=sys.stderr)
         return 1
 

tools/lora_ota/ota_wizard_catalog.py

@@ -50,6 +50,30 @@ class ReleaseAsset:
         return ota.format_version(self.version)
 
 
+@dataclass(frozen=True)
+class ReleaseChoice:
+    """One exact-board application from a published release or split family."""
+    release_tag: str
+    asset: ReleaseAsset
+    prerelease: bool
+    published_at: str
+
+    @property
+    def label(self) -> str:
+        # GitHub tags are source data; do not echo control sequences to a terminal.
+        tag = "".join(character if character.isprintable() else "?"
+                      for character in self.release_tag)
+        return tag + (" (prerelease)" if self.prerelease else "")
+
+
+@dataclass(frozen=True)
+class _CatalogRelease:
+    tag: str
+    prerelease: bool
+    published_at: str
+    assets: tuple[ReleaseAsset, ...]
+
+
 @dataclass(frozen=True)
 class UpdatePlan:
     kind: str
@@ -199,61 +223,72 @@ class GitHubCatalog:
         self.opener = opener
         self._assets: list[ReleaseAsset] | None = None
         self._all_assets: list[ReleaseAsset] = []
+        self._pages: dict[int, tuple[_CatalogRelease, ...]] = {}
+        self._last_page: int | None = None
+
+    @property
+    def has_more_releases(self) -> bool:
+        return self._last_page is None and len(self._pages) < 10
+
+    def _releases(self, max_pages: int) -> list[_CatalogRelease]:
+        if type(max_pages) is not int or not 1 <= max_pages <= 10:
+            raise ota.OtaError("Catalog page limit must be between 1 and 10")
+        releases = []
+        for page in range(1, max_pages + 1):
+            if self._last_page is not None and page > self._last_page:
+                break
+            if page not in self._pages:
+                # A split family contains thousands of assets. Cache only parsed
+                # application metadata, keeping each response bounded on a Pi.
+                url = f"https://api.github.com/repos/{self.repository}/releases?per_page=10&page={page}"
+                with _open(url, api=True, opener=self.opener) as response:
+                    raw = response.read(MAX_CATALOG_BYTES + 1)
+                if len(raw) > MAX_CATALOG_BYTES:
+                    raise ota.OtaError("GitHub release catalog is unexpectedly large")
+                try:
+                    rows = json.loads(raw)
+                except (ValueError, UnicodeError) as exc:
+                    raise ota.OtaError("GitHub returned an invalid release catalog") from exc
+                if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
+                    raise ota.OtaError("GitHub returned an invalid release catalog")
+                parsed = []
+                for release in rows:
+                    tag = release.get("tag_name")
+                    if release.get("draft") or not isinstance(tag, str) or _version(tag) is None:
+                        continue
+                    assets = release.get("assets", [])
+                    if not isinstance(assets, list):
+                        raise ota.OtaError("GitHub returned an invalid release asset list")
+                    candidates = tuple(candidate for row in assets if isinstance(row, dict)
+                                       if (candidate := parse_release_asset(row, tag, self.repository)) is not None)
+                    published = release.get("published_at") or ""
+                    parsed.append(_CatalogRelease(tag, _is_prerelease(release),
+                                                  published if isinstance(published, str) else "", candidates))
+                self._pages[page] = tuple(parsed)
+                if len(rows) < 10:
+                    self._last_page = page
+            releases.extend(self._pages[page])
+        # Extra pages loaded through the release menu also become candidates for
+        # exact-base discovery. Their downloaded body hashes remain authoritative.
+        self._all_assets = [asset for page in sorted(self._pages)
+                            for release in self._pages[page] for asset in release.assets]
+        return releases
 
     def fetch_assets(self, *, include_prerelease: bool = False, max_pages: int = 3) -> list[ReleaseAsset]:
-        if not 1 <= max_pages <= 10:
-            raise ota.OtaError("Catalog page limit must be between 1 and 10")
-        releases: list[dict] = []
-        for page in range(1, max_pages + 1):
-            # Release families contain thousands of per-board assets. Small
-            # pages keep GitHub's embedded asset metadata bounded on a Pi.
-            url = f"https://api.github.com/repos/{self.repository}/releases?per_page=10&page={page}"
-            with _open(url, api=True, opener=self.opener) as response:
-                raw = response.read(MAX_CATALOG_BYTES + 1)
-            if len(raw) > MAX_CATALOG_BYTES:
-                raise ota.OtaError("GitHub release catalog is unexpectedly large")
-            try:
-                rows = json.loads(raw)
-            except (ValueError, UnicodeError) as exc:
-                raise ota.OtaError("GitHub returned an invalid release catalog") from exc
-            if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
-                raise ota.OtaError("GitHub returned an invalid release catalog")
-            releases.extend(row for row in rows if not row.get("draft")
-                            and (include_prerelease or not _is_prerelease(row)))
-            if len(rows) < 10:
-                break
-        mains = [row for row in releases if re.match(r"^v\d", str(row.get("tag_name", "")), re.I)
-                 and _version(row["tag_name"]) is not None]
+        releases = [release for release in self._releases(max_pages)
+                    if include_prerelease or not release.prerelease]
+        mains = [release for release in releases if re.match(r"^v\d", release.tag, re.I)]
         if not mains:
             raise ota.OtaError("No published main firmware release was found")
-        main = max(mains, key=lambda row: (_version(row["tag_name"]), row.get("published_at") or ""))
-        family = main["tag_name"]
-        result: list[ReleaseAsset] = []
-        historical: list[ReleaseAsset] = []
-        for release in releases:
-            tag = str(release.get("tag_name", ""))
-            release_family = _release_family(tag)
-            if release_family is None:
-                continue
-            rows = release.get("assets", [])
-            if not isinstance(rows, list):
-                raise ota.OtaError("GitHub returned an invalid release asset list")
-            for row in rows:
-                if isinstance(row, dict):
-                    candidate = parse_release_asset(row, tag, self.repository)
-                    if candidate is not None:
-                        historical.append(candidate)
-                        if release_family == family:
-                            result.append(candidate)
-        self._all_assets = historical
-        self._assets = result
-        return list(result)
+        main = max(mains, key=lambda release: (_version(release.tag), release.published_at))
+        self._assets = [asset for release in releases if _release_family(release.tag) == main.tag
+                        for asset in release.assets]
+        return list(self._assets)
 
-    def candidates(self, target_env: str | None = None, *, target_id: int | None = None,
-                   profile: str = "full", include_embedded: bool = False) -> list[ReleaseAsset]:
+    @staticmethod
+    def _compatible(assets, target_env, target_id, profile, include_embedded=False):
         if target_env is None and target_id is None:
             raise ota.OtaError("Select an exact firmware target")
-        assets = self.fetch_assets() if self._assets is None else self._assets
         result = [row for row in assets if (target_env is None or row.target_env == target_env)
                   and (target_id is None or row.target_id == target_id)
                   and (include_embedded or not row.embedded_bootloader)]
@@ -261,7 +296,12 @@ class GitHubCatalog:
         exact = [row for row in result if row.profile == profile]
         result = exact or [row for row in result if row.profile == "standard"]
         preference = {"mota": 0, "zip": 1, "bin": 2, "hex": 3}
-        return sorted(result, key=lambda row: (-row.version, preference[row.kind], row.name))
+        return sorted(result, key=lambda row: (-row.version, preference[row.kind], row.name, row.release_tag))
+
+    def candidates(self, target_env: str | None = None, *, target_id: int | None = None,
+                   profile: str = "full", include_embedded: bool = False) -> list[ReleaseAsset]:
+        assets = self.fetch_assets() if self._assets is None else self._assets
+        return self._compatible(assets, target_env, target_id, profile, include_embedded)
 
     def choose_asset(self, target_env: str | None = None, **kwargs) -> ReleaseAsset:
         choices = self.candidates(target_env, **kwargs)
@@ -269,6 +309,35 @@ class GitHubCatalog:
             raise ota.OtaError("No published application matches this exact board, role, and profile")
         return choices[0]
 
+    def release_choices(self, target_env: str | None = None, *, target_id: int | None = None,
+                        profile: str = "full", include_prerelease: bool = True,
+                        max_pages: int = 3) -> list[ReleaseChoice]:
+        """Browse exact-board releases; loading older pages reuses parsed pages.
+
+        Known split release shards are one family choice. Board-specific preview
+        releases are also selectable here, while choose_asset stays stable-only.
+        No file is downloaded until the caller selects a choice's asset.
+        """
+        if target_env is None and target_id is None:
+            raise ota.OtaError("Select an exact firmware target")
+        families: dict[str, list[_CatalogRelease]] = {}
+        for release in self._releases(max_pages):
+            if not include_prerelease and release.prerelease:
+                continue
+            family = _release_family(release.tag) or release.tag
+            families.setdefault(family, []).append(release)
+        choices = []
+        for family, releases in families.items():
+            assets = self._compatible([asset for release in releases for asset in release.assets],
+                                      target_env, target_id, profile)
+            if assets:
+                choices.append(ReleaseChoice(family, assets[0],
+                    any(release.prerelease for release in releases),
+                    max(release.published_at for release in releases)))
+        choices.sort(key=lambda choice: (choice.asset.version, choice.published_at, choice.release_tag), reverse=True)
+        choices.sort(key=lambda choice: choice.prerelease)
+        return choices
+
     def base_candidates(self, target_env: str | None = None, *, target_id: int | None = None,
                         fw_version: int | None = None, limit: int = 8) -> list[ReleaseAsset]:
         """Find possible old images; their downloaded EndF body hash MUST be checked."""
@@ -276,8 +345,8 @@ class GitHubCatalog:
             raise ota.OtaError("Select an exact firmware target")
         if not 1 <= limit <= 32:
             raise ota.OtaError("Base candidate limit must be between 1 and 32")
-        if self._assets is None:
-            self.fetch_assets()
+        if not self._pages:
+            self._releases(3)
         preference = {"mota": 0, "zip": 1, "bin": 2, "hex": 3}
         matches = [row for row in self._all_assets
                    if not row.embedded_bootloader

tools/lora_ota/ota_wizard_packages.py

@@ -88,7 +88,8 @@ def install_builder(cache: Path, progress=print) -> str | None:
 
 
 def find_running_image(destination: Destination, cache: Path, catalog: GitHubCatalog | None,
-                       *, supplied: Path | None = None, progress=print) -> Path | None:
+                       *, supplied: Path | None = None, progress=print,
+                       candidate_limit: int = 4) -> Path | None:
     images = ImageCache(cache)
     if supplied:
         identity = load_application(supplied, target_id=destination.target_id, hw_id=destination.hw_id)
@@ -102,7 +103,7 @@ def find_running_image(destination: Destination, cache: Path, catalog: GitHubCat
         return found
     if catalog:
         for asset in catalog.base_candidates(target_id=destination.target_id,
-                                             fw_version=destination.version, limit=4):
+                                             fw_version=destination.version, limit=candidate_limit):
             try:
                 candidate = download_asset(asset, cache / "downloads")
                 identity = load_application(candidate, target_id=destination.target_id,

tools/lora_ota/test_ota_wizard_catalog.py

@@ -92,6 +92,110 @@ class CatalogTest(unittest.TestCase):
         with self.assertRaises(ota.OtaError):
             source.choose_asset("another_repeater")
 
+    def test_release_menu_groups_shards_and_offers_exact_board_previews(self):
+        shard = f"lora-ota-10-{FAMILY}"
+        preview = "ikoka-v1.18.0.2-alpha2"
+        shard_asset = asset_row(tag=shard)
+        preview_asset = asset_row(f"{ENV}-full-ota-v1.18.0.2-alpha2.zip", preview)
+        old_asset = asset_row(f"{ENV}-full-ota-{OLD_FAMILY}.zip", OLD_FAMILY)
+        rows = [release(preview, [preview_asset], prerelease=True),
+                release(FAMILY, [asset_row(f"{ENV}-full-ota-{FAMILY}.bin")]),
+                release(shard, [shard_asset]),
+                release(f"full-profiles-{FAMILY}", [asset_row(
+                    f"{ENV}-reduced-ota-{FAMILY}.zip", f"full-profiles-{FAMILY}")]),
+                release(OLD_FAMILY, [old_asset]),
+                release("v9.0.0-draft", [asset_row()], draft=True),
+                release("another-v8.0.0", [asset_row(f"another_repeater-full-ota-v8.0.0.zip",
+                                                     "another-v8.0.0")])]
+        source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url))
+        choices = source.release_choices(target_id=TARGET)
+        self.assertEqual([choice.release_tag for choice in choices], [FAMILY, OLD_FAMILY, preview])
+        self.assertEqual(choices[0].asset.name, shard_asset["name"])
+        self.assertEqual(choices[0].asset.release_tag, shard)
+        self.assertFalse(choices[0].prerelease)
+        self.assertTrue(choices[-1].prerelease)
+        self.assertIn("prerelease", choices[-1].label)
+        self.assertEqual(source.choose_asset(target_id=TARGET), choices[0].asset)
+        self.assertEqual([choice.release_tag for choice in source.release_choices(
+            target_id=TARGET, include_prerelease=False)], [FAMILY, OLD_FAMILY])
+        self.assertFalse(source.has_more_releases)
+
+    def test_menu_prefers_full_and_does_not_offer_embedded_or_wrong_role(self):
+        rows = [release(FAMILY, [
+            asset_row(f"{ENV}-with-bootloader-2.4.12-full-ota-{FAMILY}.zip"),
+            asset_row(f"ikoka_stick_nrf_33dbm_room_server_lora_ota-full-ota-{FAMILY}.zip"),
+            asset_row(f"{ENV}-reduced-ota-{FAMILY}.zip"),
+            asset_row(f"{ENV}-ota-{FAMILY}.bin"), asset_row()])]
+        source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url))
+        choices = source.release_choices(ENV)
+        self.assertEqual(len(choices), 1)
+        self.assertEqual(choices[0].asset.profile, "full")
+        self.assertEqual(source.release_choices(ENV, profile="reduced")[0].asset.profile, "reduced")
+        self.assertEqual(source.release_choices("no_matching_repeater"), [])
+
+    def test_menu_loads_older_pages_once_and_adds_exact_base_candidates(self):
+        def page_rows(number):
+            if number == 1:
+                head = release(FAMILY, [asset_row()])
+            elif number == 2:
+                head = release(OLD_FAMILY, [asset_row(f"{ENV}-full-ota-{OLD_FAMILY}.zip", OLD_FAMILY)])
+            else:
+                return []
+            return [head] + [release(f"v0.0.0-empty-{number}-{index}") for index in range(9)]
+        calls = []
+        def opener(request, timeout):
+            number = int(request.full_url.rsplit("=", 1)[1])
+            calls.append(number)
+            return Response(json.dumps(page_rows(number)).encode(), request.full_url)
+        source = catalog.GitHubCatalog(opener=opener)
+        self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=1)), 1)
+        self.assertTrue(source.has_more_releases)
+        self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=3)), 2)
+        self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=6)), 2)
+        self.assertEqual(calls, [1, 2, 3])
+        self.assertFalse(source.has_more_releases)
+        self.assertEqual(source.base_candidates(target_id=TARGET, fw_version=0x01110109)[0].release_tag, OLD_FAMILY)
+        self.assertEqual(source.choose_asset(target_id=TARGET).release_tag, FAMILY)
+        self.assertEqual(calls, [1, 2, 3])
+
+    def test_preview_only_menu_still_supplies_an_exact_base_candidate(self):
+        tag = "ikoka-v1.18.0.2-alpha1"
+        rows = [release(tag, [asset_row(f"{ENV}-full-ota-v1.18.0.2-alpha1.zip", tag)], prerelease=True)]
+        source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url))
+        choices = source.release_choices(target_id=TARGET)
+        self.assertEqual(len(choices), 1)
+        self.assertTrue(choices[0].prerelease)
+        self.assertEqual(source.base_candidates(target_id=TARGET, fw_version=0x01120002), [choices[0].asset])
+        with self.assertRaisesRegex(ota.OtaError, "main firmware release"):
+            source.choose_asset(target_id=TARGET)
+
+    def test_menu_page_errors_are_reported_without_caching_partial_results(self):
+        calls = []
+        def opener(request, timeout):
+            page = int(request.full_url.rsplit("=", 1)[1])
+            calls.append(page)
+            if page == 1:
+                rows = [release(FAMILY, [asset_row()])] + [release(f"v0.0.0-empty-{index}") for index in range(9)]
+                return Response(json.dumps(rows).encode(), request.full_url)
+            return Response(b"not a release catalog" if calls.count(2) == 1 else b"[]", request.full_url)
+        source = catalog.GitHubCatalog(opener=opener)
+        self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=1)), 1)
+        with self.assertRaisesRegex(ota.OtaError, "invalid release catalog"):
+            source.release_choices(target_id=TARGET, max_pages=3)
+        self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=3)), 1)
+        self.assertEqual(calls, [1, 2, 2])
+
+    def test_menu_validates_target_page_limits_and_terminal_labels(self):
+        source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(b"[]", req.full_url))
+        with self.assertRaisesRegex(ota.OtaError, "exact firmware target"):
+            source.release_choices()
+        for limit in (0, 11, True, "3"):
+            with self.subTest(limit=limit), self.assertRaisesRegex(ota.OtaError, "page limit"):
+                source.release_choices(target_id=TARGET, max_pages=limit)
+        item = catalog.ReleaseChoice("v1.18.0.2\x1b[1m\n", catalog.parse_release_asset(asset_row(), FAMILY), True, "")
+        self.assertNotIn("\x1b", item.label)
+        self.assertNotIn("\n", item.label)
+
     def test_untrusted_asset_metadata_is_not_selectable(self):
         for mutation in ({"browser_download_url": "http://github.com/file.zip"},
                          {"browser_download_url": "https://example.com/file.zip"},

tools/lora_ota/test_ota_wizard_menu.py

@@ -0,0 +1,335 @@
+"""Exercise the normal menu journeys without radios, compilers, or downloads."""
+import asyncio
+from contextlib import ExitStack, redirect_stderr, redirect_stdout
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+import ota_wizard as wizard
+import ota_wizard_bundle as bundles
+import ota_wizard_packages as packages
+from ota_wizard_catalog import ReleaseChoice
+from test_ota_wizard_bundle import KEY, destination, image, signed_package
+
+
+class WizardMenuTests(unittest.TestCase):
+    def setUp(self):
+        self.files = tempfile.TemporaryDirectory(prefix="meshcore-wizard-menu-")
+        self.addCleanup(self.files.cleanup)
+        self.cache = Path(self.files.name)
+        self.stack = ExitStack()
+        self.addCleanup(self.stack.close)
+        self.stack.enter_context(mock.patch.object(wizard, "cache_directory", return_value=self.cache))
+        self.stack.enter_context(mock.patch.object(wizard.sys.stdin, "isatty", return_value=True))
+        self.output = io.StringIO()
+        self.stack.enter_context(redirect_stdout(self.output))
+        self.stack.enter_context(redirect_stderr(self.output))
+
+    def answers(self, values):
+        return mock.patch("builtins.input", side_effect=values)
+
+    def test_plain_launch_can_exit_without_usb_or_dependency_setup(self):
+        with self.answers(["6"]), \
+                mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \
+                mock.patch.object(wizard, "run", side_effect=AssertionError("radio access")):
+            self.assertEqual(wizard.main([]), 0)
+        self.assertIn("Update a nearby radio", self.output.getvalue())
+        self.assertIn("Prepare an update bundle on this computer", self.output.getvalue())
+
+    def test_update_and_check_need_only_menu_numbers(self):
+        for action, menu_number in (("update", "1"), ("check", "2")):
+            with self.subTest(action=action), self.answers([menu_number, "6"]), \
+                    mock.patch.object(wizard, "ensure_dependencies") as setup, \
+                    mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run:
+                self.assertEqual(wizard.main([]), 0)
+            run.assert_awaited_once()
+            options = run.await_args.args[0]
+            self.assertEqual(options.check, action == "check")
+            self.assertFalse(options.recover)
+            self.assertTrue(options.interactive_menu)
+            self.assertIsNone(options.port)
+            self.assertIsNone(options.target)
+            self.assertEqual(setup.call_args.args[1],
+                             ["--menu-action", action, "--cache", str(self.cache)])
+
+    def test_runtime_resume_runs_selected_action_before_showing_main_menu(self):
+        with self.answers(["6"]) as input_call, \
+                mock.patch.object(wizard, "ensure_dependencies"), \
+                mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run:
+            self.assertEqual(wizard.main(["--runtime-ready", "--menu-action", "check"]), 0)
+        self.assertTrue(run.await_args.args[0].check)
+        self.assertEqual(input_call.call_count, 1)  # Only the post-check Exit choice.
+
+    def test_failed_update_returns_to_menu_without_a_traceback(self):
+        with self.answers(["1", "6"]), mock.patch.object(wizard, "ensure_dependencies"), \
+                mock.patch.object(wizard, "run", new_callable=mock.AsyncMock,
+                                  side_effect=ota.OtaError("Connect a source radio")):
+            self.assertEqual(wizard.main([]), 0)
+        self.assertIn("Update stopped: Connect a source radio", self.output.getvalue())
+        self.assertNotIn("Traceback", self.output.getvalue())
+
+    def test_bad_menu_numbers_retry_and_back_cancels_before_usb_setup(self):
+        with self.answers(["nonsense", "0", "3", "", "6"]), \
+                mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")):
+            self.assertEqual(wizard.main([]), 0)
+        self.assertEqual(self.output.getvalue().count("Enter one of the numbers above."), 2)
+
+    def test_settings_round_trip_without_saving_environment_credentials(self):
+        source, download_cache = self.cache / "custom source", self.cache / "downloads elsewhere"
+        values = ["1", "2", "2", "2", "3", "2", "4", "COM12", "5", "/tools/motatool",
+                  "6", str(source), "7", "v1.18.0.4-custom", "8", str(download_cache), "9"]
+        options = wizard.arguments([])
+        with self.answers(values), mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "never save me"}):
+            wizard.settings_menu(options)
+        stored = json.loads(wizard.menu_settings_path().read_text())
+        self.assertNotIn("never save me", json.dumps(stored))
+        self.assertEqual(set(stored), {"full", "no_bootloader", "mode", "port", "motatool",
+                                      "source_tree", "version", "cache"})
+        restored = wizard.arguments([])
+        wizard.load_menu_settings(restored)
+        self.assertTrue(restored.full)
+        self.assertTrue(restored.no_bootloader)
+        self.assertEqual((restored.mode, restored.port), ("repeater", "COM12"))
+        self.assertEqual(restored.motatool, "/tools/motatool")
+        self.assertEqual(restored.source_tree, source)
+        self.assertEqual(restored.cache, download_cache)
+        self.assertEqual(restored.version, "v1.18.0.4-custom")
+
+    def test_blank_settings_can_restore_automatic_port_builder_and_version(self):
+        options = wizard.arguments([])
+        options.port, options.motatool, options.version = "COM12", "/tools/motatool", "v1.18.0.4-custom"
+        with self.answers(["4", "", "5", "", "7", "", "9"]):
+            wizard.settings_menu(options)
+        self.assertIsNone(options.port)
+        self.assertIsNone(options.motatool)
+        self.assertIsNone(options.version)
+
+    def test_menu_prepares_real_full_bundle_from_github_and_cached_request(self):
+        request = self.cache / "requests" / (KEY + ".json")
+        bundles.save_request(request, destination())
+        firmware = self.cache / "published.bin"
+        firmware.write_bytes(image().image)
+        asset = SimpleNamespace(release_tag="v1.18.0.2", name="ikoka-repeater-v1.18.0.2.zip")
+        catalog = SimpleNamespace(choose_asset=mock.Mock(return_value=asset))
+        with self.answers(["3", "1", "1", "", "6"]), \
+                mock.patch.object(wizard, "GitHubCatalog", return_value=catalog), \
+                mock.patch.object(wizard, "download_asset", return_value=firmware) as download, \
+                mock.patch.object(wizard, "small_host", return_value=True), \
+                mock.patch.object(wizard, "find_builder", return_value=None), \
+                mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \
+                mock.patch.object(wizard, "choose_port", side_effect=AssertionError("USB access")), \
+                mock.patch.object(wizard, "install_builder", side_effect=AssertionError("compiler setup")):
+            self.assertEqual(wizard.main([]), 0)
+        catalog.choose_asset.assert_called_once_with(target_id=destination().target_id)
+        download.assert_called_once_with(asset, self.cache / "downloads")
+        outputs = list((self.cache / "bundles").glob("*.ota.zip"))
+        self.assertEqual(len(outputs), 1)
+        prepared_image, package = bundles.load_bundle(outputs[0], destination())
+        self.assertEqual(prepared_image.image, image().image)
+        self.assertTrue(package.info.is_full)
+        self.assertEqual(package.body_hash, image().body_hash)
+        self.assertIn("Local firmware file or prepared bundle", self.output.getvalue())
+
+    def test_recovery_menu_lists_only_unfinished_records_and_supplies_full_key(self):
+        folder = self.cache / "sessions"
+        folder.mkdir()
+        (folder / (KEY + ".json")).write_text(json.dumps(
+            {"format": 1, "target_key": KEY, "phase": "needs cleanup"}))
+        (folder / ("22" * 32 + ".json")).write_text(json.dumps(
+            {"format": 1, "target_key": "22" * 32, "phase": "finished"}))
+        (folder / "invalid.json").write_text("not json")
+        with self.answers(["4", "1", "6"]) as input_call, \
+                mock.patch.object(wizard, "ensure_dependencies"), \
+                mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run:
+            self.assertEqual(wizard.main([]), 0)
+        self.assertEqual(input_call.call_count, 3)
+        self.assertTrue(run.await_args.args[0].recover)
+        self.assertEqual(run.await_args.args[0].target, KEY)
+        self.assertIn(KEY[:12].upper(), self.output.getvalue())
+        self.assertNotIn(KEY.upper(), self.output.getvalue())
+        self.assertNotIn("finished", self.output.getvalue())
+
+    def test_release_browser_can_load_older_pages_and_marks_previews(self):
+        preview_asset, stable_asset = object(), object()
+        preview = ReleaseChoice("v1.18.0.3-alpha1", preview_asset, True, "2026-10-26")
+        stable = ReleaseChoice("v1.18.0.2", stable_asset, False, "2026-10-25")
+        catalog = SimpleNamespace(has_more_releases=True, release_choices=mock.Mock(
+            side_effect=[[preview], [stable, preview]]))
+        with self.answers(["2", "1"]):
+            chosen = asyncio.run(wizard.choose_release(catalog, destination()))
+        self.assertIs(chosen, stable_asset)
+        self.assertEqual([call.kwargs["max_pages"] for call in catalog.release_choices.call_args_list], [3, 6])
+        self.assertTrue(all(call.kwargs["include_prerelease"] for call in catalog.release_choices.call_args_list))
+        self.assertIn("v1.18.0.3-alpha1 (prerelease)", self.output.getvalue())
+
+    def test_firmware_back_does_not_download_or_open_a_file(self):
+        with self.answers(["5"]), mock.patch.object(wizard, "small_host", return_value=False), \
+                mock.patch.object(wizard, "download_asset", side_effect=AssertionError("download")), \
+                mock.patch.object(wizard, "file_prompt", side_effect=AssertionError("file")), \
+                self.assertRaises(wizard.MenuCancelled):
+            asyncio.run(wizard.choose_firmware(wizard.arguments([]), destination(), object(), None, {}))
+
+    def test_remembered_file_has_an_explicit_back_choice(self):
+        previous = self.cache / "previous.bin"
+        previous.write_bytes(image().image)
+        with self.answers(["3"]) as input_call, self.assertRaises(wizard.MenuCancelled):
+            wizard.file_prompt("Firmware file", str(previous))
+        self.assertEqual(input_call.call_count, 1)
+        self.assertIn("Choose another file", self.output.getvalue())
+        self.assertIn("Back", self.output.getvalue())
+
+    def test_missing_remembered_file_can_cancel_instead_of_retrying_stale_default(self):
+        missing = self.cache / "removed.bin"
+        with self.answers([""]) as input_call, self.assertRaises(wizard.MenuCancelled):
+            wizard.file_prompt("Firmware file", str(missing))
+        self.assertEqual(input_call.call_count, 1)
+        self.assertNotIn("Use " + str(missing), self.output.getvalue())
+
+    def test_previous_boards_local_file_does_not_change_new_board_default(self):
+        previous = self.cache / "other-board.bin"
+        previous.write_bytes(b"other board")
+        published = self.cache / "matching.bin"
+        asset = SimpleNamespace(release_tag="v1.18.0.2", name="matching-board.zip")
+        remembered = {"target": "22" * 32, "firmware_source": "local", "firmware_file": str(previous)}
+        with self.answers([""]), mock.patch.object(wizard, "small_host", return_value=True), \
+                mock.patch.object(wizard, "download_asset", return_value=published) as download, \
+                mock.patch.object(wizard, "file_prompt", side_effect=AssertionError("other board's file")):
+            selected, source = asyncio.run(wizard.choose_firmware(
+                wizard.arguments([]), destination(), object(), asset, remembered))
+        self.assertEqual((selected, source), (published, "latest"))
+        download.assert_called_once_with(asset, self.cache / "downloads")
+
+    def test_local_firmware_selection_does_not_query_github(self):
+        local = self.cache / "local.bin"
+        local.write_bytes(image().image)
+        catalog = SimpleNamespace(choose_asset=mock.Mock(side_effect=AssertionError("GitHub query")),
+                                  release_choices=mock.Mock(side_effect=AssertionError("GitHub query")))
+        with self.answers(["3", str(local)]), mock.patch.object(wizard, "small_host", return_value=True), \
+                mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock,
+                                  side_effect=AssertionError("GitHub query")), \
+                mock.patch.object(wizard, "download_asset", side_effect=AssertionError("GitHub download")):
+            selected, source = asyncio.run(wizard.choose_firmware(
+                wizard.arguments([]), destination(), catalog, None, {}))
+        self.assertEqual((selected, source), (local, "local"))
+        catalog.choose_asset.assert_not_called()
+        catalog.release_choices.assert_not_called()
+
+    def test_release_back_returns_to_firmware_menu_then_accepts_local_file(self):
+        local = self.cache / "local.bin"
+        local.write_bytes(image().image)
+        preview = ReleaseChoice("v1.18.0.3-alpha1", object(), True, "2026-10-26")
+        catalog = SimpleNamespace(has_more_releases=False,
+                                  release_choices=mock.Mock(return_value=[preview]))
+        with self.answers(["2", "2", "3", str(local)]), \
+                mock.patch.object(wizard, "small_host", return_value=True), \
+                mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock,
+                                  side_effect=AssertionError("latest release query")), \
+                mock.patch.object(wizard, "download_asset", side_effect=AssertionError("preview download")):
+            selected, source = asyncio.run(wizard.choose_firmware(
+                wizard.arguments([]), destination(), catalog, None, {}))
+        self.assertEqual((selected, source), (local, "local"))
+        catalog.release_choices.assert_called_once_with(
+            target_id=destination().target_id, include_prerelease=True, max_pages=3)
+        self.assertIn("v1.18.0.3-alpha1 (prerelease)", self.output.getvalue())
+
+    def test_signed_preparation_installs_verifier_then_retries_the_same_file(self):
+        options = wizard.arguments([])
+        chosen = self.cache / "signed-full.mota"
+        new_image, package = image(), signed_package()
+        chosen.write_bytes(package.blob)
+        catalog = object()
+        with mock.patch.object(wizard, "choose_request", return_value=destination()), \
+                mock.patch.object(wizard, "GitHubCatalog", return_value=catalog), \
+                mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, return_value=None), \
+                mock.patch.object(wizard, "choose_firmware", new_callable=mock.AsyncMock,
+                                  return_value=(chosen, "local")) as choose, \
+                mock.patch.object(wizard, "prepare_selected_firmware", new_callable=mock.AsyncMock,
+                                  side_effect=[ImportError("signature verifier missing"),
+                                               (new_image, package)]) as prepare, \
+                mock.patch.object(wizard, "ensure_signature_support") as verifier, \
+                mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \
+                mock.patch.object(wizard, "choose_port", side_effect=AssertionError("USB access")), \
+                mock.patch.object(wizard, "ask", side_effect=lambda prompt, default=None: default):
+            self.assertEqual(asyncio.run(wizard.prepare_interactively(options)), 0)
+        choose.assert_awaited_once()
+        verifier.assert_called_once_with()
+        self.assertEqual(prepare.await_args_list, [mock.call(options, destination(), chosen, catalog)] * 2)
+        outputs = list((self.cache / "bundles").glob("*.ota.zip"))
+        self.assertEqual(len(outputs), 1)
+        _, prepared = bundles.load_bundle(outputs[0], destination())
+        self.assertEqual(prepared.blob, package.blob)
+        self.assertEqual(prepared.signer, package.signer)
+
+    def test_failed_signature_setup_does_not_reselect_or_write_a_bundle(self):
+        options = wizard.arguments([])
+        chosen = self.cache / "signed-full.mota"
+        with mock.patch.object(wizard, "choose_request", return_value=destination()), \
+                mock.patch.object(wizard, "GitHubCatalog", return_value=object()), \
+                mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, return_value=None), \
+                mock.patch.object(wizard, "choose_firmware", new_callable=mock.AsyncMock,
+                                  return_value=(chosen, "local")) as choose, \
+                mock.patch.object(wizard, "prepare_selected_firmware", new_callable=mock.AsyncMock,
+                                  side_effect=ImportError("signature verifier missing")) as prepare, \
+                mock.patch.object(wizard, "ensure_signature_support",
+                                  side_effect=ota.OtaError("Verifier download failed")) as verifier, \
+                mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \
+                mock.patch.object(wizard, "save_bundle", side_effect=AssertionError("partial bundle")), \
+                self.assertRaisesRegex(ota.OtaError, "Verifier download failed"):
+            asyncio.run(wizard.prepare_interactively(options))
+        choose.assert_awaited_once()
+        prepare.assert_awaited_once()
+        verifier.assert_called_once_with()
+        self.assertFalse((self.cache / "bundles").exists())
+
+    def test_local_base_fallback_rejects_wrong_image_then_accepts_exact_running_image(self):
+        wrong, correct = self.cache / "wrong.bin", self.cache / "running.bin"
+        wrong.write_bytes(image().image)
+        expected = image(0x01120001, body=b"original application" * 16)
+        correct.write_bytes(expected.image)
+        with self.answers(["2", "2"]), \
+                mock.patch.object(wizard, "file_prompt", side_effect=[wrong, correct]), \
+                mock.patch.object(wizard, "GitHubCatalog", side_effect=AssertionError("network")):
+            base = asyncio.run(wizard.choose_missing_base(wizard.arguments([]), destination(), object()))
+        self.assertIsNotNone(base)
+        self.assertEqual(ota.parse_endf(base.read_bytes()).body_hash, destination().base_hash)
+        self.assertIn("not the exact running firmware", self.output.getvalue())
+
+    def test_older_github_base_search_checks_past_four_same_version_candidates(self):
+        candidates, paths = [object() for _ in range(5)], []
+        for index in range(4):
+            path = self.cache / f"other-build-{index}.bin"
+            path.write_bytes(image(0x01120001, body=bytes([index])*64).image)
+            paths.append(path)
+        exact = self.cache / "running.bin"
+        exact.write_bytes(image(0x01120001, body=b"original application"*16).image)
+        paths.append(exact)
+        catalog = SimpleNamespace(has_more_releases=True, release_choices=mock.Mock(return_value=[]),
+            base_candidates=mock.Mock(side_effect=lambda **kwargs: candidates[:kwargs["limit"]]))
+        with self.answers(["1"]), mock.patch.object(packages, "download_asset",
+                side_effect=lambda asset, cache: paths[candidates.index(asset)]) as download:
+            base = asyncio.run(wizard.choose_missing_base(wizard.arguments([]), destination(), catalog))
+        catalog.release_choices.assert_called_once_with(
+            target_id=destination().target_id, include_prerelease=True, max_pages=6)
+        catalog.base_candidates.assert_called_once_with(
+            target_id=destination().target_id, fw_version=destination().version, limit=16)
+        self.assertEqual(download.call_count, 5)
+        self.assertEqual(ota.parse_endf(base.read_bytes()).body_hash, destination().base_hash)
+
+    def test_existing_automation_flags_do_not_enter_main_menu(self):
+        with mock.patch.object(wizard, "menu_main", side_effect=AssertionError("interactive menu")), \
+                mock.patch.object(wizard, "ensure_dependencies") as setup, \
+                mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run:
+            self.assertEqual(wizard.main(["--check", "--port", "COM7"]), 0)
+        setup.assert_called_once()
+        self.assertTrue(run.await_args.args[0].check)
+        self.assertEqual(run.await_args.args[0].port, "COM7")
+
+
+if __name__ == "__main__":
+    unittest.main()