Restore OTA deployment tuning

docs/rak3401_mota_chain.md

@@ -119,12 +119,17 @@ python3 tools/lora_ota/rak3401_mota_chain.py \
   --target-key 63d8df63 \
   --temp-radio 909.950,500,5,5,120 \
   --ota-hops 0 \
+  --legacy-full-airtime \
   --motatool /path/to/motatool \
   --yes
 ```
 
 Keep the work directory. Rerunning the same command resumes only when the live
 EndF body hash matches an exact chain node. Never manually skip a package.
+`--legacy-full-airtime` temporarily sets the destination airtime factor to
+zero and restores it at the endpoint. Use that option only where the selected
+frequency and local duty-cycle rules permit a full transmit budget; omit it
+otherwise.
 
 ## Manual operation
 
@@ -157,6 +162,7 @@ get system.watchdog
 get radio.rxps
 powersaving
 get rxdelay
+get af
 ota config
 ```
 
@@ -183,6 +189,12 @@ ota config hops 0
 tempradio 909.950,500,5,5,120
 ```
 
+Where local duty-cycle rules permit it, `set af 0` can also remove the legacy
+firmware's saved airtime wait during this bounded maintenance window. Record
+`get af` first and restore that exact value after the final step. Current
+firmware grants the bounded TempRadio transfer budget without overwriting the
+saved airtime factor.
+
 `powersaving off` is an isolation guardrail, not the fix for the observed
 failure. The actual source failure was an oversized USB CDC reply; current
 firmware splits those reads. Some historical bridge builds may report RXPS as
@@ -274,14 +286,27 @@ airtime and duty spacing, clamped to 100-1000 ms. These two delays solve
 different problems.
 
 After step 9 is proven, restore each saved destination and relay value exactly,
-then restore normal radio operation. Re-enable the RAK system watchdog last:
+including `rxdelay`, RXPS, CPU power saving, `af`, `ota config hops`, and relay
+timing. Ordinary repeater firmware does not implement `normalradio`. While the
+node is still reachable on the temporary channel, shorten its lease to one
+minute with the same tuple:
+
+```text
+tempradio 909.950,500,5,5,1
+```
+
+Wait for the lease to expire, return the controller and ordinary source to
+their saved normal channels, and then re-enable the RAK system watchdog:
 
 ```text
-normalradio
 set system.watchdog on
 get system.watchdog
 ```
 
+A Full Companion source is the exception: its local TCP console supports
+`normalradio` and the automated runner uses it before restoring the shared
+Binary API radio tuple.
+
 Final success requires version `v1.17.1.02-halo-keymind-cascade-dev-e742333a`,
 body hash `4BB1526BF647547D`, target `2FA509C1`, hardware `RAK_3401`, and
 `get system.watchdog` reporting `> on`. If a relay cannot be restored before
@@ -323,11 +348,13 @@ completes still requires local USB recovery because the deployed start image
 predates the rescue command.
 
 Before the first mutation, the runner saves the destination's RXPS periods,
-CPU power-saving state, and RX flood delay in the persistent work directory.
-It verifies `radio.rxps off`, `powersaving off`, and `rxdelay 0` after every
-bridge reboot, then restores the original values only after the exact endpoint
-is proven. An interrupted run deliberately leaves those transfer guardrails
-active; rerunning with the same work directory resumes and restores them.
+CPU power-saving state, RX flood delay, airtime factor, and OTA hop reach in
+the persistent work directory. It verifies `radio.rxps off`, `powersaving off`,
+and `rxdelay 0` after every bridge reboot, plus `af 0` when
+`--legacy-full-airtime` was explicitly selected. It restores every original
+value only after the exact endpoint is proven. An interrupted run deliberately
+leaves those transfer guardrails active; rerun the same command with the same
+work directory to resume and restore them.
 CPU power saving was not the cause of the observed folder-source failure: that
 was a USB CDC receive-ring overrun fixed by bounded serial reads. The runtime
 settings remain conservative isolation and fast-link reliability guardrails.

tools/lora_ota/rak3401_mota_chain.py

@@ -384,6 +384,8 @@ class TargetTransferSettings:
     rxps_sleep_us: int
     powersaving_enabled: bool
     rxdelay: str
+    airtime_factor: str
+    ota_hops: int
 
 
 def sha256_file(path: Path) -> str:
@@ -927,12 +929,29 @@ def read_target_transfer_settings(
     if rxdelay_match is None:
         raise ota.OtaError(f"could not read destination RX delay: {rxdelay_reply}")
 
+    airtime_reply = controller.remote_command(target_name, "get af")
+    airtime_match = re.fullmatch(
+        r"\s*>\s*([0-9]+(?:\.[0-9]+)?)\s*",
+        airtime_reply,
+    )
+    if airtime_match is None:
+        raise ota.OtaError(
+            f"could not read destination airtime factor: {airtime_reply}"
+        )
+    airtime_factor = float(airtime_match.group(1))
+    if not math.isfinite(airtime_factor) or airtime_factor < 0.0:
+        raise ota.OtaError(
+            f"destination returned an invalid airtime factor: {airtime_reply}"
+        )
+
     return TargetTransferSettings(
         rxps_enabled=rxps_match.group(1).lower() == "on",
         rxps_rx_us=int(rxps_match.group(2)),
         rxps_sleep_us=int(rxps_match.group(3)),
         powersaving_enabled=powersaving_match.group(1).lower() == "on",
         rxdelay=rxdelay_match.group(1),
+        airtime_factor=airtime_match.group(1),
+        ota_hops=read_ota_hops(controller, target_name),
     )
 
 
@@ -954,14 +973,25 @@ def load_or_capture_transfer_settings(
                 saved["powersaving_enabled"], bool
             ):
                 raise TypeError("saved power states must be JSON booleans")
+            if isinstance(saved["ota_hops"], bool) or not isinstance(
+                saved["ota_hops"], int
+            ):
+                raise TypeError("saved OTA hop reach must be a JSON integer")
             settings = TargetTransferSettings(
                 rxps_enabled=saved["rxps_enabled"],
                 rxps_rx_us=int(saved["rxps_rx_us"]),
                 rxps_sleep_us=int(saved["rxps_sleep_us"]),
                 powersaving_enabled=saved["powersaving_enabled"],
                 rxdelay=str(saved["rxdelay"]),
+                airtime_factor=str(saved["airtime_factor"]),
+                ota_hops=saved["ota_hops"],
             )
             float(settings.rxdelay)
+            airtime_factor = float(settings.airtime_factor)
+            if not math.isfinite(airtime_factor) or airtime_factor < 0.0:
+                raise ValueError("saved airtime factor is invalid")
+            if not 0 <= settings.ota_hops <= 8:
+                raise ValueError("saved OTA hop reach is invalid")
         except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
             raise ota.OtaError(f"invalid saved transfer settings in {path}") from exc
         print(f"[guardrail] loaded original destination settings from {path}")
@@ -975,6 +1005,8 @@ def load_or_capture_transfer_settings(
         "rxps_sleep_us": settings.rxps_sleep_us,
         "powersaving_enabled": settings.powersaving_enabled,
         "rxdelay": settings.rxdelay,
+        "airtime_factor": settings.airtime_factor,
+        "ota_hops": settings.ota_hops,
     }
     temporary = path.with_suffix(path.suffix + ".tmp")
     temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="ascii")
@@ -987,6 +1019,8 @@ def load_or_capture_transfer_settings(
 def enforce_transfer_guardrails(
     controller: ota.Controller,
     target_name: str,
+    *,
+    legacy_full_airtime: bool = False,
 ) -> None:
     current = read_target_transfer_settings(controller, target_name)
     if current.powersaving_enabled:
@@ -1001,18 +1035,30 @@ def enforce_transfer_guardrails(
         reply = controller.remote_command(target_name, "set rxdelay 0")
         if not reply.upper().startswith("OK"):
             raise ota.OtaError(f"target did not disable RX flood delay: {reply}")
+    if legacy_full_airtime and abs(float(current.airtime_factor)) > 0.0001:
+        reply = controller.remote_command(target_name, "set af 0")
+        if not reply.upper().startswith("OK"):
+            raise ota.OtaError(f"target did not enable full legacy airtime: {reply}")
 
     verified = read_target_transfer_settings(controller, target_name)
     if (
         verified.powersaving_enabled
         or verified.rxps_enabled
         or abs(float(verified.rxdelay)) > 0.0001
+        or (
+            legacy_full_airtime
+            and abs(float(verified.airtime_factor)) > 0.0001
+        )
     ):
         raise ota.OtaError(
             "destination transfer guardrails did not read back as "
             "powersaving=off, RXPS=off, rxdelay=0"
+            + (", af=0" if legacy_full_airtime else "")
         )
-    print("[guardrail] destination verified: RXPS off, rxdelay 0, CPU power saving off")
+    detail = "RXPS off, rxdelay 0, CPU power saving off"
+    if legacy_full_airtime:
+        detail += ", legacy airtime factor 0"
+    print(f"[guardrail] destination verified: {detail}")
 
 
 def restore_transfer_settings(
@@ -1043,10 +1089,28 @@ def restore_transfer_settings(
         if re.search(r"\boff\b", reply, re.IGNORECASE) is None:
             raise ota.OtaError(f"target did not restore RXPS-off state: {reply}")
 
+    if abs(float(current.airtime_factor) - float(saved.airtime_factor)) > 0.0001:
+        reply = controller.remote_command(
+            target_name, f"set af {saved.airtime_factor}"
+        )
+        if not reply.upper().startswith("OK"):
+            raise ota.OtaError(f"target did not restore airtime factor: {reply}")
+
+    if current.ota_hops != saved.ota_hops:
+        reply = controller.remote_command(
+            target_name, f"ota config hops {saved.ota_hops}"
+        )
+        if not reply.startswith(f"OK OTA reach = {saved.ota_hops} hop"):
+            raise ota.OtaError(f"target did not restore OTA hop policy: {reply}")
+
     verified = read_target_transfer_settings(controller, target_name)
     if (
         verified.rxps_enabled != saved.rxps_enabled
         or abs(float(verified.rxdelay) - float(saved.rxdelay)) > 0.0001
+        or abs(
+            float(verified.airtime_factor) - float(saved.airtime_factor)
+        ) > 0.0001
+        or verified.ota_hops != saved.ota_hops
         or (
             saved.rxps_enabled
             and (
@@ -1339,6 +1403,14 @@ def build_parser() -> argparse.ArgumentParser:
         default=3,
         help="destination OTA receive/relay reach to enforce before every step",
     )
+    parser.add_argument(
+        "--legacy-full-airtime",
+        action="store_true",
+        help=(
+            "temporarily set destination af=0 for legacy TempRadio firmware; "
+            "use only where the selected frequency and local rules allow it"
+        ),
+    )
     parser.add_argument("--controller-baud", type=int, default=115200)
     parser.add_argument("--source-baud", type=int, default=115200)
     parser.add_argument("--meshcli", default="meshcli")
@@ -1469,8 +1541,19 @@ def confirm_chain(
             f"  relay timing: rxdelay 0, txdelay "
             f"{ota.format_decimal(args.relay_txdelay)} (saved/restored)"
         )
-    print(f"  OTA reach   : enforce {args.ota_hops} hops before every step")
-    print("  guardrails  : save, disable RXPS/rxdelay/CPU sleep, restore at endpoint")
+    print(
+        f"  OTA reach   : enforce {args.ota_hops} hops before every step, "
+        "then restore"
+    )
+    print(
+        "  guardrails  : save, disable RXPS/rxdelay/CPU sleep, restore all "
+        "settings at endpoint"
+    )
+    if args.legacy_full_airtime:
+        print(
+            "  full airtime: temporarily set af=0; operator accepts local "
+            "duty-cycle responsibility"
+        )
     print("  watchdog    : disable, prove stable, gate every install, then re-enable")
     if args.preflight_only:
         print("Live preflight passed; no radio or watchdog settings were changed.")
@@ -1557,7 +1640,6 @@ def main(argv: list[str] | None = None) -> int:
             return 0
 
         if first_index == len(steps):
-            enforce_ota_hops(controller, target_name, args.ota_hops)
             transfer_path = work_dir / TRANSFER_SETTINGS_FILE
             if transfer_path.exists():
                 transfer_settings = load_or_capture_transfer_settings(
@@ -1578,7 +1660,11 @@ def main(argv: list[str] | None = None) -> int:
             controller, target_name, full_key, work_dir
         )
         prepare_watchdog(controller, target_name)
-        enforce_transfer_guardrails(controller, target_name)
+        enforce_transfer_guardrails(
+            controller,
+            target_name,
+            legacy_full_airtime=args.legacy_full_airtime,
+        )
         enforce_ota_hops(controller, target_name, args.ota_hops)
         for index in range(first_index, len(steps)):
             step = steps[index]
@@ -1595,7 +1681,11 @@ def main(argv: list[str] | None = None) -> int:
                     f"live target is at chain index {current_index}, expected {index}"
                 )
             require_watchdog_state(controller, target_name, "off")
-            enforce_transfer_guardrails(controller, target_name)
+            enforce_transfer_guardrails(
+                controller,
+                target_name,
+                legacy_full_airtime=args.legacy_full_airtime,
+            )
             enforce_ota_hops(controller, target_name, args.ota_hops)
             print(
                 f"\n[chain] step {step.number:02d}/{len(steps)}: "
@@ -1670,7 +1760,7 @@ def main(argv: list[str] | None = None) -> int:
         print(
             "The target watchdog and transfer guardrails may intentionally remain off. "
             "Rerun the same command and work directory to resume and restore the saved "
-            "settings; do not skip a chain step.",
+            "settings, including OTA reach and airtime factor; do not skip a chain step.",
             file=sys.stderr,
         )
         return 2

tools/lora_ota/rebuild_rak3401_bundle.py

@@ -378,7 +378,10 @@ ZIP and a persistent work directory. The runner enforces exact target, start
 body hash, package order, watchdog state, post-boot identity, and normal-radio
 restoration.
 
-Direct bench: `--temp-radio 909.950,500,5,5,120 --ota-hops 0`.
+Direct bench: use `--temp-radio 909.950,500,5,5,120 --ota-hops 0`. Add
+`--legacy-full-airtime` only where the frequency and local duty-cycle rules
+permit it; omit it otherwise. The runner saves and restores the original
+airtime factor and OTA hop reach.
 
 Live private network with two intermediate relays: use
 `--temp-radio 909.950,250,5,5,120 --ota-hops 3`, plus two `--relay` arguments

tools/lora_ota/test_lora_ota.py

@@ -163,6 +163,11 @@ class FormatTests(unittest.TestCase):
         chain = rak_chain.build_parser().parse_args([])
         self.assertEqual(generic.temp_radio, "909.950,250,5,5,120")
         self.assertEqual(chain.temp_radio, "909.950,250,5,5,120")
+        self.assertFalse(chain.legacy_full_airtime)
+
+    def test_rak_chain_full_airtime_is_explicit(self) -> None:
+        args = rak_chain.build_parser().parse_args(["--legacy-full-airtime"])
+        self.assertTrue(args.legacy_full_airtime)
 
     def test_offline_sd_nrf52_does_not_require_a_base_hash(self) -> None:
         parser = ota.build_parser()
@@ -1572,6 +1577,8 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
             self.rxps_sleep_us = 60000
             self.powersaving_enabled = True
             self.rxdelay = "2.0"
+            self.airtime_factor = "1.0"
+            self.ota_hops = 3
             self.commands: list[str] = []
 
         def remote_command(self, _target: str, command: str) -> str:
@@ -1601,16 +1608,31 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
             if command.startswith("set rxdelay "):
                 self.rxdelay = command.split(maxsplit=2)[2]
                 return "OK"
+            if command == "get af":
+                return f"> {self.airtime_factor}"
+            if command.startswith("set af "):
+                self.airtime_factor = command.split(maxsplit=2)[2]
+                return "OK"
+            if command == "ota config":
+                return f"ota config: checkpoint=4 advert=1440min hops={self.ota_hops}"
+            if command.startswith("ota config hops "):
+                self.ota_hops = int(command.rsplit(maxsplit=1)[1])
+                return f"OK OTA reach = {self.ota_hops} hops (saved)"
             raise AssertionError(command)
 
     def test_guardrails_disable_and_restore_all_original_settings(self) -> None:
         controller = self.Controller()
         saved = rak_chain.read_target_transfer_settings(controller, "remote")
 
-        rak_chain.enforce_transfer_guardrails(controller, "remote")
+        rak_chain.enforce_transfer_guardrails(
+            controller, "remote", legacy_full_airtime=True
+        )
+        rak_chain.enforce_ota_hops(controller, "remote", 0)
         self.assertFalse(controller.rxps_enabled)
         self.assertFalse(controller.powersaving_enabled)
         self.assertEqual(controller.rxdelay, "0")
+        self.assertEqual(controller.airtime_factor, "0")
+        self.assertEqual(controller.ota_hops, 0)
 
         rak_chain.restore_transfer_settings(controller, "remote", saved)
         self.assertTrue(controller.rxps_enabled)
@@ -1618,8 +1640,18 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
         self.assertEqual(controller.rxps_sleep_us, 60000)
         self.assertTrue(controller.powersaving_enabled)
         self.assertEqual(controller.rxdelay, "2.0")
+        self.assertEqual(controller.airtime_factor, "1.0")
+        self.assertEqual(controller.ota_hops, 3)
         self.assertEqual(controller.commands[-1], "powersaving on")
 
+    def test_guardrails_preserve_airtime_without_opt_in(self) -> None:
+        controller = self.Controller()
+
+        rak_chain.enforce_transfer_guardrails(controller, "remote")
+
+        self.assertEqual(controller.airtime_factor, "1.0")
+        self.assertNotIn("set af 0", controller.commands)
+
     def test_original_settings_are_persisted_for_resume(self) -> None:
         controller = self.Controller()
         with tempfile.TemporaryDirectory() as directory:
@@ -1629,12 +1661,16 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
             )
             command_count = len(controller.commands)
             controller.rxdelay = "0"
+            controller.airtime_factor = "0"
+            controller.ota_hops = 0
             loaded = rak_chain.load_or_capture_transfer_settings(
                 controller, "remote", "AA" * 32, work_dir
             )
             mode = (work_dir / rak_chain.TRANSFER_SETTINGS_FILE).stat().st_mode
 
         self.assertEqual(loaded, saved)
+        self.assertEqual(loaded.airtime_factor, "1.0")
+        self.assertEqual(loaded.ota_hops, 3)
         self.assertEqual(len(controller.commands), command_count)
         self.assertEqual(mode & 0o777, 0o600)
 
@@ -1652,6 +1688,8 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
                         "rxps_sleep_us": 60000,
                         "powersaving_enabled": False,
                         "rxdelay": "2.0",
+                        "airtime_factor": "1.0",
+                        "ota_hops": 3,
                     }
                 ),
                 encoding="ascii",
@@ -1661,6 +1699,31 @@ class Rak3401TransferGuardrailTests(unittest.TestCase):
                     controller, "remote", "AA" * 32, work_dir
                 )
 
+    def test_saved_settings_without_restore_fields_are_rejected(self) -> None:
+        controller = self.Controller()
+        with tempfile.TemporaryDirectory() as directory:
+            work_dir = Path(directory)
+            path = work_dir / rak_chain.TRANSFER_SETTINGS_FILE
+            path.write_text(
+                json.dumps(
+                    {
+                        "target_key": "aa" * 32,
+                        "rxps_enabled": False,
+                        "rxps_rx_us": 65625,
+                        "rxps_sleep_us": 60000,
+                        "powersaving_enabled": False,
+                        "rxdelay": "2.0",
+                    }
+                ),
+                encoding="ascii",
+            )
+            with self.assertRaisesRegex(
+                ota.OtaError, "invalid saved transfer settings"
+            ):
+                rak_chain.load_or_capture_transfer_settings(
+                    controller, "remote", "AA" * 32, work_dir
+                )
+
 
 class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
     def test_chain_resume_uses_exact_body_hash_not_runtime_label(self) -> None: