Support signed fleet radio schedules and clock updates

.github/workflows/run-unit-tests.yml

@@ -107,6 +107,8 @@ jobs:
           python3 -B test/test_companion_fleet_control.py -v
           python3 -B test/test_fleet_channel_runtime.py -v
           python3 -B test/test_fleet_role_wiring.py -v
+          python3 -B test/test_fleet_radio_schedules.py -v
+          python3 -B test/test_fleet_clock_control.py -v
 
       - name: Verify OTA channel persistence and published image tags
         run: |

README.md

@@ -3,7 +3,7 @@
 * **[Two LoRa profiles on one device](https://mikecarper.github.io/MeshCore/radio_profiles/).** Time-share one transceiver between two frequencies and/or modulation settings. Choose transmission profiles per contact or channel, enable or isolate crossover traffic, schedule profiles, and use a temporary update profile alongside the normal network. Includes timing calibration and separate R2 screen pages.
 * **[Firmware updates over LoRa](https://mikecarper.github.io/MeshCore/lora_ota_automation/).** Update repeaters, room servers, and sensors over LoRa on supported firmware and storage layouts.
 * **[Repeater forwarding policies](https://mikecarper.github.io/MeshCore/flood_filtering/).** Actions include dropping, rate limiting, assigning scopes, selecting retries, and controlling forwarding priority. Separate policies cover ordinary radio forwarding, bridges, and profile crossover.
-* **[Private fleet management](https://mikecarper.github.io/MeshCore/fleet_management/).** Deploy filtering and secondary-radio commands to a whole fleet, one node, or a comma-separated list of public-key targets. The publisher signs commands; receivers hold its public verification key, with replay protection and acknowledgements before radio handoff.
+* **[Private fleet management](https://mikecarper.github.io/MeshCore/fleet_management/).** Deploy filtering, secondary-radio changes, radio schedules, and clock updates to a whole fleet, one node, or a comma-separated list of public-key targets. The publisher signs commands; receivers hold its public verification key, with replay protection and acknowledgements before radio handoff.
 * **[Flood retries that account for network topology](https://mikecarper.github.io/MeshCore/halo_keymind_settings/#flood-and-advert-settings).** Configure retry counts, path limits, packet-type limits, target prefixes, ignored repeaters, and channel-specific eligibility. Bridge buckets keep retrying until traffic is heard on the intended sides of a relay or the retry budget expires. Qualifying echoes cancel unnecessary attempts.
 * **[Enhanced direct messages and direct-route retries](https://mikecarper.github.io/MeshCore/halo_keymind_settings/#direct-retry-settings).** Adapt the coding rate across bounded retry attempts and cancel unnecessary retries when a qualifying echo is heard. [Single Key DMs](https://mikecarper.github.io/MeshCore/companion_protocol/#private-text-when-only-the-sender-knows-the-recipient) let compatible Companions send private text when only the sender knows the recipient's key, using a signed, encrypted introduction. The recipient accepts the sender through the app or opts into automatic acceptance; compact STM32WL builds omit this feature.
 * **[Enhanced telemetry history](https://mikecarper.github.io/MeshCore/telemetry_decoder/).** Send compact temperature and battery-voltage snapshots of up to 165 samples per packet, plus external-voltage history in chunks. Read retained GPS history through the CLI; GPS history is not included in the automatic telemetry packets. Decode snapshots and CLI history in the browser.

docs/fleet_management.md

@@ -132,6 +132,9 @@ or normal terminal for longer lists.
 | --- | --- |
 | Saved second profile | `get radio2`, `set radio2 <tuple|off>` |
 | Temporary second profile | `get tempradio2`, `set tempradio2 <tuple|off>` |
+| Scheduled primary profile | `get/set/del radioat`, `get/set/del tempradioat` |
+| Scheduled second profile | `get/set/del radioat2`, `get/set/del tempradioat2` |
+| UTC clock | `clock`, `clock sync`, `time <epoch-seconds>` |
 | Profile diagnostics | `get radio2.status`, `get radio2.timing` |
 | Crossover | `get radio2.cross`, `set radio2.cross <auto|on|off>` |
 | Filter rules | `get/set/del flood.filter[.n]`, `flood.rule[.n]` |
@@ -141,14 +144,44 @@ or normal terminal for longer lists.
 | Group-data forwarding | `get/set flood.channel.data`, `flood.channel.data.hops` |
 | Flood hop limits | `get/set flood.max`, `flood.max.unscoped`, `flood.max.advert` |
 
-Each role still checks whether it supports the requested setting. Primary radio
-changes, identity export, passwords, ACL edits, filesystem operations, firmware
+Each role still checks whether it supports the requested setting. Immediate
+primary radio changes, identity export, passwords, ACL edits, filesystem operations, firmware
 installation, reboot, and unrestricted CLI are outside fleet capability.
 
-Secondary-radio mutations keep the old profile until their exact acknowledgement
+## Schedule radio changes and set clocks
+
+Use the normal schedule syntax, with UTC epoch seconds or `+N` whole minutes.
+Primary schedules are supported by repeaters; a role without a primary scheduler
+replies that the command is unsupported. Second-profile schedules use `rx` or
+`rxtx` after the coding rate:
+
+```text
+fleet send 3 all clock
+fleet send 3 all clock sync
+fleet send 3 all set radioat 910.5,500,8,5,+5
+fleet send 3 all set tempradioat 910.5,500,8,5,+5,+15
+fleet send 3 all set radioat2 910.5,500,8,5,rxtx,+5
+fleet send 3 all set tempradioat2 910.5,500,8,5,rxtx,+5,+15
+fleet send 3 1257aee5 get tempradioat2 1
+fleet send 3 1257aee5 del tempradioat2 all
+```
+
+Both relative endpoints use the receiver's clock at command acceptance. For a
+coordinated fleet switch, use the same absolute UTC start/end timestamps on all
+nodes. An optional final preamble value or `auto` uses the existing radio syntax.
+Schedule capacity, radio-range validation, and overlap checks remain unchanged;
+pending schedules still do not survive reboot.
+
+`clock sync` advances a receiver to the publishing Companion's signed timestamp
+plus one second. `time <epoch-seconds>` sets an explicit UTC timestamp. Both use
+the existing forward-only clock rule. Initial setup or a clock far out of sync
+still requires local or individually authenticated clock correction: fleet clock
+commands retain the signature, expiry, replay, and maximum-clock-lead checks.
+
+Radio mutations and schedule edits keep the old state until their exact acknowledgement
 packets drain, with at least one successful physical transmission. Queue failure,
 failed transmissions, or an acknowledgement timeout cancels the staged mutation.
-Temporary-radio leases retain their original monotonic end time; waiting for an
+Temporary-radio leases and scheduled windows retain their original monotonic end time; waiting for an
 acknowledgement does not lengthen them.
 
 Exact single-node acknowledgements wait 0.5–1.5 seconds; prefix, multiple-target,
@@ -165,8 +198,9 @@ can still process an authenticated command when a filter or `repeat off` blocks
 forwarding. **Intermediate nodes can still block delivery to downstream nodes.**
 Management traffic has no automatic exemption from forwarding filters. Preserve
 an appropriate group-data/channel forwarding rule and an existing direct-admin
-recovery path before changing network-wide filters. The primary radio is retained
-by this capability, but a forwarding policy can still partition the fleet.
+recovery path before changing network-wide filters. Primary-radio schedules can
+move nodes off the normal network; use a temporary window or a planned recovery
+path for experiments. A forwarding policy can also partition the fleet.
 
 The receiver is allocated lazily when fleet settings are present or queried, with
 one bounded mailbox. Cryptographic work and storage writes occur after the receive

examples/simple_repeater/MyMesh.cpp

@@ -4641,6 +4641,11 @@ void MyMesh::formatScheduledRadioSetting(char* reply, int setting_idx, int displ
 
 void MyMesh::addScheduledRadioParams(bool temporary, float freq, float bw, uint8_t sf, uint8_t cr,
                                      uint32_t start_time, uint32_t end_time, char* reply, uint16_t preamble) {
+#if MESH_ENABLE_FLEET_CONTROL
+  if (fleet_schedule_pending_) {
+    strcpy(reply, "Error: fleet schedule acknowledgement pending"); return;
+  }
+#endif
   uint32_t now = getRTCClock()->getCurrentTime();
   if (!isValidScheduledRadioParams(freq, bw, sf, cr)) {
     strcpy(reply, "Error, invalid radio params");
@@ -4673,19 +4678,23 @@ void MyMesh::addScheduledRadioParams(bool temporary, float freq, float bw, uint8
     return;
   }
 
-  scheduled_radio_settings[slot].active = true;
-  scheduled_radio_settings[slot].temporary = temporary;
-  scheduled_radio_settings[slot].started = false;
-  scheduled_radio_settings[slot].freq = freq;
-  scheduled_radio_settings[slot].bw = bw;
-  scheduled_radio_settings[slot].sf = sf;
-  scheduled_radio_settings[slot].cr = cr;
-  scheduled_radio_settings[slot].preamble = preamble;
-  scheduled_radio_settings[slot].start_time = start_time;
-  scheduled_radio_settings[slot].end_time = temporary ? end_time : 0;
+  ScheduledRadioSetting* accepted = &scheduled_radio_settings[slot];
+#if MESH_ENABLE_FLEET_CONTROL
+  if (fleet_command_) accepted = &fleet_schedule_setting_;
+#endif
+  accepted->active = true;
+  accepted->temporary = temporary;
+  accepted->started = false;
+  accepted->freq = freq;
+  accepted->bw = bw;
+  accepted->sf = sf;
+  accepted->cr = cr;
+  accepted->preamble = preamble;
+  accepted->start_time = start_time;
+  accepted->end_time = temporary ? end_time : 0;
   const uint64_t current_uptime_millis =
       uptime_millis + (uint32_t)(millis() - last_millis);
-  scheduled_radio_settings[slot].hard_end_uptime_millis = temporary
+  accepted->hard_end_uptime_millis = temporary
       ? mesh::TempRadioLeaseDeadline::fromEpochEnd(
             current_uptime_millis, now, end_time)
       : 0;
@@ -4693,13 +4702,32 @@ void MyMesh::addScheduledRadioParams(bool temporary, float freq, float bw, uint8
   // apply failure, especially when its deadline is sooner than that retry.
   scheduled_radio_retry_at = 0;
   scheduled_radio_retry_failures = 0;
+#if MESH_ENABLE_FLEET_CONTROL
+  if (fleet_command_) {
+    fleet_schedule_slot_ = slot;
+    fleet_schedule_pending_ = true;
+    fleet_schedule_delivered_ = false;
+    fleet_schedule_delete_all_temp_ = false;
+    memset(fleet_schedule_delete_mask_, 0, sizeof(fleet_schedule_delete_mask_));
+  }
+#endif
   refreshScheduledRadioState();
 
   char delay[16];
   formatScheduledRadioDuration(delay, sizeof(delay), start_time);
+  int accepted_index = getScheduledRadioSettingIndex(temporary, slot);
+  if (accepted != &scheduled_radio_settings[slot]) {
+    accepted_index = 1;
+    for (int i = 0; i < MAX_SCHEDULED_RADIO_SETTINGS; ++i) {
+      const auto& previous = scheduled_radio_settings[i];
+      if (previous.active && previous.temporary == temporary
+          && (previous.start_time < start_time || (previous.start_time == start_time && i < slot)))
+        ++accepted_index;
+    }
+  }
   snprintf(reply, 160, "OK - %s %d in %s",
            temporary ? "tempradioat" : "radioat",
-           getScheduledRadioSettingIndex(temporary, slot),
+           accepted_index,
            delay);
   if (temporary) {
     char duration[64];
@@ -4765,6 +4793,34 @@ void MyMesh::formatScheduledRadioParams(bool temporary, const char* selector, ch
 }
 
 void MyMesh::deleteScheduledRadioParams(bool temporary, const char* selector, char* reply) {
+#if MESH_ENABLE_FLEET_CONTROL
+  if (fleet_schedule_pending_) {
+    strcpy(reply, "Error: fleet schedule acknowledgement pending"); return;
+  }
+  if (fleet_command_) {
+    const bool all = selectorIsEmpty(selector) || selectorIsAll(selector);
+    int idx = -1;
+    if (!all) {
+      int wanted = 0;
+      if (!parsePositiveSelector(selector, wanted)) {
+        strcpy(reply, temporary ? "Error, use: del tempradioat [n]" : "Error, use: del radioat [n]"); return;
+      }
+      idx = findScheduledRadioSettingByIndex(temporary, wanted);
+      if (idx < 0) { strcpy(reply, "Error: not found"); return; }
+    }
+    memset(fleet_schedule_delete_mask_, 0, sizeof(fleet_schedule_delete_mask_));
+    for (int i = 0; i < MAX_SCHEDULED_RADIO_SETTINGS; ++i) {
+      if (scheduled_radio_settings[i].active && scheduled_radio_settings[i].temporary == temporary
+          && (all || idx == i)) fleet_schedule_delete_mask_[i / 8] |= uint8_t(1U << (i % 8));
+    }
+    fleet_schedule_slot_ = -1;
+    fleet_schedule_delete_all_temp_ = all && temporary;
+    fleet_schedule_pending_ = true;
+    fleet_schedule_delivered_ = false;
+    strcpy(reply, "OK - schedule deletion after reply");
+    return;
+  }
+#endif
   if (selectorIsEmpty(selector) || selectorIsAll(selector)) {
     int deleted = 0;
     bool restore_radio = false;
@@ -4805,7 +4861,52 @@ void MyMesh::deleteScheduledRadioParams(bool temporary, const char* selector, ch
   strcpy(reply, "OK");
 }
 
+#if MESH_ENABLE_FLEET_CONTROL
+void MyMesh::finishFleetReplyMutation(bool delivered) {
+  if (!fleet_schedule_pending_) return;
+  fleet_schedule_delivered_ = delivered;
+  if (!delivered) {
+    fleet_schedule_pending_ = false;
+    fleet_schedule_setting_ = {};
+    memset(fleet_schedule_delete_mask_, 0, sizeof(fleet_schedule_delete_mask_));
+  }
+}
+
+void MyMesh::serviceFleetScheduleReply() {
+  if (!fleet_schedule_pending_ || !fleet_schedule_delivered_) return;
+  // Dispatcher completion hooks only mark delivery. All schedule/radio work
+  // runs here, with the original UTC endpoints and monotonic lease bound.
+  fleet_schedule_pending_ = fleet_schedule_delivered_ = false;
+  if (fleet_schedule_slot_ >= 0) {
+    const uint64_t uptime = uptime_millis + (uint32_t)(millis() - last_millis);
+    const bool expired = fleet_schedule_setting_.temporary
+        && (getRTCClock()->getCurrentTime() >= fleet_schedule_setting_.end_time
+            || mesh::TempRadioLeaseDeadline::expired(uptime, fleet_schedule_setting_.hard_end_uptime_millis));
+    // A local recovery may have reused the slot while this reply was pending.
+    if (!expired && !scheduled_radio_settings[fleet_schedule_slot_].active)
+      scheduled_radio_settings[fleet_schedule_slot_] = fleet_schedule_setting_;
+  } else {
+    for (int i = 0; i < MAX_SCHEDULED_RADIO_SETTINGS; ++i) {
+      if (fleet_schedule_delete_mask_[i / 8] & uint8_t(1U << (i % 8)))
+        clearScheduledRadioSetting(i, true);
+    }
+    if (fleet_schedule_delete_all_temp_ && temp_radio_handoff_pending) {
+      temp_radio_handoff_pending = false;
+      queueSavedRadioApply();
+    }
+  }
+  fleet_schedule_setting_ = {};
+  memset(fleet_schedule_delete_mask_, 0, sizeof(fleet_schedule_delete_mask_));
+  scheduled_radio_retry_at = 0;
+  scheduled_radio_retry_failures = 0;
+  refreshScheduledRadioState();
+}
+#endif
+
 void MyMesh::processScheduledRadioSettings() {
+#if MESH_ENABLE_FLEET_CONTROL
+  serviceFleetScheduleReply();
+#endif
   const uint64_t current_uptime_millis =
       uptime_millis + (uint32_t)(millis() - last_millis);
   bool hard_temp_end_due = false;
@@ -4819,7 +4920,11 @@ void MyMesh::processScheduledRadioSettings() {
     }
   }
 
-  if (temp_radio_reply_barrier.waiting() && !hard_temp_end_due) {
+  if ((temp_radio_reply_barrier.waiting()
+#if MESH_ENABLE_FLEET_CONTROL
+       || fleet_schedule_pending_
+#endif
+      ) && !hard_temp_end_due) {
     // A parameterized-TempRadio reply copy is queued or on air. Unlike a
     // fixed RTC delay, this remains correct under CAD, duty throttling, and
     // unrelated queue pressure.  TX completion/failure releases the barrier.
@@ -5068,6 +5173,10 @@ uint32_t MyMesh::getPowerSaveSleepSeconds(uint32_t max_secs) const {
 }
 
 void MyMesh::applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins, uint16_t preamble) {
+#if MESH_ENABLE_FLEET_CONTROL
+  // Direct administration supersedes an earlier unconfirmed fleet intent.
+  finishFleetReplyMutation(false);
+#endif
   ++primary_radio_mutation_generation;
   primary_radio_mutation_starts_temp = true;
   if (_cli.radioProfiles().hasReplyMutation()) {
@@ -5127,6 +5236,9 @@ void MyMesh::applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr,
 }
 
 bool MyMesh::scheduleNormalRadio() {
+#if MESH_ENABLE_FLEET_CONTROL
+  finishFleetReplyMutation(false);
+#endif
   ++primary_radio_mutation_generation;
   if (_cli.radioProfiles().hasReplyMutation()) {
     _cli.radioProfiles().finishReplyMutation(false);

examples/simple_repeater/MyMesh.h

@@ -584,6 +584,15 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
   unsigned long scheduled_radio_save_retry_at;
   uint8_t scheduled_radio_retry_failures;
   ScheduledRadioSetting scheduled_radio_settings[MAX_SCHEDULED_RADIO_SETTINGS];
+#if MESH_ENABLE_FLEET_CONTROL
+  // One verified fleet schedule waits for its own physical acknowledgement.
+  ScheduledRadioSetting fleet_schedule_setting_{};
+  uint8_t fleet_schedule_delete_mask_[(MAX_SCHEDULED_RADIO_SETTINGS + 7) / 8]{};
+  int fleet_schedule_slot_ = -1;
+  bool fleet_command_ = false, fleet_schedule_pending_ = false;
+  bool fleet_schedule_delivered_ = false, fleet_schedule_delete_all_temp_ = false;
+  void serviceFleetScheduleReply();
+#endif
   uint8_t _ota_update_channel = 0;  // Snapshot: later ota branch commands cannot retarget a queued update
   int  matching_peer_indexes[MAX_CLIENTS];
 #if defined(WITH_MQTT_BRIDGE)
@@ -1093,6 +1102,12 @@ public:
 
   // CommonCLICallbacks
   bool supportsFleetControl() const override { return true; }
+#if MESH_ENABLE_FLEET_CONTROL
+  void beginFleetCommand() override { fleet_command_ = true; }
+  void endFleetCommand() override { fleet_command_ = false; }
+  bool hasFleetReplyMutation() const override { return fleet_schedule_pending_; }
+  void finishFleetReplyMutation(bool delivered) override;
+#endif
   mesh::Radio* getProfileRadio() override { return _radio; }
   void applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins, uint16_t preamble = 0) override;
   bool scheduleNormalRadio() override;

src/helpers/CommonCLI.cpp

@@ -2916,7 +2916,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, char* command, char* re
       strcpy(reply, "OK - Advert sent");
     } else if (memcmp(command, "clock sync", 10) == 0) {
       uint32_t curr = getRTCClock()->getCurrentTime();
-      if (sender_timestamp > curr) {
+      if (sender_timestamp != UINT32_MAX && sender_timestamp > curr) {
         getRTCClock()->setCurrentTime(sender_timestamp + 1);
         _callbacks->onManualClockSet();
         uint32_t now = getRTCClock()->getCurrentTime();

src/helpers/CommonCLI.h

@@ -480,7 +480,11 @@ struct LegacyObserverTail {
 };
 #endif
 
-class CommonCLICallbacks {
+class CommonCLICallbacks
+#if MESH_ENABLE_FLEET_CONTROL
+  : public mesh::FleetReplyHooks
+#endif
+{
 public:
   virtual bool supportsFleetControl() const { return false; }
   // True only after an actual NTP response this boot, not a plausible RTC.

src/helpers/CommonCLI_Management.cpp

@@ -248,7 +248,7 @@ void CommonCLI::beginManagement(mesh::Mesh& mesh, FILESYSTEM* fs) {
 #if MESH_ENABLE_FLEET_CONTROL
   if (_callbacks->supportsFleetControl()
       && (fs->exists("/fleet_channel") || fs->exists("/fleet_channel.bak"))) {
-    _fleet_channel = new (std::nothrow) mesh::FleetChannel(fs);
+    _fleet_channel = new (std::nothrow) mesh::FleetChannel(fs, _callbacks);
   }
 #endif
   if (!_data_route) {
@@ -279,7 +279,7 @@ bool CommonCLI::handleFleetCommand(const char* command, char* reply) {
   if (!_callbacks->supportsFleetControl()) {
     strcpy(reply, "Err - fleet control is available on infrastructure roles"); return true;
   }
-  if (!_fleet_channel) _fleet_channel = new (std::nothrow) mesh::FleetChannel(_management_fs);
+  if (!_fleet_channel) _fleet_channel = new (std::nothrow) mesh::FleetChannel(_management_fs, _callbacks);
   if (!_fleet_channel) {
     strcpy(reply, "Err - fleet control memory unavailable"); return true;
   }

src/helpers/FleetChannel.cpp

@@ -215,7 +215,11 @@ bool FleetChannel::acknowledge(Mesh& mesh, const char* name, uint32_t sequence,
 
 void FleetChannel::finish(RadioProfileCLI& profiles, bool delivered) {
   barrier_.clear(); ack_deadline_ = 0;
-  profiles.finishReplyMutation(delivered);
+  if (ack_profile_mutation_
+      && profiles.replyMutationGeneration() == ack_profile_generation_)
+    profiles.finishReplyMutation(delivered);
+  ack_profile_mutation_ = false;
+  if (hooks_) hooks_->finishFleetReplyMutation(delivered);
 }
 void FleetChannel::complete(Packet* packet, RadioProfileCLI& profiles) {
   if (barrier_.complete(packet) && !barrier_.waiting()) finish(profiles, true);

src/helpers/FleetChannel.h

@@ -9,6 +9,15 @@
 
 namespace mesh {
 
+// Role-owned mutations must complete after the exact fleet reply, just like
+// shared secondary-profile changes. Completion hooks stage work for loop().
+struct FleetReplyHooks {
+  virtual void beginFleetCommand() {}
+  virtual void endFleetCommand() {}
+  virtual bool hasFleetReplyMutation() const { return false; }
+  virtual void finishFleetReplyMutation(bool) {}
+};
+
 // Scope codes authenticate the original packet's payload. A new reply needs
 // fresh codes from the exact matching region key, including rotated keys.
 template <typename Regions, typename Region>
@@ -31,16 +40,19 @@ bool captureFleetReplyScope(Regions& regions, const Region* region,
 // the deliberately narrower fleet capability.
 class FleetChannel {
   FILESYSTEM* fs_;
+  FleetReplyHooks* hooks_;
   GroupChannel channel_{};
   Identity controller_;
   uint32_t last_sequence_ = 0;
   uint32_t ack_deadline_ = 0;
+  uint32_t ack_profile_generation_ = 0;
   uint32_t verify_window_ = 0;
   uint32_t accepted_ = 0, rejected_ = 0, busy_ = 0;
   uint8_t verify_attempts_ = 0;
   bool configured_ = false, controller_set_ = false, healthy_ = false;
   bool pending_ = false;
   bool broadcast_ = false;
+  bool ack_profile_mutation_ = false;
   uint8_t incoming_[MAX_PACKET_PAYLOAD]{};
   uint8_t incoming_len_ = 0, rx_profile_ = 0, path_hash_size_ = 1;
   uint32_t rx_generation_ = 0;
@@ -58,7 +70,8 @@ class FleetChannel {
   void serviceDeadline(Mesh& mesh, RadioProfileCLI& profiles);
 
  public:
-  explicit FleetChannel(FILESYSTEM* fs) : fs_(fs) { healthy_ = load(); }
+  explicit FleetChannel(FILESYSTEM* fs, FleetReplyHooks* hooks = nullptr)
+      : fs_(fs), hooks_(hooks) { healthy_ = load(); }
   bool handleConfig(const char* command, char* reply, size_t capacity);
   void receive(Packet* packet, Mesh& mesh, const TransportKey* scope = nullptr);
   void complete(Packet* packet, RadioProfileCLI& profiles);
@@ -72,7 +85,8 @@ class FleetChannel {
     serviceDeadline(mesh, profiles);
     if (!pending_) return;
     pending_ = false;
-    if (barrier_.waiting() || profiles.hasReplyMutation()) { ++busy_; return; }
+    if (barrier_.waiting() || profiles.hasReplyMutation()
+        || (hooks_ && hooks_->hasFleetReplyMutation())) { ++busy_; return; }
     FleetCommand::Decoded command;
     if (!decode(mesh, command)) { ++rejected_; return; }
     // Reserve before dispatch, including failed commands. A lost reply, reboot,
@@ -87,10 +101,18 @@ class FleetChannel {
     Dispatcher::ReceiveProfileScope receive_scope(mesh, rx_profile_, rx_generation_);
     char reply[160] = {};
     const uint32_t generation = profiles.replyMutationGeneration();
+    if (hooks_) hooks_->beginFleetCommand();
     profiles.beginReplyCommand();
     handle(command.sequence, command.command, reply);
     profiles.endReplyCommand();
-    const bool changed = profiles.replyMutationGeneration() != generation;
+    if (hooks_) hooks_->endFleetCommand();
+    // Only this acknowledgement may release the secondary mutation it staged.
+    // Direct recovery can cancel it and accept a newer independent command.
+    ack_profile_generation_ = profiles.replyMutationGeneration();
+    ack_profile_mutation_ = profiles.hasReplyMutation()
+        && ack_profile_generation_ != generation;
+    const bool changed = ack_profile_mutation_
+        || (hooks_ && hooks_->hasFleetReplyMutation());
     if (!acknowledge(mesh, name, command.sequence, reply, changed) && changed) {
       finish(profiles, false);
     }

src/helpers/FleetCommand.cpp

@@ -136,6 +136,69 @@ bool integerToken(const char* text, size_t length) {
   return true;
 }
 
+bool boundedIntegerToken(const char* text, size_t length, uint32_t maximum,
+                         uint32_t& value) {
+  if (!length) return false;
+  value = 0;
+  for (size_t i = 0; i < length; ++i) {
+    if (text[i] < '0' || text[i] > '9') return false;
+    const uint32_t digit = uint32_t(text[i] - '0');
+    if (digit > maximum || value > (maximum - digit) / 10U) return false;
+    value = value * 10U + digit;
+  }
+  return true;
+}
+
+bool scheduleTimeToken(const char* text, size_t length) {
+  if (!length) return false;
+  const bool relative = text[0] == '+';
+  uint32_t value;
+  // The scheduler resolves +N whole minutes against its current RTC and
+  // performs future/overflow/endpoint-order checks using that one snapshot.
+  return boundedIntegerToken(text + relative, length - relative,
+                             relative ? UINT32_MAX / 60U : UINT32_MAX, value)
+      && value != 0;
+}
+
+bool preambleToken(const char* text, size_t length) {
+  if (keyEquals(text, length, "auto")) return true;
+  uint32_t value;
+  return boundedIntegerToken(text, length, 65528U, value)
+      && (value == 0 || value >= 8);
+}
+
+bool scheduleSelector(const char* arguments, bool secondary) {
+  if (!*arguments || !strcmp(arguments, "all")) return true;
+  if (*arguments < '1' || *arguments > '9') return false;
+  uint32_t value;
+  // Primary capacity is role/build configurable. Match the bounded row
+  // authorization policy and let the role reject nonexistent slots.
+  return boundedIntegerToken(arguments, strlen(arguments), secondary ? 4U : 255U, value);
+}
+
+bool scheduledRadioTuple(const char* arguments, bool temporary, bool secondary) {
+  unsigned fields = 0;
+  const unsigned time_index = secondary ? 5 : 4;
+  const unsigned expected = time_index + (temporary ? 2 : 1);
+  const char* cursor = arguments;
+  for (;;) {
+    const char* end = strchr(cursor, ',');
+    const size_t length = end ? size_t(end - cursor) : strlen(cursor);
+    bool valid = false;
+    if (fields < 2) valid = decimalToken(cursor, length);
+    else if (fields < 4) {
+      uint32_t value;
+      valid = boundedIntegerToken(cursor, length, 255U, value);
+    } else if (secondary && fields == 4) {
+      valid = keyEquals(cursor, length, "rx") || keyEquals(cursor, length, "rxtx");
+    } else if (fields < expected) valid = scheduleTimeToken(cursor, length);
+    else if (fields == expected) valid = preambleToken(cursor, length);
+    if (!valid || ++fields > expected + 1) return false;
+    if (!end) return fields == expected || fields == expected + 1;
+    cursor = end + 1;
+  }
+}
+
 bool radioTuple(const char* arguments, bool temporary) {
   if (!strcmp(arguments, "off")) return true;
   unsigned fields = 0;
@@ -231,6 +294,12 @@ bool FleetCommand::commandAllowed(const char* command) {
         || byte == '\\') return false;
   }
   if (!length || command[0] == ' ' || command[length - 1] == ' ') return false;
+  if (!strcmp(command, "clock") || !strcmp(command, "clock sync")) return true;
+  if (!strncmp(command, "time ", 5)) {
+    uint32_t value;
+    return boundedIntegerToken(command + 5, length - 5, UINT32_MAX, value)
+        && value >= MinEpoch;
+  }
   bool get = false, set = false, del = false;
   if (!strncmp(command, "get ", 4)) get = true;
   else if (!strncmp(command, "set ", 4)) set = true;
@@ -242,6 +311,14 @@ bool FleetCommand::commandAllowed(const char* command) {
   const char* arguments = separator ? separator + 1 : "";
   if (!key_length || (separator && (!*arguments || *arguments == ' '))) return false;
 
+  const bool scheduled_primary = keyEquals(key, key_length, "radioat")
+      || keyEquals(key, key_length, "tempradioat");
+  const bool scheduled_secondary = keyEquals(key, key_length, "radioat2")
+      || keyEquals(key, key_length, "tempradioat2");
+  if (scheduled_primary || scheduled_secondary) {
+    return set ? scheduledRadioTuple(arguments, key[0] == 't', scheduled_secondary)
+        : (get || del) && scheduleSelector(arguments, scheduled_secondary);
+  }
   if (keyEquals(key, key_length, "radio2")
       || keyEquals(key, key_length, "tempradio2")) {
     return get ? !*arguments : set && radioTuple(arguments, key_length == 10);

src/helpers/FleetCommand.h

@@ -3,15 +3,7 @@
 #include <stddef.h>
 #include <stdint.h>
 #include "../Identity.h"
-
-#ifndef MESH_ENABLE_FLEET_CONTROL
-  // Fixed 240 KiB STM32 images have no room for fleet control.
-  #if defined(STM32_PLATFORM)
-    #define MESH_ENABLE_FLEET_CONTROL 0
-  #else
-    #define MESH_ENABLE_FLEET_CONTROL 1
-  #endif
-#endif
+#include "FleetControlConfig.h"
 
 namespace mesh {
 

src/helpers/FleetControlConfig.h

@@ -0,0 +1,10 @@
+#pragma once
+
+#ifndef MESH_ENABLE_FLEET_CONTROL
+  // Fixed 240 KiB STM32 images have no room for fleet control.
+  #if defined(STM32_PLATFORM)
+    #define MESH_ENABLE_FLEET_CONTROL 0
+  #else
+    #define MESH_ENABLE_FLEET_CONTROL 1
+  #endif
+#endif

src/helpers/RadioProfileCLI.cpp

@@ -287,6 +287,20 @@ bool RadioProfileCLI::finishReplyMutation(bool delivered) {
       || !fs_ || !fs_->exists(ReplyPath) || fs_->remove(ReplyPath);
 }
 
+#if MESH_ENABLE_FLEET_CONTROL
+bool RadioProfileCLI::stageRemoteSchedule(unsigned slot, const Schedule& schedule) {
+  if (slot >= 4 || !stageRemoteMutation(schedule.temporary
+          ? RemoteMutation::AddTemporarySchedule : RemoteMutation::AddSavedSchedule,
+          schedule.config, schedule.end, uint8_t(1U << slot))) return false;
+  // Reuse the pending mutation's two time fields for the accepted UTC start
+  // and end. Keep the separate monotonic end anchored at command admission;
+  // neither a late reply nor a wall-clock correction may extend the lease.
+  remote_start_ms_ = schedule.start;
+  remote_schedule_end_ms_ = schedule.end_ms;
+  return true;
+}
+#endif
+
 bool RadioProfileCLI::applyReplyMutation() {
   const auto kind = remote_mutation_;
   if ((kind == RemoteMutation::Saved || kind == RemoteMutation::Off)
@@ -302,11 +316,37 @@ bool RadioProfileCLI::applyReplyMutation() {
   } else if (kind == RemoteMutation::Saved) {
     saved_ = remote_config_;
     publish_pending_ = true; publish_after_ms_ = millis();
+#if MESH_ENABLE_FLEET_CONTROL
+  } else if (kind == RemoteMutation::AddSavedSchedule
+             || kind == RemoteMutation::AddTemporarySchedule) {
+    const bool temporary = kind == RemoteMutation::AddTemporarySchedule;
+    const uint32_t now = millis(), epoch = rtc_ ? rtc_->getCurrentTime() : 0;
+    if (temporary && (!remainingMillis(remote_schedule_end_ms_, now)
+                      || epoch >= remote_duration_ms_)) return true;
+    for (unsigned i = 0; i < 4; ++i) if (remote_delete_mask_ & (1U << i)) {
+      Schedule& scheduled = schedules_[(temporary ? 4 : 0) + i];
+      scheduled = {};
+      scheduled.config = remote_config_;
+      scheduled.start = remote_start_ms_;
+      scheduled.end = remote_duration_ms_;
+      scheduled.end_ms = remote_schedule_end_ms_;
+      scheduled.temporary = temporary;
+      scheduled.active = true;
+      break;
+    }
+#endif
   } else if (kind == RemoteMutation::DeleteTemp) {
     for (unsigned i = 0; i < 4; ++i) if (remote_delete_mask_ & (1U << i)) {
       if (schedules_[4+i].started) { temp_active_ = false; temp_remaining_ms_ = 0; }
       schedules_[4+i] = {};
     }
+#if MESH_ENABLE_FLEET_CONTROL
+  } else if (kind == RemoteMutation::DeleteSavedSchedule) {
+    for (unsigned i = 0; i < 4; ++i) if (remote_delete_mask_ & (1U << i)) {
+      schedules_[i] = {};
+    }
+    schedule_retry_ms_ = 0;
+#endif
   } else {
     if (kind == RemoteMutation::Off) { saved_ = {}; schedule_retry_ms_ = 0; }
     temp_pending_ = temp_active_ = false; temp_remaining_ms_ = 0;
@@ -633,8 +673,14 @@ bool RadioProfileCLI::handle(const char* command, char* reply, size_t capacity,
     }
     const unsigned first = scheduled_temp ? 4 : 0;
     if (verb == Delete) {
-      if (remote_command_ && scheduled_temp) {
+      if (remote_command_ && (scheduled_temp || MESH_ENABLE_FLEET_CONTROL)) {
+#if MESH_ENABLE_FLEET_CONTROL
+        stageRemoteMutation(scheduled_temp ? RemoteMutation::DeleteTemp
+                                          : RemoteMutation::DeleteSavedSchedule,
+                            {}, 0, index ? 1U << (index - 1) : 15);
+#else
         stageRemoteMutation(RemoteMutation::DeleteTemp, {}, 0, index ? 1U << (index - 1) : 15);
+#endif
         snprintf(reply, capacity, "OK - schedule cleared after reply"); return true;
       }
       if (!scheduled_temp) schedule_retry_ms_ = 0;
@@ -709,9 +755,20 @@ bool RadioProfileCLI::handle(const char* command, char* reply, size_t capacity,
     }
     const unsigned first = scheduled_temp ? 4 : 0;
     for (unsigned i = 0; i < 4; ++i) if (!schedules_[first+i].active) {
+#if MESH_ENABLE_FLEET_CONTROL
+      Schedule s;
+#else
       auto& s = schedules_[first+i]; s = {};
+#endif
       s.config = config; s.start = start; s.end = end; s.temporary = scheduled_temp;
       s.end_ms = scheduled_temp ? millis() + (end-epoch)*1000UL : 0; s.active = true;
+#if MESH_ENABLE_FLEET_CONTROL
+      if (remote_command_) {
+        if (!stageRemoteSchedule(i, s)) {
+          snprintf(reply, capacity, "Error: radio acknowledgement pending"); return true;
+        }
+      } else schedules_[first+i] = s;
+#endif
       snprintf(reply, capacity, "OK - %s %u queued, preamble=%u", key, i+1, preview.preamble(1, 32));
       appendChirpWarning(reply, capacity, preview); return true;
     }

src/helpers/RadioProfileCLI.h

@@ -3,6 +3,7 @@
 #include <Dispatcher.h>
 #include <helpers/IdentityStore.h>
 #include <helpers/CLICommandUtils.h>
+#include <helpers/FleetControlConfig.h>
 
 namespace mesh {
 
@@ -41,16 +42,27 @@ class RadioProfileCLI {
   bool recoverable_corrupt_store_ = false;
   bool publish_pending_ = false;
   enum class ImageReadResult : uint8_t { Missing, Valid, Invalid, Unreadable };
-  enum class RemoteMutation : uint8_t { None, Saved, Temporary, Off, TempOff, DeleteTemp };
+  enum class RemoteMutation : uint8_t {
+    None, Saved, Temporary, Off, TempOff, DeleteTemp
+#if MESH_ENABLE_FLEET_CONTROL
+    , DeleteSavedSchedule, AddSavedSchedule, AddTemporarySchedule
+#endif
+  };
   RemoteMutation remote_mutation_ = RemoteMutation::None;
   RadioProfileConfig remote_config_;
   uint32_t remote_duration_ms_ = 0, remote_start_ms_ = 0;
   uint32_t remote_commit_retry_ms_ = 0;
+#if MESH_ENABLE_FLEET_CONTROL
+  uint32_t remote_schedule_end_ms_ = 0;
+#endif
   uint32_t remote_generation_ = 0;
   uint8_t remote_delete_mask_ = 0;
   bool remote_command_ = false, remote_delivered_ = false;
   bool stageRemoteMutation(RemoteMutation kind, const RadioProfileConfig& config = {},
                            uint32_t duration_ms = 0, uint8_t delete_mask = 0);
+#if MESH_ENABLE_FLEET_CONTROL
+  bool stageRemoteSchedule(unsigned slot, const Schedule& schedule);
+#endif
   bool save(const RadioProfileConfig& config, uint16_t preamble, RadioCrossMode cross);
   bool prepareSavedImage(const char* path, const RadioProfileConfig& config,
                          uint16_t preamble, RadioCrossMode cross);

test/test_companion_fleet_control.py

@@ -272,6 +272,103 @@ int main(int argc, char** argv) {
     targets += ",44556677"; // More than the entire target-record capacity.
     rejected(node, ("fleet send 1 " + targets + " set radio2 off").c_str());
     assert(node.queued == 1);
+  } else if (scenario == "scheduled_sets") {
+    const char* relative[] = {
+      "set radioat 910.5,500,5,5,+1,auto",
+      "set tempradioat 910.5,500,5,5,+1,+2,32",
+      "set radioat2 910.5,500,5,5,rxtx,+1,auto",
+      "set tempradioat2 910.5,500,5,5,rx,+1,+2,32",
+    };
+    uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+    auto sendAndDecode = [&](const std::string& scheduled) {
+      const std::string command = "fleet send 1 all " + scheduled;
+      rejected(node, command.c_str(), 1);
+      const unsigned before = node.queued;
+      assert(node.handleCommand(command.c_str(), 0, reply));
+      assert(!strncmp(reply, "OK - fleet command queued", 25));
+      assert(node.queued == before + 1 && node.sent_policy == 2);
+      mesh::FleetCommand::Decoded decoded;
+      assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+          node.sent, node.sent_length, node.clock.now, recipient, decoded));
+      assert(!strcmp(decoded.command, scheduled.c_str()));
+      assert(decoded.sequence == node.clock.now);
+      ++node.clock.now;
+    };
+    for (const char* scheduled : relative) sendAndDecode(scheduled);
+    const std::string start = std::to_string(node.clock.now + 60);
+    const std::string end = std::to_string(node.clock.now + 120);
+    sendAndDecode("set radioat 910.5,500,5,5," + start);
+    sendAndDecode("set tempradioat 910.5,500,5,5," + start + "," + end);
+    sendAndDecode("set radioat2 910.5,500,5,5,rx," + start);
+    sendAndDecode("set tempradioat2 910.5,500,5,5,rxtx," + start + "," + end);
+    assert(node.queued == 8);
+  } else if (scenario == "schedule_get_delete") {
+    uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+    for (const char* scheduled : {
+        "get radioat", "get radioat 1", "get radioat 255", "get radioat all",
+        "get tempradioat", "get tempradioat 2", "get tempradioat all",
+        "get radioat2", "get radioat2 4", "get radioat2 all",
+        "get tempradioat2", "get tempradioat2 1", "get tempradioat2 all",
+        "del radioat", "del radioat 1", "del radioat all",
+        "del tempradioat", "del tempradioat 2", "del tempradioat all",
+        "del radioat2", "del radioat2 4", "del radioat2 all",
+        "del tempradioat2", "del tempradioat2 1", "del tempradioat2 all"}) {
+      const std::string command = "fleet send 1 42424242 " + std::string(scheduled);
+      rejected(node, command.c_str(), 100);
+      const unsigned before = node.queued;
+      assert(node.handleCommand(command.c_str(), 0, reply));
+      assert(node.queued == before + 1 && node.sent_policy == 2);
+      mesh::FleetCommand::Decoded decoded;
+      assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+          node.sent, node.sent_length, node.clock.now, recipient, decoded));
+      assert(!strcmp(decoded.command, scheduled));
+      ++node.clock.now;
+    }
+  } else if (scenario == "malformed_schedules") {
+    for (const char* scheduled : {
+        "set radioat 910.5,500,5,5", "set tempradioat 910.5,500,5,5,+1",
+        "set radioat2 910.5,500,5,5,rx", "set tempradioat2 910.5,500,5,5,rxtx,+1",
+        "set radioat 910.5,500,5,5,+0", "set radioat 910.5,500,5,5,-1",
+        "set radioat 910.5,500,5,5,+1.5", "set radioat 910.5,500,5,5,4294967296",
+        "set radioat2 910.5,500,5,5,bogus,+1",
+        "set tempradioat2 910.5,500,5,5,rx,+1,+2,auto,extra",
+        "set tempradioat 910.5,500,5,5,+1,+2,7",
+        "set radioat 910.5,500,5,5,+1,65529",
+        "get radioat 256", "get radioat2 5", "get radioat 01", "get radioat 1x",
+        "del tempradioat2 0", "del radioat2 all extra", "radioat 910.5,500,5,5,+1"}) {
+      rejected(node, ("fleet send 1 all " + std::string(scheduled)).c_str());
+    }
+    assert(node.handleCommand("fleet send 1 all set radioat 910.5,500,5,5,+1", 0, reply));
+    assert(node.queued == 1);
+  } else if (scenario == "schedule_packet_budget") {
+    const std::string scheduled = "set tempradioat2 910.5,500,5,5,rxtx,+1,+2,auto";
+    std::string targets = fullKey(0x42) + "," + fullKey(0x51) + ","
+        + fullKey(0x63) + "," + fullKey(0x74);
+    mesh::FleetCommand::Targets parsed;
+    assert(mesh::FleetCommand::parseTargets(targets.c_str(), targets.size(), parsed));
+    assert(mesh::FleetCommand::commandAllowed(scheduled.c_str()));
+    assert(node.handleCommand(("fleet send 1 " + targets + " " + scheduled).c_str(), 0, reply));
+    assert(strstr(reply, "exceed one packet") && node.queued == 0);
+    assert(node.handleCommand(("fleet send 1 42424242 " + scheduled).c_str(), 0, reply));
+    assert(node.queued == 1 && node.sent_policy == 2);
+  } else if (scenario == "clock_controls") {
+    uint8_t recipient[32]; memset(recipient, 0x42, sizeof(recipient));
+    for (const char* clock_command : {"clock", "clock sync", "time 1800000000"}) {
+      const std::string command = "fleet send 1 all " + std::string(clock_command);
+      rejected(node, command.c_str(), 1);
+      const unsigned before = node.queued;
+      assert(node.handleCommand(command.c_str(), 0, reply));
+      assert(node.queued == before + 1 && node.sent_policy == 2);
+      mesh::FleetCommand::Decoded decoded;
+      assert(mesh::FleetCommand::decode(node.self_id, node.channels[1].channel.secret,
+          node.sent, node.sent_length, node.clock.now, recipient, decoded));
+      assert(!strcmp(decoded.command, clock_command));
+      ++node.clock.now;
+    }
+    for (const char* clock_command : {"get clock", "set clock", "clock synctime",
+        "clock sync extra", "time", "time -1", "time +1800000000", "time NaN",
+        "time 4294967296", "time 1735689599", "time 1800000000 extra"})
+      rejected(node, ("fleet send 1 all " + std::string(clock_command)).c_str());
   } else assert(false);
   printf("Companion fleet scenario %s passed\n", argv[1]);
 }
@@ -350,6 +447,21 @@ class CompanionFleetControlTests(unittest.TestCase):
     def test_valid_list_exceeding_packet_capacity_never_partially_sends(self):
         self.scenario("target_capacity")
 
+    def test_four_schedule_set_forms_are_signed_without_rewriting(self):
+        self.scenario("scheduled_sets")
+
+    def test_schedule_get_and_delete_forms_preserve_access_and_tx_route(self):
+        self.scenario("schedule_get_delete")
+
+    def test_malformed_and_truncated_schedules_never_queue_or_consume_sequence(self):
+        self.scenario("malformed_schedules")
+
+    def test_schedule_and_target_list_share_one_packet_budget(self):
+        self.scenario("schedule_packet_budget")
+
+    def test_exact_clock_controls_are_signed_and_remote_origin_is_denied(self):
+        self.scenario("clock_controls")
+
     def test_disabled_profiles_require_no_sender_peripherals_or_crypto_link(self):
         harness = r'''
 #include <cassert>

test/test_fleet_clock_control.py

@@ -0,0 +1,362 @@
+#!/usr/bin/env python3
+"""Deliver real signed fleet requests to the production CommonCLI clock branches.
+
+Receiver, codec and anti-replay storage run unchanged. Shared fleet-runtime
+fixtures provide hardware I/O and real OpenSSL Ed25519/AES/MAC interfaces. The
+CommonCLI setter and manual-clock hook ordering are extracted from production;
+only the RTC peripheral and reply date formatting are adapted for the host.
+"""
+
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from cpp_source import body
+from test_fleet_channel_runtime import FILESYSTEM, HARNESS as RUNTIME, MESH, PROFILE, UTILS
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+HARNESS = r'''
+#include <ctime>
+
+// Date formatting is a host peripheral adapter; it never mutates the RTC.
+class DateTime {
+  std::tm value_{};
+ public:
+  explicit DateTime(uint32_t seconds) {
+    const std::time_t epoch = seconds;
+    const std::tm* value = std::gmtime(&epoch);
+    assert(value); value_ = *value;
+  }
+  int hour() const { return value_.tm_hour; }
+  int minute() const { return value_.tm_min; }
+  int day() const { return value_.tm_mday; }
+  int month() const { return value_.tm_mon + 1; }
+  int year() const { return value_.tm_year + 1900; }
+};
+
+@ATOI@
+struct ManualClockCallbacks {
+  Clock& clock;
+  unsigned hooks = 0;
+  uint32_t observed = 0;
+  void onManualClockSet() {
+    assert(clock.sets == hooks + 1); // The production hook follows the setter.
+    ++hooks; observed = clock.getCurrentTime();
+  }
+};
+class CommonCLI {
+  Clock& rtc_;
+ public:
+  ManualClockCallbacks callbacks;
+  ManualClockCallbacks* _callbacks = &callbacks;
+  explicit CommonCLI(Clock& rtc) : rtc_(rtc), callbacks{rtc} {}
+  Clock* getRTCClock() { return &rtc_; }
+  void clockCommand(uint32_t sender_timestamp, const char* command, char* reply) {
+    @CLOCK_BRANCHES@
+    else assert(false && "fleet clock dispatch accepted an unexpected command");
+  }
+};
+
+struct Fixture {
+  fs::FS disk;
+  Mesh mesh;
+  RadioProfileCLI profiles;
+  FleetChannel fleet{&disk};
+  LocalIdentity publisher;
+  CommonCLI cli{mesh.clock};
+  unsigned dispatched = 0;
+  char reply[160] = {};
+  Fixture() { enroll(fleet, publisher); bind(fleet, mesh, profiles); }
+  unsigned offer(Packet& packet, FleetChannel* receiver = nullptr) {
+    fake_ms += 1000; // Admission checks, rather than rate throttling, decide each case.
+    const unsigned before = dispatched;
+    FleetChannel& destination = receiver ? *receiver : fleet;
+    destination.receive(&packet, mesh);
+    destination.service(mesh, profiles, "clock node",
+        [&](uint32_t sequence, const char* command_text, char* response) {
+          ++dispatched;
+          assert(storage::readLE32(disk.files.at(Path).data() + 56) == sequence);
+          assert(profiles.command);
+          cli.clockCommand(sequence, command_text, response);
+          strcpy(reply, response);
+        });
+    return dispatched - before;
+  }
+  void unchanged(Packet& packet) {
+    const uint32_t now = mesh.clock.now;
+    const unsigned setters = mesh.clock.sets, hooks = cli.callbacks.hooks;
+    const auto saved = disk.files.at(Path);
+    assert(offer(packet) == 0);
+    assert(mesh.clock.now == now && mesh.clock.sets == setters);
+    assert(cli.callbacks.hooks == hooks && disk.files.at(Path) == saved);
+  }
+};
+
+int main(int argc, char** argv) {
+  assert(argc == 2);
+  const std::string scenario = argv[1];
+  Fixture f;
+  const uint32_t initial = f.mesh.clock.now;
+  if (scenario == "clock_read") {
+    Packet packet = command(f.publisher, initial, "clock");
+    assert(f.offer(packet) == 1);
+    assert(strstr(f.reply, "UTC") && !strstr(f.reply, "clock set"));
+    assert(f.mesh.clock.now == initial && f.mesh.clock.sets == 0);
+    assert(f.cli.callbacks.hooks == 0);
+    assert(storage::readLE32(f.disk.files.at(Path).data() + 56) == initial);
+  } else if (scenario == "sync_forward") {
+    const uint32_t sender_time = initial + 5;
+    Packet packet = command(f.publisher, sender_time, "clock sync");
+    assert(f.offer(packet) == 1);
+    assert(f.mesh.clock.now == sender_time + 1);
+    assert(f.mesh.clock.sets == 1 && f.cli.callbacks.hooks == 1);
+    assert(f.cli.callbacks.observed == sender_time + 1);
+    assert(!strncmp(f.reply, "OK - clock set:", 15));
+    assert(storage::readLE32(f.disk.files.at(Path).data() + 56) == sender_time);
+  } else if (scenario == "sync_uint32_boundary") {
+    // Fleet envelopes require a later expiry and cannot carry this timestamp.
+    // Exercise the production CommonCLI branch directly to cover other CLI
+    // transports as well: adding one to UINT32_MAX must never reset the RTC.
+    const auto saved = f.disk.files.at(Path);
+    f.mesh.clock.now = UINT32_MAX - 1;
+    f.cli.clockCommand(UINT32_MAX, "clock sync", f.reply);
+    assert(f.mesh.clock.now == UINT32_MAX - 1);
+    assert(f.mesh.clock.sets == 0 && f.cli.callbacks.hooks == 0);
+    assert(strstr(f.reply, "ERR"));
+    assert(f.disk.files.at(Path) == saved && f.dispatched == 0);
+    // The adjacent representable result remains a valid forward update.
+    f.mesh.clock.now = UINT32_MAX - 2;
+    f.cli.clockCommand(UINT32_MAX - 1, "clock sync", f.reply);
+    assert(f.mesh.clock.now == UINT32_MAX);
+    assert(f.mesh.clock.sets == 1 && f.cli.callbacks.hooks == 1);
+    assert(f.cli.callbacks.observed == UINT32_MAX);
+    assert(!strncmp(f.reply, "OK - clock set:", 15));
+  } else if (scenario == "explicit_forward") {
+    const uint32_t requested = initial + 20;
+    const std::string text = "time " + std::to_string(requested);
+    Packet packet = command(f.publisher, initial, text.c_str());
+    assert(f.offer(packet) == 1);
+    assert(f.mesh.clock.now == requested);
+    assert(f.mesh.clock.sets == 1 && f.cli.callbacks.hooks == 1);
+    assert(f.cli.callbacks.observed == requested);
+    assert(!strncmp(f.reply, "OK - clock set:", 15));
+    // The signed request sequence, rather than the requested clock value,
+    // remains the durable anti-replay identity.
+    assert(storage::readLE32(f.disk.files.at(Path).data() + 56) == initial);
+  } else if (scenario == "backward_and_equal") {
+    Packet sync = command(f.publisher, initial - 1, "clock sync");
+    assert(f.offer(sync) == 1);
+    assert(strstr(f.reply, "clock cannot go backwards"));
+    Packet equal = command(f.publisher, initial, "clock sync");
+    assert(f.offer(equal) == 1);
+    assert(strstr(f.reply, "clock cannot go backwards"));
+    const std::string backward = "time " + std::to_string(initial - 1);
+    Packet earlier = command(f.publisher, initial + 1, backward.c_str());
+    assert(f.offer(earlier) == 1);
+    assert(strstr(f.reply, "clock cannot go backwards"));
+    const std::string same = "time " + std::to_string(initial);
+    Packet unchanged = command(f.publisher, initial + 2, same.c_str());
+    assert(f.offer(unchanged) == 1);
+    assert(strstr(f.reply, "clock cannot go backwards"));
+    assert(f.mesh.clock.now == initial && f.mesh.clock.sets == 0);
+    assert(f.cli.callbacks.hooks == 0);
+  } else if (scenario == "signatures_and_target") {
+    const std::string text = "time " + std::to_string(initial + 20);
+    Packet valid = command(f.publisher, initial, text.c_str());
+    uint8_t plain[184] = {};
+    const int size = Utils::MACThenDecrypt(channel_key, plain, valid.payload + 1,
+                                          valid.payload_len - 1);
+    assert(size > 3 && plain[2] > FleetCommand::SignatureSize);
+    plain[3 + plain[2] - 1] ^= 1;
+    Packet invalid_signature = raw(plain, 3 + plain[2]);
+    const unsigned before = verify_calls;
+    f.unchanged(invalid_signature);
+    assert(verify_calls == before + 1); // MAC was valid; Ed25519 denied control.
+    LocalIdentity stranger;
+    Packet wrong_publisher = command(stranger, initial, text.c_str());
+    f.unchanged(wrong_publisher);
+    Packet wrong_target = command(f.publisher, initial, text.c_str(), publicHex(stranger).c_str());
+    f.unchanged(wrong_target);
+    assert(f.offer(valid) == 1); // Rejected requests did not reserve the sequence.
+    assert(f.mesh.clock.now == initial + 20 && f.cli.callbacks.hooks == 1);
+  } else if (scenario == "durable_replay") {
+    Packet packet = command(f.publisher, initial + 5, "clock sync");
+    assert(f.offer(packet) == 1);
+    f.unchanged(packet);
+    FleetChannel rebooted(&f.disk);
+    const unsigned before = f.dispatched;
+    const auto saved = f.disk.files.at(Path);
+    assert(f.offer(packet, &rebooted) == 0);
+    assert(f.dispatched == before && f.mesh.clock.now == initial + 6);
+    assert(f.mesh.clock.sets == 1 && f.cli.callbacks.hooks == 1);
+    assert(f.disk.files.at(Path) == saved);
+  } else if (scenario == "invalid_rtc") {
+    Packet packet = command(f.publisher, initial + 5, "clock sync");
+    for (uint32_t invalid : {0u, FleetCommand::MinEpoch - 1}) {
+      f.mesh.clock.now = invalid;
+      f.unchanged(packet);
+    }
+    assert(f.dispatched == 0 && f.mesh.clock.sets == 0);
+    f.mesh.clock.now = initial;
+    assert(f.offer(packet) == 1);
+    assert(f.mesh.clock.now == initial + 6 && f.cli.callbacks.hooks == 1);
+  } else if (scenario == "expired_and_future") {
+    Packet expired = signedUnchecked(f.publisher, initial - 300, initial - 1, "clock sync");
+    f.unchanged(expired);
+    Packet future = command(f.publisher, initial + FleetCommand::MaxClockLead + 1, "clock sync");
+    f.unchanged(future);
+    assert(f.dispatched == 0 && f.mesh.clock.sets == 0 && f.cli.callbacks.hooks == 0);
+    Packet current = command(f.publisher, initial + 1, "clock sync");
+    assert(f.offer(current) == 1);
+    assert(f.mesh.clock.now == initial + 2 && f.cli.callbacks.hooks == 1);
+  } else if (scenario == "signed_invalid_clock_forms") {
+    for (const char* text : {"get clock", "set clock", "clock sync extra", "time NaN",
+                            "time -1", "time 4294967296", "time 1735689599"}) {
+      Packet packet = signedUnchecked(f.publisher, initial, initial + 300, text);
+      f.unchanged(packet);
+    }
+    assert(f.dispatched == 0 && f.mesh.clock.sets == 0 && f.cli.callbacks.hooks == 0);
+  } else assert(false);
+  printf("Fleet clock scenario %s passed\n", argv[1]);
+}
+'''
+
+
+class FleetClockControlTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        if compiler is None:
+            raise unittest.SkipTest("host C++ compiler required")
+        cls.work = tempfile.TemporaryDirectory(prefix="meshcore-fleet-clock-")
+        work = Path(cls.work.name)
+        (work / "helpers").mkdir()
+        clock_getter = "uint32_t getCurrentTime() const{return now;}"
+        assert MESH.count(clock_getter) == 1
+        mesh = MESH.replace(clock_getter, clock_getter + "\n unsigned sets=0;"
+                            " void setCurrentTime(uint32_t value){++sets;now=value;}")
+        for name, content in {
+            "Utils.h": UTILS, "FS.h": FILESYSTEM, "Mesh.h": mesh,
+            "helpers/RadioProfileCLI.h": PROFILE,
+            "Packet.h": '#pragma once\n#include <Mesh.h>\n',
+            "Arduino.h": '#pragma once\n#include <Mesh.h>\n',
+            "Adafruit_LittleFS.h": r'''
+#pragma once
+#include <FS.h>
+using Adafruit_LittleFS = fs::FS;
+namespace Adafruit_LittleFS_Namespace {}
+#define FILE_O_READ 1
+#define FILE_O_WRITE 2
+#define LFS_ERR_NOENT -2
+struct lfs_info {};
+inline int lfs_stat(fs::FS* fs,const char* path,lfs_info*) {return fs->exists(path)?0:LFS_ERR_NOENT;}
+''',
+            "InternalFileSystem.h": '#pragma once\n#include <FS.h>\ninline fs::FS InternalFS;\n',
+        }.items():
+            (work / name).write_text(content)
+        sha = ROOT / "test/mocks/SHA256.h"
+        (work / "SHA256.h").write_text(f'''
+#pragma once
+#define SHA256 SHA256Base
+#include "{sha}"
+#undef SHA256
+class SHA256 : public SHA256Base {{ public:
+ void finalize(void* out,size_t length) {{ SHA256Base::finalize(static_cast<uint8_t*>(out),length); }}
+ void finalizeHMAC(const uint8_t* key,size_t size,void* out,size_t length) {{
+  SHA256Base::finalizeHMAC(key,size,static_cast<uint8_t*>(out),length); }}
+}};
+''')
+        common = (ROOT / "src/helpers/CommonCLI.cpp").read_text()
+        # The ESP-only *next* command opens its #if before the clock-sync
+        # branch's closing brace. Separate that following command's guard from
+        # the unconditional clock body without altering any clock statements.
+        sync_start = common.index('if (memcmp(command, "clock sync", 10) == 0)')
+        next_guard = '#if defined(ESP_PLATFORM) && defined(ADMIN_PASSWORD) && !defined(WEBCONFIG_DISABLED)'
+        guard_start = common.index(next_guard, sync_start)
+        assert common[guard_start + len(next_guard):].startswith(
+            '\n    } else if (memcmp(command, "start webconfig", 15) == 0')
+        clock_source = common[:guard_start] + common[guard_start + len(next_guard):]
+        branches = "\nelse ".join(body(clock_source, signature) for signature in (
+            'if (memcmp(command, "clock sync", 10) == 0)',
+            'if (memcmp(command, "clock", 5) == 0)',
+            'if (memcmp(command, "time ", 5) == 0)',
+        ))
+        clock_fixture = HARNESS.replace("@ATOI@", body(common, "static uint32_t _atoi("))
+        clock_fixture = clock_fixture.replace("@CLOCK_BRANCHES@", branches)
+        transport = (ROOT / "src/helpers/TransportKeyStore.cpp").read_text()
+        transport_methods = "\n".join(body(transport, signature) for signature in (
+            "uint16_t TransportKey::calcTransportCode(", "bool TransportKey::isNull(",
+        ))
+        shared = RUNTIME[:RUNTIME.index("static unsigned apply(")]
+        fixture = work / "fixture.cpp"
+        fixture.write_text(shared.replace("@TRANSPORT_METHODS@", transport_methods) + clock_fixture)
+        sanitizer_flags = (["-fsanitize=address,undefined", "-fno-omit-frame-pointer", "-no-pie"]
+                           if os.environ.get("MESHCORE_FLEET_SANITIZERS") == "1" else [])
+        cls.binaries = {}
+        for backend in ("RP2040", "NRF52"):
+            binary = work / ("fleet-clock-" + backend.lower())
+            sources = [str(ROOT / "src/helpers/FleetChannel.cpp"),
+                       str(ROOT / "src/helpers/FleetCommand.cpp")]
+            if backend == "NRF52":
+                sources.append(str(ROOT / "src/helpers/AtomicFileWriter.cpp"))
+            built = subprocess.run([
+                compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                "-Wno-unused-function", "-Wno-unused-parameter", "-Wno-misleading-indentation", "-Wno-sign-compare",
+                "-O2", *sanitizer_flags, "-D" + backend + "_PLATFORM",
+                "-I", str(work), "-isystem", str(ROOT / "test/mocks"),
+                "-I", str(ROOT / "src"), str(fixture), *sources, "-lcrypto", "-o", str(binary),
+            ], capture_output=True, text=True, timeout=60)
+            if built.returncode:
+                cls.work.cleanup()
+                raise AssertionError(backend + ":\n" + built.stdout + built.stderr)
+            cls.binaries[backend] = binary
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.work.cleanup()
+
+    def scenario(self, name):
+        for backend, binary in self.binaries.items():
+            with self.subTest(backend=backend):
+                result = subprocess.run([str(binary), name], capture_output=True,
+                                        text=True, timeout=10)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_clock_getter_preserves_rtc_and_durably_consumes_request(self):
+        self.scenario("clock_read")
+
+    def test_clock_sync_uses_signed_sender_utc_plus_one_and_calls_manual_hook(self):
+        self.scenario("sync_forward")
+
+    def test_clock_sync_uint32_boundary_never_wraps_rtc_or_fires_manual_hook(self):
+        self.scenario("sync_uint32_boundary")
+
+    def test_explicit_forward_time_calls_production_setter_and_hook(self):
+        self.scenario("explicit_forward")
+
+    def test_backward_and_equal_clock_requests_never_set_rtc_or_call_hook(self):
+        self.scenario("backward_and_equal")
+
+    def test_invalid_signature_publisher_and_target_cannot_change_clock(self):
+        self.scenario("signatures_and_target")
+
+    def test_replayed_clock_change_is_denied_before_and_after_receiver_reboot(self):
+        self.scenario("durable_replay")
+
+    def test_invalid_or_unset_rtc_has_no_signed_clock_bootstrap_exception(self):
+        self.scenario("invalid_rtc")
+
+    def test_expired_and_excessively_future_clock_envelopes_never_dispatch(self):
+        self.scenario("expired_and_future")
+
+    def test_valid_signature_never_grants_malformed_clock_commands(self):
+        self.scenario("signed_invalid_clock_forms")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_fleet_command/test_fleet_command.cpp

@@ -84,13 +84,120 @@ TEST_F(FleetCommandTest, ExplicitRadioAndFilterFamiliesAreAllowed) {
   }
 }
 
+TEST_F(FleetCommandTest, ScheduledPrimaryAndSecondaryFamiliesAcceptCanonicalForms) {
+  for (const char* family : {"radioat", "tempradioat", "radioat2", "tempradioat2"}) {
+    for (const char* verb : {"get", "del"}) {
+      for (const char* selector : {"", " all", " 1", " 3"}) {
+        const std::string command = std::string(verb) + ' ' + family + selector;
+        EXPECT_TRUE(Fleet::commandAllowed(command.c_str())) << command;
+      }
+    }
+  }
+  for (const char* command : {
+      "get radioat 255", "del tempradioat 255", "get radioat2 4", "del tempradioat2 4",
+      "set radioat 910.5,500,5,5,+1",
+      "set radioat 910.5,500,5,5,1800000000,auto",
+      "set radioat 910.5,500,5,5,+1,0",
+      "set radioat 910.5,500,5,5,+1,8",
+      "set radioat 910.5,500,5,5,4294967295,65528",
+      "set tempradioat 910.5,500,5,5,+1,+2",
+      "set tempradioat 910.5,500,5,5,1800000000,1800000600,32",
+      "set tempradioat 910.5,500,5,5,+1,1800000600,auto",
+      "set tempradioat 910.5,500,5,5,1800000000,+2,auto",
+      "set radioat2 910.5,500,5,5,rx,+1",
+      "set radioat2 910.5,500,5,5,rxtx,1800000000,0",
+      "set radioat2 910.5,500,5,5,rxtx,+1,auto",
+      "set tempradioat2 910.5,500,5,5,rx,+1,+2",
+      "set tempradioat2 910.5,500,5,5,rxtx,1800000000,1800000600,65528",
+      "set tempradioat2 910.5,500,5,5,rxtx,+1,1800000600,auto"}) {
+    EXPECT_TRUE(Fleet::commandAllowed(command)) << command;
+    const auto bytes = encode(command);
+    ASSERT_FALSE(bytes.empty()) << command;
+    ASSERT_TRUE(decode(bytes)) << command;
+    EXPECT_STREQ(command, decoded.command);
+  }
+}
+
+TEST_F(FleetCommandTest, ScheduledCommandsRejectMalformedTimesFieldsAndSelectors) {
+  for (const char* command : {
+      "get radioat 0", "get radioat 01", "get radioat +1", "get radioat -1",
+      "get radioat 256", "del tempradioat 4294967295", "del radioat 1 all",
+      "get radioat2 5", "del tempradioat2 5", "get radioat2 01", "get radioat2 ALL",
+      "get radioat.all", "del tempradioat2.extra all", "get radioat  all",
+      "set radioat off", "set tempradioat off", "set radioat2 off", "set tempradioat2 off",
+      "set radioat 910.5,500,5,5", "set tempradioat 910.5,500,5,5,+1",
+      "set radioat2 910.5,500,5,5,+1", "set tempradioat2 910.5,500,5,5,rx,+1",
+      "set radioat 910.5,500,5,5,rx,+1",
+      "set radioat2 910.5,500,5,5,rx&tx,+1", "set radioat2 910.5,500,5,5,RX,+1",
+      "set radioat 910.5,500,5,5,+0", "set radioat 910.5,500,5,5,0",
+      "set radioat 910.5,500,5,5,+", "set radioat 910.5,500,5,5,++1",
+      "set radioat 910.5,500,5,5,+1m", "set radioat 910.5,500,5,5,+1h",
+      "set radioat 910.5,500,5,5,+1.5", "set radioat 910.5,500,5,5,-1",
+      "set radioat 910.5,500,5,5,1e9", "set radioat 910.5,500,5,5,4294967296",
+      "set radioat 910.5,500,5,5,+71582789",
+      "set tempradioat 910.5,500,5,5,+1,+0",
+      "set tempradioat2 910.5,500,5,5,rx,+1,4294967296",
+      "set radioat 910.5,500,256,5,+1", "set radioat 910.5,500,5,256,+1",
+      "set radioat 910.5,500,+5,5,+1", "set radioat 910.5,500,5,5,+1,7",
+      "set radioat 910.5,500,5,5,+1,65529", "set radioat 910.5,500,5,5,+1,AUTO",
+      "set radioat 910.5,500,5,5,+1,+8", "set radioat 910.5,500,5,5,+1,auto,8",
+      "set radioat 910.5,500,5,5,+1,", "set radioat 910.5,500,5,5, +1",
+      "set tempradioat2 910.5,500,5,5,rx,+1,+2;reboot",
+      "set tempradioat2 910.5,500,5,5,rx,+1,+2\nreboot"}) {
+    EXPECT_FALSE(Fleet::commandAllowed(command)) << command;
+    EXPECT_TRUE(encode(command).empty()) << command;
+  }
+  const std::string prefix = "set tempradioat2 910.5,500,5,5,rxtx,+";
+  const std::string suffix = "1,+2,auto";
+  const std::string maximum = prefix
+      + std::string(Fleet::MaxCommandLength - prefix.size() - suffix.size(), '0') + suffix;
+  ASSERT_EQ(Fleet::MaxCommandLength, maximum.size());
+  EXPECT_TRUE(Fleet::commandAllowed(maximum.c_str()));
+  EXPECT_TRUE(decode(encode(maximum.c_str())));
+  const std::string oversized = prefix + '0' + maximum.substr(prefix.size());
+  EXPECT_FALSE(Fleet::commandAllowed(oversized.c_str()));
+}
+
+TEST_F(FleetCommandTest, ClockManagementOnlyAdmitsExactBoundedCommands) {
+  for (const char* command : {"clock", "clock sync", "time 1735689600",
+                              "time 1800000000", "time 4294967295"}) {
+    EXPECT_TRUE(Fleet::commandAllowed(command)) << command;
+    ASSERT_TRUE(decode(encode(command))) << command;
+    EXPECT_STREQ(command, decoded.command);
+  }
+  for (const char* command : {
+      "get clock", "set clock 1800000000", "clock now", "clock sync now", "clock.sync",
+      "clkreboot", "clock sync;reboot", "clock sync\nreboot", "clock  sync",
+      "time", "time ", "time  1800000000", "time +1800000000", "time -1800000000",
+      "time 1735689599", "time 0", "time 4294967296", "time 9999999999999999999",
+      "time 1800000000.0", "time 1e9", "time 1800000000 now", "time 1800000000;reboot",
+      "time 1800000000\nreboot", "time\t1800000000", "TIME 1800000000"}) {
+    EXPECT_FALSE(Fleet::commandAllowed(command)) << command;
+    EXPECT_TRUE(encode(command).empty()) << command;
+  }
+}
+
+TEST_F(FleetCommandTest, ClockManagementDoesNotBypassEnvelopeClockAndExpiryPolicy) {
+  for (const char* command : {"clock", "clock sync", "time 1800000000"}) {
+    const auto bytes = encode(command);
+    ASSERT_TRUE(decode(bytes)) << command;
+    EXPECT_EQ(Now, decoded.sequence);
+    EXPECT_FALSE(decode(bytes, 0)) << command;
+    EXPECT_FALSE(decode(bytes, Fleet::MinEpoch - 1)) << command;
+    EXPECT_FALSE(decode(bytes, Now - 61)) << command;
+    EXPECT_FALSE(decode(bytes, Now + 121)) << command;
+    EXPECT_TRUE(encode(command, Now, Now + 601).empty()) << command;
+    EXPECT_TRUE(encode(command, Fleet::MinEpoch - 1, Fleet::MinEpoch).empty()) << command;
+  }
+}
+
 TEST_F(FleetCommandTest, NoGenericAdminNamespaceOrPrefixEscapes) {
   for (const char* command : {
       "get prv.key", "backup prv.key 1234567890ABCDEF", "get wifi.password",
       "set wifi.password abc", "setperm abc 3", "reboot", "start ota",
       "get fleet", "set fleet.channel abc", "region add usa", "get radio",
       "set radio 910.5,500,5,5", "get radio2.reply", "get radio2.scan",
-      "set tx.reply both", "set radioat2 910.5,500,5,5,rxtx,+5",
+      "set tx.reply both", "get tempradio", "set tempradio 910.5,500,5,5,10",
       "get flood.filter.secret", "get flood.filtering", "get flood.rulebook",
       "get flood.rule.1.secret", "get flood.filter.blacklistx",
       "get flood.channel.scope.required", "get flood.channel.data.secret",

test/test_fleet_command_crypto.py

@@ -235,6 +235,104 @@ class FleetCommandCryptoTests(unittest.TestCase):
             with self.subTest(command=command):
                 self.assertEqual(self.decode(envelope(command)), "reject")
 
+    def test_scheduled_radio_commands_have_cross_language_real_signatures(self):
+        for command in [
+                "set radioat 910.5,500,5,5,+1,auto",
+                "set radioat 910.5,500,5,5,1800000000,65528",
+                "set tempradioat 910.5,500,5,5,+1,+2,32",
+                "set tempradioat 910.5,500,5,5,+1,1800000600,auto",
+                "set radioat2 910.5,500,5,5,rx,+1,0",
+                "set tempradioat2 910.5,500,5,5,rxtx,1800000000,1800000600,auto"]:
+            with self.subTest(command=command):
+                expected = envelope(command)
+                actual = self.run_tool("encode", CHANNEL.hex(), "all", NOW, NOW + 120, command)
+                self.assertEqual(actual, expected.hex())
+                PUBLISHER.public_key().verify(expected[-64:], DOMAIN + CHANNEL + expected[:-64])
+                self.assertEqual(self.decode(expected), f"{NOW}\n{NOW + 120}\n{command}")
+
+    def test_schedule_inspection_and_deletion_are_bounded_and_exact(self):
+        for family in ["radioat", "tempradioat", "radioat2", "tempradioat2"]:
+            for verb in ["get", "del"]:
+                maximum = 4 if family.endswith("2") else 255
+                for selector in ["", " all", " 1", f" {maximum}"]:
+                    command = f"{verb} {family}{selector}"
+                    with self.subTest(command=command):
+                        self.assertEqual(self.decode(envelope(command)), f"{NOW}\n{NOW + 120}\n{command}")
+                for selector in [" 0", " 01", " +1", " -1", " ALL", " 1 all", f" {maximum + 1}"]:
+                    command = f"{verb} {family}{selector}"
+                    with self.subTest(command=command):
+                        self.assertEqual(self.decode(envelope(command)), "reject")
+
+    def test_signed_malformed_schedule_and_alias_commands_are_rejected(self):
+        for command in [
+                "set radioat off", "set tempradioat off", "set radioat2 off", "set tempradioat2 off",
+                "set radioat 910.5,500,5,5", "set tempradioat 910.5,500,5,5,+1",
+                "set radioat2 910.5,500,5,5,+1", "set tempradioat2 910.5,500,5,5,rx,+1",
+                "set radioat 910.5,500,5,5,rx,+1", "set radioat2 910.5,500,5,5,rx&tx,+1",
+                "set radioat 910.5,500,5,5,+0", "set radioat 910.5,500,5,5,++1",
+                "set radioat 910.5,500,5,5,+1m", "set radioat 910.5,500,5,5,+1.5",
+                "set radioat 910.5,500,5,5,-1", "set radioat 910.5,500,5,5,1e9",
+                "set radioat 910.5,500,5,5,4294967296", "set radioat 910.5,500,5,5,+71582789",
+                "set radioat 910.5,500,256,5,+1", "set radioat 910.5,500,5,5,+1,7",
+                "set radioat 910.5,500,5,5,+1,65529", "set radioat 910.5,500,5,5,+1,AUTO",
+                "set radioat 910.5,500,5,5,+1,+8", "set radioat 910.5,500,5,5,+1,auto,8",
+                "set radioat.extra 910.5,500,5,5,+1", "get radioat.all", "set radioat 910.5,500,5,5, +1",
+                "set tempradioat2 910.5,500,5,5,rx,+1,+2;reboot",
+                "set tempradioat2 910.5,500,5,5,rx,+1,+2\nreboot",
+                "set radio 910.5,500,5,5", "set tempradio 910.5,500,5,5,10"]:
+            with self.subTest(command=command):
+                self.assertEqual(self.decode(envelope(command)), "reject")
+
+    def test_multitarget_scheduled_request_is_signed_and_body_bounded(self):
+        command = "set tempradioat2 910.5,500,5,5,rxtx,+1,+2,auto"
+        entries = [(4, PUBLIC[:4]), (6, OTHER_PUBLIC[:6])]
+        token = f"{PUBLIC[:4].hex()},{OTHER_PUBLIC[:6].hex()}"
+        expected = envelope2(entries, command)
+        actual = self.run_tool("encode2", CHANNEL.hex(), token, NOW, NOW + 120, command)
+        self.assertEqual(actual, expected.hex())
+        self.assertEqual(self.decode(expected, metadata=True), f"{NOW}\n{NOW + 120}\n{command}\n1")
+        self.assertNotEqual(self.decode(expected, self_key=OTHER_PUBLIC), "reject")
+        changed_command = command.replace("+2", "+3")
+        self.assertEqual(self.decode(envelope2(entries, changed_command)[:-64] + expected[-64:]), "reject")
+        prefix, suffix = "set tempradioat2 910.5,500,5,5,rxtx,+", "1,+2,auto"
+        maximum = prefix + "0" * (72 - len(prefix) - len(suffix)) + suffix
+        self.assertEqual(len(maximum), 72)
+        self.assertNotEqual(self.decode(envelope(maximum)), "reject")
+        self.assertNotEqual(self.decode(envelope2(entries, maximum)), "reject")
+        oversized = prefix + "0" + maximum[len(prefix):]
+        self.assertEqual(self.decode(envelope(oversized)), "reject")
+
+    def test_clock_management_has_real_signatures_and_no_prefix_aliases(self):
+        for command in ["clock", "clock sync", "time 1735689600", "time 1800000000", "time 4294967295"]:
+            with self.subTest(command=command):
+                expected = envelope(command)
+                actual = self.run_tool("encode", CHANNEL.hex(), "all", NOW, NOW + 120, command)
+                self.assertEqual(actual, expected.hex())
+                self.assertEqual(self.decode(expected), f"{NOW}\n{NOW + 120}\n{command}")
+                self.assertEqual(self.decode(expected, publisher=OTHER_PUBLIC), "reject")
+                self.assertEqual(self.decode(expected, channel=bytes(range(17, 33))), "reject")
+        for command in [
+                "get clock", "set clock 1800000000", "clock now", "clock sync now", "clock.sync",
+                "clkreboot", "clock sync;reboot", "clock sync\nreboot", "clock  sync", "time",
+                "time  1800000000", "time +1800000000", "time -1800000000", "time 1735689599",
+                "time 0", "time 4294967296", "time 9999999999999999999", "time 1800000000.0",
+                "time 1e9", "time 1800000000 now", "time 1800000000;reboot", "TIME 1800000000"]:
+            with self.subTest(command=command):
+                self.assertEqual(self.decode(envelope(command)), "reject")
+
+    def test_signed_clock_requests_cannot_bootstrap_or_bypass_replay_time_boundaries(self):
+        for command in ["clock", "clock sync", "time 1800000000"]:
+            with self.subTest(command=command):
+                for sequence, expires, now in [
+                        (NOW, NOW + 120, 0), (NOW, NOW + 120, 1735689599),
+                        (NOW, NOW + 120, NOW - 61), (NOW, NOW + 120, NOW + 121),
+                        (NOW, NOW + 601, NOW), (1735689599, 1735689600, 1735689600)]:
+                    self.assertEqual(self.decode(envelope(command, sequence, expires), now=now), "reject")
+                original = envelope(command)
+                forged = envelope(command, sequence=NOW + 1)[:-64] + original[-64:]
+                self.assertEqual(self.decode(forged), "reject")
+                self.assertNotEqual(self.decode(original, now=NOW + 120), "reject")
+
     def test_exact_and_oversized_command_boundaries(self):
         maximum = "set flood.rule.1 " + "a" * (72 - len("set flood.rule.1 "))
         accepted = envelope(maximum)

test/test_fleet_radio_schedules.py

@@ -0,0 +1,443 @@
+#!/usr/bin/env python3
+"""Run signed fleet commands through the actual primary and secondary schedulers.
+
+The receiver, signed codec, secondary scheduler and storage transactions are
+production code; primary CommonCLI branches and repeater methods are extracted
+unchanged. Clock, filesystem I/O, physical radio, primary-save endpoint and
+packet admission are mocked; signatures, AES and MAC checks remain real.
+"""
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from cpp_source import body
+from test_fleet_channel_runtime import UTILS, FILESYSTEM, MESH, HARNESS as RUNTIME_HARNESS
+from test_repeater_radio_timing_integration import HARNESS as PRIMARY_HARNESS
+
+ROOT = Path(__file__).resolve().parents[1]
+HARDWARE = r'''
+#pragma once
+#include <RadioProfiles.h>
+namespace mesh {
+class RTCClock {public:virtual ~RTCClock()=default;virtual uint32_t getCurrentTime()=0;};
+enum class RadioParamApplyResult:uint8_t {APPLIED,BUSY,FAILED};
+class Radio {public:
+ virtual ~Radio()=default;
+ static constexpr uint32_t CarrierWaveDefaultMillis=10000,CarrierWaveMaxMillis=60000;
+ virtual bool supportsCarrierWave()const{return false;}
+ virtual bool isCarrierWaveActive()const{return false;}
+ virtual uint32_t carrierWaveRemainingMillis()const{return 0;}
+ virtual uint8_t carrierWaveProfile()const{return 0;}
+ virtual RadioParamApplyResult setCarrierWave(uint8_t,uint32_t){return RadioParamApplyResult::FAILED;}
+ virtual bool isInRecvMode()const{return true;}
+ virtual RadioProfiles* profiles()=0;
+ virtual const RadioProfiles* profiles()const=0;
+ virtual bool validateProfile(const RadioProfileParams&)const=0;
+ virtual uint16_t profilePreamble(uint8_t)const=0;
+ virtual RadioParamApplyResult trySetPrimaryParams(const RadioProfileParams&,bool,const uint32_t*)=0;
+};
+}
+'''
+
+HARNESS = r'''
+struct PhysicalRadio:mesh::Radio {
+ mesh::RadioProfiles state;
+ PhysicalRadio(){state.primary.freq=909.5;state.primary.bw=62.5;state.primary.sf=7;state.primary.cr=5;}
+ mesh::RadioProfiles* profiles()override{return &state;}
+ const mesh::RadioProfiles* profiles()const override{return &state;}
+ bool validateProfile(const mesh::RadioProfileParams& p)const override{return mesh::RadioProfiles::valid(p)&&p.bw<=500;}
+ uint16_t profilePreamble(uint8_t profile)const override{return state.preamble(profile,32);}
+ mesh::RadioParamApplyResult trySetPrimaryParams(const mesh::RadioProfileParams&,bool,const uint32_t*)override{
+  assert(false&&"radio2 scheduler must never mutate primary");return mesh::RadioParamApplyResult::FAILED;}
+};
+static uint32_t stageDirectProfile(RadioProfileCLI& profiles){
+ const uint32_t before=profiles.replyMutationGeneration();char reply[160]={};
+ profiles.beginReplyCommand();
+ assert(profiles.handle("set radio2 912.5,500,8,5,rxtx",reply,sizeof(reply),true));
+ profiles.endReplyCommand();
+ assert(!strncmp(reply,"OK",2)&&profiles.hasReplyMutation());
+ assert(profiles.replyMutationGeneration()!=before);return profiles.replyMutationGeneration();
+}
+struct Fixture {
+ fs::FS disk;Mesh mesh;PhysicalRadio radio;RadioProfileCLI profiles;FleetChannel fleet{&disk};LocalIdentity publisher;
+ unsigned dispatched=0;char reply[160]={};
+ Fixture(){profiles.begin(&disk,&radio,&mesh.clock,true);enroll(fleet,publisher);bind(fleet,mesh,profiles);}
+ const char* get(const char* text){assert(profiles.handle(text,reply,sizeof(reply)));return reply;}
+ Packet send(const char* text,bool mutation_expected=true){
+  fake_ms+=1000; // Each request occupies a fresh verification-budget interval.
+  Packet packet=command(publisher,mesh.clock.now,text);fleet.receive(&packet,mesh);
+  const unsigned before=dispatched;
+  fleet.service(mesh,profiles,"scheduler",[&](uint32_t sequence,const char* command,char* response){
+   ++dispatched;assert(storage::readLE32(disk.files[Path].data()+56)==sequence);
+   assert(profiles.handle(command,response,160,true));strcpy(reply,response);
+  });
+  assert(dispatched==before+1);assert(!strncmp(reply,"OK",2)||!strncmp(text,"get ",4));
+  assert(fleet.waiting()==mutation_expected);return packet;
+ }
+ void drain(bool success){
+  while(!mesh.manager.outbound.empty()){
+   Packet* p=mesh.manager.removeOutboundByIdx(0);
+   if(success)fleet.complete(p,profiles);else fleet.fail(p,profiles);
+  }
+  profiles.loop();
+ }
+ void advance(uint32_t ms,bool wall=true){fake_ms+=ms;if(wall)mesh.clock.now+=ms/1000;profiles.loop();}
+};
+int main(int argc,char** argv){
+ assert(argc==2);const std::string scenario=argv[1];Fixture f;
+ if(scenario=="permanent_ack"){
+  Packet packet=f.send("set radioat2 910.5,500,8,5,rxtx,+1");
+  assert(!f.radio.state.enabled());f.advance(61000);assert(!f.radio.state.enabled());
+  assert(f.profiles.hasReplyMutation());f.drain(true);
+  assert(!f.fleet.waiting()&&!f.profiles.hasReplyMutation());
+  assert(f.radio.state.secondary.params.freq==910.5f&&!f.radio.state.secondary_temporary);
+  PhysicalRadio reboot;RadioProfileCLI restored;restored.begin(&f.disk,&reboot,&f.mesh.clock,true);
+  assert(reboot.state.secondary.params.freq==910.5f);
+  const unsigned before=f.dispatched;f.fleet.receive(&packet,f.mesh);
+  f.fleet.service(f.mesh,f.profiles,"node",[&](uint32_t,const char*,char*){++f.dispatched;});
+  assert(f.dispatched==before);
+ }else if(scenario=="temporary_ack"){
+  f.send("set tempradioat2 910.5,500,8,5,rxtx,+1,+2");
+  f.advance(61000);assert(!f.radio.state.enabled());f.drain(true);
+  assert(f.radio.state.secondary_temporary&&f.radio.state.secondary.params.freq==910.5f);
+  f.advance(59000);assert(!f.radio.state.enabled());
+ }else if(scenario=="ack_fail"){
+  f.send("set radioat2 910.5,500,8,5,rxtx,+1");f.drain(false);
+  f.advance(61000);assert(!f.radio.state.enabled());assert(!strcmp(f.get("get radioat2 all"),"> slots: "));
+  ++f.mesh.clock.now;f.send("set tempradioat2 910.5,500,8,5,rxtx,+1,+2");f.drain(false);
+  f.advance(61000);assert(!f.radio.state.enabled());
+ }else if(scenario=="queue_fail"){
+  f.mesh.queue_ok=false;f.send("set radioat2 910.5,500,8,5,rxtx,+1",false);
+  f.advance(61000);assert(!f.radio.state.enabled());assert(!f.profiles.hasReplyMutation());
+ }else if(scenario=="ack_timeout"){
+  f.send("set radioat2 910.5,500,8,5,rxtx,+1");f.advance(301000);
+  f.fleet.service(f.mesh,f.profiles,"scheduler",[](uint32_t,const char*,char*){assert(false);});
+  f.profiles.loop();assert(!f.fleet.waiting()&&!f.profiles.hasReplyMutation());
+  assert(!f.radio.state.enabled()&&f.mesh.manager.outbound.empty());
+ }else if(scenario=="both_copies"){
+  f.mesh.fanout=true;f.send("set radioat2 910.5,500,8,5,rxtx,+1");f.advance(61000);
+  Packet* one=f.mesh.manager.removeOutboundByIdx(0);f.fleet.complete(one,f.profiles);f.profiles.loop();
+  assert(f.fleet.waiting()&&!f.radio.state.enabled());f.drain(true);
+  assert(f.radio.state.secondary.params.freq==910.5f);
+ }else if(scenario=="expired_lease"){
+  f.send("set tempradioat2 910.5,500,8,5,rxtx,+1,+2");
+  f.advance(121000);assert(!f.radio.state.enabled());f.drain(true);
+  assert(!f.radio.state.enabled());assert(!strcmp(f.get("get tempradioat2 all"),"> slots: "));
+ }else if(scenario=="save_fail"){
+  f.send("set radioat2 910.5,500,8,5,rxtx,+1");f.drain(true);f.disk.fail_write=true;
+  f.advance(61000);assert(!f.radio.state.enabled());assert(strstr(f.get("get radioat2 all"),"1@"));
+  f.disk.fail_write=false;f.advance(60000);assert(f.radio.state.secondary.params.freq==910.5f);
+ }else if(scenario=="delete_ack"){
+  assert(f.profiles.handle("set tempradioat2 910.5,500,8,5,rxtx,+1,+3",f.reply,160));
+  f.advance(61000);assert(f.radio.state.secondary_temporary);
+  f.send("del tempradioat2 1");assert(f.radio.state.secondary_temporary);
+  f.drain(false);assert(f.radio.state.secondary_temporary);
+  ++f.mesh.clock.now;f.send("del tempradioat2 all");f.drain(true);
+  assert(!f.radio.state.enabled());
+  ++f.mesh.clock.now;f.send("set radioat2 911.5,500,8,5,rxtx,+1");f.drain(true);
+  ++f.mesh.clock.now;f.send("del radioat2 all");f.drain(false);
+  assert(strstr(f.get("get radioat2 all"),"1@"));
+  ++f.mesh.clock.now;f.send("del radioat2 1");f.drain(true);
+  assert(!strcmp(f.get("get radioat2 all"),"> slots: "));
+ }else if(scenario=="unsigned_no_schedule"){
+  Packet packet=command(f.publisher,f.mesh.clock.now,"set radioat2 910.5,500,8,5,rxtx,+1");
+  packet.payload[1]^=1;f.fleet.receive(&packet,f.mesh);
+  f.fleet.service(f.mesh,f.profiles,"node",[&](uint32_t,const char*,char*){assert(false);});
+  f.advance(61000);assert(!f.radio.state.enabled());assert(f.disk.files.count("/radio_profiles")==0);
+ }else if(scenario=="bad_signature_no_schedule"){
+  Packet packet=command(f.publisher,f.mesh.clock.now,"set radioat2 910.5,500,8,5,rxtx,+1");
+  uint8_t clear[184];const int size=Utils::MACThenDecrypt(channel_key,clear,packet.payload+1,packet.payload_len-1);
+  assert(size>0);clear[3+clear[2]-1]^=1;packet=raw(clear,size);
+  const auto before=f.disk.files;f.fleet.receive(&packet,f.mesh);
+  f.fleet.service(f.mesh,f.profiles,"node",[&](uint32_t,const char*,char*){assert(false);});
+  assert(verify_calls&&f.disk.files==before);f.advance(61000);assert(!f.radio.state.enabled());
+ }else if(scenario=="clock_backwards"){
+  f.send("set tempradioat2 910.5,500,8,5,rxtx,+1,+2");f.drain(true);
+  f.advance(61000);assert(f.radio.state.secondary_temporary);f.mesh.clock.now-=3600;
+  f.advance(59000,false);assert(!f.radio.state.enabled());
+ }else if(scenario=="get_slots"){
+  f.send("set radioat2 910.5,500,8,5,rxtx,+1");f.drain(true);
+  ++f.mesh.clock.now;f.send("get radioat2 all",false);assert(strstr(f.reply,"1@"));
+  ++f.mesh.clock.now;f.send("get radioat2 1",false);assert(strstr(f.reply,"910.5"));
+ }else if(scenario=="replaced_secondary_ack"||scenario=="replaced_secondary_fail"||scenario=="replaced_secondary_timeout"){
+  f.send("set radioat2 910.5,500,8,5,rxtx,+1");
+  // Local primary recovery cancels the old secondary transition. A newer
+  // direct admin command then stages an independent, separately owned reply.
+  assert(f.profiles.finishReplyMutation(false));const uint32_t generation=stageDirectProfile(f.profiles);
+  if(scenario=="replaced_secondary_timeout"){
+   f.advance(301000);f.fleet.service(f.mesh,f.profiles,"node",[](uint32_t,const char*,char*){assert(false);});
+  }else f.drain(scenario=="replaced_secondary_ack");
+  f.profiles.loop();assert(!f.fleet.waiting()&&f.profiles.hasReplyMutation());
+  assert(f.profiles.replyMutationGeneration()==generation&&!f.radio.state.enabled());
+  assert(f.disk.files.count("/radio_profiles")==0);
+  assert(f.profiles.finishReplyMutation(true));f.profiles.loop();
+  assert(!f.profiles.hasReplyMutation()&&f.radio.state.secondary.params.freq==912.5f);
+ }else assert(false);
+ printf("fleet scheduled radio %s passed\n",argv[1]);
+}
+'''
+
+PRIMARY_EXTRA = r'''
+struct PrimaryFixture {
+ fs::FS disk;Mesh mesh;PhysicalRadio physical;RadioProfileCLI profiles;MyMesh node;
+ FleetChannel fleet{&disk,&node};LocalIdentity publisher;unsigned dispatched=0;char reply[160]={};
+ PrimaryFixture(){fake_ms=1000;node.rtc.now=mesh.clock.now;profiles.begin(&disk,&physical,&mesh.clock,true);
+  enroll(fleet,publisher);bind(fleet,mesh,profiles);}
+ void handle(const char* text,char* reply){
+  if(!strncmp(text,"set ",4)){
+   const char* config=text+4;RTC* _rtc=&node.rtc;MyMesh* _callbacks=&node;
+   RadioProfileCLI& _radio_profiles=profiles;RadioPrefs* _prefs=&node._prefs;
+   auto appendRxPowerSavingAdjustmentNote=[](char*,RadioPrefs*,uint8_t,float){};
+   if(false){
+@PRIMARY_SET@
+   }else assert(false);
+  }else if(!strncmp(text,"del tempradioat",15))node.deleteScheduledRadioParams(true,text+15,reply);
+  else if(!strncmp(text,"del radioat",11))node.deleteScheduledRadioParams(false,text+11,reply);
+  else if(!strncmp(text,"get tempradioat",15))node.formatScheduledRadioParams(true,text+15,reply);
+  else if(!strncmp(text,"get radioat",11))node.formatScheduledRadioParams(false,text+11,reply);
+  else assert(false);
+ }
+ void send(const char* text,bool mutation_expected=true){
+  fake_ms+=1000;Packet packet=command(publisher,mesh.clock.now,text);fleet.receive(&packet,mesh);
+  const unsigned before=dispatched;
+  fleet.service(mesh,profiles,"primary",[&](uint32_t seq,const char* text,char* response){
+   ++dispatched;assert(storage::readLE32(disk.files[Path].data()+56)==seq);
+   assert(node.fleet_command_);handle(text,response);strcpy(reply,response);
+  });
+  assert(dispatched==before+1&&!node.fleet_command_);assert(!strncmp(reply,"OK",2)||!strncmp(text,"get ",4));
+  assert(fleet.waiting()==mutation_expected);
+ }
+ void advance(uint32_t seconds,bool wall=true){
+  fake_ms+=seconds*1000;node.uptime_millis+=uint64_t(seconds)*1000;node.last_millis=fake_ms;
+  if(wall){node.rtc.now+=seconds;mesh.clock.now+=seconds;}node.processScheduledRadioSettings();
+ }
+ void drain(bool success){
+  while(!mesh.manager.outbound.empty()){
+   Packet* p=mesh.manager.removeOutboundByIdx(0);
+   if(success)fleet.complete(p,profiles);else fleet.fail(p,profiles);
+  }
+ }
+};
+static int primary(int argc,char** argv){
+ assert(argc==2);const std::string scenario=argv[1];PrimaryFixture f;
+ if(scenario=="primary_permanent"){
+  f.send("set radioat 910.5,500,8,5,+1");assert(f.node.countScheduledRadioSettings(false)==0);
+  f.advance(61);assert(f.node.saves==0&&f.node.restores==0);f.drain(true);
+  assert(f.node.saves==0&&f.node.restores==0);f.node.processScheduledRadioSettings();
+  assert(f.node._prefs.freq==910.5f&&f.node.saves==1&&f.node.restores==1);
+ }else if(scenario=="primary_both_copies"){
+  f.mesh.fanout=true;f.send("set tempradioat 910.5,500,8,5,+1,+2");f.advance(61);
+  Packet* one=f.mesh.manager.removeOutboundByIdx(0);f.fleet.complete(one,f.profiles);
+  f.node.processScheduledRadioSettings();assert(!f.node.temp_radio_applied&&f.fleet.waiting());
+  f.drain(true);assert(!f.node.temp_radio_applied);f.node.processScheduledRadioSettings();
+  assert(f.node.temp_radio_applied&&f.node.applies==1);f.advance(59);assert(!f.node.temp_radio_applied);
+ }else if(scenario=="primary_queue_failure"){
+  f.mesh.queue_ok=false;f.send("set radioat 910.5,500,8,5,+1",false);
+  assert(!f.node.hasFleetReplyMutation());f.advance(61);assert(f.node.saves==0&&f.node.restores==0);
+ }else if(scenario=="primary_tx_failure"){
+  f.send("set tempradioat 910.5,500,8,5,+1,+2");f.drain(false);f.advance(61);
+  assert(!f.node.temp_radio_applied&&f.node.countScheduledRadioSettings(true)==0);
+ }else if(scenario=="primary_ack_timeout"){
+  f.send("set radioat 910.5,500,8,5,+1");f.advance(301);
+  f.fleet.service(f.mesh,f.profiles,"primary",[](uint32_t,const char*,char*){assert(false);});
+  f.node.processScheduledRadioSettings();assert(!f.fleet.waiting()&&!f.node.hasFleetReplyMutation());
+  assert(f.node.saves==0&&f.node.restores==0&&f.mesh.manager.outbound.empty());
+ }else if(scenario=="primary_expired_window"){
+  f.send("set tempradioat 910.5,500,8,5,+1,+2");f.advance(121);f.drain(true);
+  f.node.processScheduledRadioSettings();assert(!f.node.temp_radio_applied&&f.node.applies==0);
+ }else if(scenario=="primary_local_busy"){
+  f.send("set radioat 910.5,500,8,5,+1");
+  f.handle("set radioat 911.5,500,8,5,+2",f.reply);
+  assert(strstr(f.reply,"acknowledgement pending")&&f.node.countScheduledRadioSettings(false)==0);
+  f.handle("del radioat all",f.reply);assert(strstr(f.reply,"acknowledgement pending"));f.drain(false);
+  f.handle("set radioat 911.5,500,8,5,+2",f.reply);assert(!strncmp(f.reply,"OK",2));
+  assert(f.node.countScheduledRadioSettings(false)==1);
+ }else if(scenario=="primary_active_delete"){
+  f.handle("set tempradioat 910.5,500,8,5,+1,+3",f.reply);f.advance(61);assert(f.node.temp_radio_applied);
+  f.send("del tempradioat all");f.drain(false);f.node.processScheduledRadioSettings();assert(f.node.temp_radio_applied);
+  ++f.mesh.clock.now;++f.node.rtc.now;f.send("del tempradioat 1");f.drain(true);
+  assert(f.node.temp_radio_applied);f.node.processScheduledRadioSettings();assert(!f.node.temp_radio_applied);
+ }else if(scenario=="primary_expiry_during_pending"){
+  f.handle("set tempradioat 910.5,500,8,5,+1,+2",f.reply);f.advance(61);assert(f.node.temp_radio_applied);
+  f.send("set radioat 911.5,500,8,5,+2");f.advance(60);
+  assert(!f.node.temp_radio_applied&&f.node._prefs.freq==909.5f&&f.fleet.waiting());f.drain(false);
+ }else if(scenario=="primary_save_failure"){
+  f.send("set radioat 910.5,500,8,5,+1");f.drain(true);f.node._cli.save_success=false;f.advance(61);
+  assert(f.node._prefs.freq==909.5f&&f.node.restores==0);f.node._cli.save_success=true;
+  f.advance(61);assert(f.node._prefs.freq==910.5f&&f.node.restores==1);
+ }else if(scenario=="primary_get_slots"){
+  f.send("set radioat 910.5,500,8,5,+1");f.drain(true);f.node.processScheduledRadioSettings();
+  ++f.mesh.clock.now;++f.node.rtc.now;f.send("get radioat all",false);assert(strstr(f.reply,"910.5"));f.drain(true);
+  ++f.mesh.clock.now;++f.node.rtc.now;f.send("get radioat 1",false);assert(strstr(f.reply,"910.5"));
+  assert(f.node.countScheduledRadioSettings(false)==1&&f.node.saves==0);
+ }else if(scenario=="primary_bad_signature"){
+  Packet packet=command(f.publisher,f.mesh.clock.now,"set radioat 910.5,500,8,5,+1");
+  uint8_t clear[184];const int length=Utils::MACThenDecrypt(channel_key,clear,packet.payload+1,packet.payload_len-1);
+  assert(length>0);clear[3+clear[2]-1]^=1;packet=raw(clear,length);const auto before=f.disk.files;
+  f.fleet.receive(&packet,f.mesh);f.fleet.service(f.mesh,f.profiles,"primary",[&](uint32_t,const char*,char*){assert(false);});
+  assert(verify_calls&&f.disk.files==before&&!f.node.fleet_command_&&!f.node.hasFleetReplyMutation());
+  f.advance(61);assert(f.node.saves==0&&f.node.restores==0);
+ }else if(scenario=="primary_recovery_supersedes_delete"){
+  f.handle("set tempradioat 910.5,500,8,5,+1,+3",f.reply);f.advance(61);
+  f.send("del tempradioat all");f.node.applyTempRadioParams(911.5f,500,8,5,20);
+  f.drain(true);f.node.processScheduledRadioSettings();
+  assert(f.node.countScheduledRadioSettings(true)==1);
+  assert(f.node.scheduled_radio_settings[0].freq==911.5f);
+  f.advance(3);assert(f.node.temp_radio_applied&&f.node.applies==2);
+ }else if(scenario=="primary_recovery_supersedes_add"){
+  f.send("set tempradioat 910.5,500,8,5,+1,+2");assert(f.node.scheduleNormalRadio());
+  f.drain(true);f.node.processScheduledRadioSettings();
+  assert(f.node.countScheduledRadioSettings(true)==0);f.advance(61);assert(!f.node.temp_radio_applied);
+ }else if(scenario=="primary_ack_ignores_secondary"||scenario=="primary_fail_ignores_secondary"||scenario=="primary_timeout_ignores_secondary"){
+  f.send("set radioat 910.5,500,8,5,+1");const uint32_t generation=stageDirectProfile(f.profiles);
+  if(scenario=="primary_timeout_ignores_secondary"){
+   f.advance(301);f.fleet.service(f.mesh,f.profiles,"primary",[](uint32_t,const char*,char*){assert(false);});
+  }else f.drain(scenario=="primary_ack_ignores_secondary");
+  f.node.processScheduledRadioSettings();f.profiles.loop();
+  assert(!f.fleet.waiting()&&f.profiles.hasReplyMutation());
+  assert(f.profiles.replyMutationGeneration()==generation&&!f.physical.state.enabled());
+  assert(f.disk.files.count("/radio_profiles")==0);
+  assert(f.profiles.finishReplyMutation(true));f.profiles.loop();
+  assert(!f.profiles.hasReplyMutation()&&f.physical.state.secondary.params.freq==912.5f);
+ }else assert(false);
+ printf("fleet primary scheduled radio %s passed\n",argv[1]);return 0;
+}
+'''
+
+class FleetRadioSchedulesTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+  compiler=shutil.which("g++") or shutil.which("clang++")
+  if not compiler:raise unittest.SkipTest("host C++ compiler required")
+  cls.work=tempfile.TemporaryDirectory(prefix="fleet-radio-schedules-");work=Path(cls.work.name)
+  mesh=MESH.replace('#include <Identity.h>','#include <Hardware.h>\n#include <Identity.h>')
+  mesh=mesh.replace('struct Clock {','struct Clock :RTCClock {').replace('uint32_t getCurrentTime() const{return now;}','uint32_t getCurrentTime()override{return now;}')
+  utils=UTILS.replace('class Utils { public:', 'class Utils { public:\n static int parseTextParts(char*,const char*[],int,char);')
+  for name,content in {"Utils.h":utils,"FS.h":FILESYSTEM,"Mesh.h":mesh,"Hardware.h":HARDWARE,
+                       "Dispatcher.h":'#pragma once\n#include <Mesh.h>\n',
+                       "Packet.h":'#pragma once\n#include <Mesh.h>\n',
+                       "Arduino.h":'#pragma once\n#include <Mesh.h>\n#include <cstdlib>\n'}.items():
+   (work/name).write_text(content)
+  # Preserve production transport-key MAC algorithms with host Crypto void* parity.
+  sha=ROOT/"test/mocks/SHA256.h"
+  (work/"SHA256.h").write_text(f'''#pragma once\n#define SHA256 SHA256Base\n#include "{sha}"\n#undef SHA256
+class SHA256:public SHA256Base{{public:
+void finalize(void* out,size_t n){{SHA256Base::finalize(static_cast<uint8_t*>(out),n);}}
+void finalizeHMAC(const uint8_t* key,size_t size,void* out,size_t n){{SHA256Base::finalizeHMAC(key,size,static_cast<uint8_t*>(out),n);}}
+}};''')
+  transport=(ROOT/"src/helpers/TransportKeyStore.cpp").read_text()
+  methods="\n".join(body(transport,sig) for sig in ["uint16_t TransportKey::calcTransportCode(","bool TransportKey::isNull("])
+  prefix=RUNTIME_HARNESS[:RUNTIME_HARNESS.index('static unsigned apply(')]
+  source=(ROOT/"examples/simple_repeater/MyMesh.cpp").read_text()
+  primary=PRIMARY_HARNESS[:PRIMARY_HARNESS.index('static constexpr uint32_t HOUR')]
+  primary=primary.replace('static uint32_t now_ms = 1;\nuint32_t millis() { return now_ms; }','#define now_ms fake_ms')
+  primary=primary.replace('struct Packet {};','').replace('class MyMesh {','class MyMesh :public FleetReplyHooks {')
+  primary=primary.replace('struct Radio { uint32_t last_rx = 0; uint32_t getLastRecvMillis() { return last_rx; } };',
+   'struct PrimaryPHY { uint32_t last_rx=0; mesh::RadioProfiles state; uint32_t getLastRecvMillis(){return last_rx;} mesh::RadioProfiles* profiles(){return &state;} };')
+  primary=primary.replace('Radio radio; Radio* _radio','PrimaryPHY radio; PrimaryPHY* _radio')
+  primary=primary.replace('  MyMesh() {',r'''
+  ScheduledRadioSetting fleet_schedule_setting_{};
+  uint8_t fleet_schedule_delete_mask_[(MAX_SCHEDULED_RADIO_SETTINGS+7)/8]{};
+  int fleet_schedule_slot_=-1;
+  bool fleet_command_=false,fleet_schedule_pending_=false,fleet_schedule_delivered_=false,fleet_schedule_delete_all_temp_=false;
+  void beginFleetCommand()override{fleet_command_=true;}
+  void endFleetCommand()override{fleet_command_=false;}
+  bool hasFleetReplyMutation()const override{return fleet_schedule_pending_;}
+  void finishFleetReplyMutation(bool)override;
+  void serviceFleetScheduleReply();
+  int findScheduledRadioSettingByIndex(bool,int)const;
+  void deleteScheduledRadioParams(bool,const char*,char*);
+  void formatScheduledRadioParams(bool,const char*,char*);
+  void formatScheduledRadioSetting(char*,int,int)const;
+  void formatRadioParamTuple(char*,size_t,const ScheduledRadioSetting&)const;
+  MyMesh() {''')
+  signatures=[
+   'static const char* skipLocalSpaces(', 'static bool selectorIsEmpty(',
+   'static bool selectorIsAll(', 'static bool parsePositiveSelector(', 'static void formatFixed3(',
+   'void MyMesh::checkRxInactivityWatchdog()', 'void MyMesh::setTempRadioTiming(',
+   'void MyMesh::updateAdvertTimer()', 'void MyMesh::updateFloodAdvertTimer()',
+   'void MyMesh::queueSavedRadioApply()', 'bool MyMesh::applySavedRadioParams()',
+   'void MyMesh::refreshScheduledRadioState()', 'void MyMesh::processScheduledRadioSettings()',
+   'void MyMesh::applyTempRadioParams(', 'bool MyMesh::scheduleNormalRadio()',
+   'void MyMesh::clearScheduledRadioSetting(', 'int MyMesh::findFreeScheduledRadioSlot() const',
+   'int MyMesh::countScheduledRadioSettings(', 'bool MyMesh::scheduledRadioConflicts(',
+   'void MyMesh::addScheduledRadioParams(', 'int MyMesh::findScheduledRadioSettingByIndex(',
+   'void MyMesh::deleteScheduledRadioParams(', 'void MyMesh::formatScheduledRadioParams(',
+   'void MyMesh::formatScheduledRadioSetting(', 'void MyMesh::formatRadioParamTuple(',
+   'void MyMesh::finishFleetReplyMutation(', 'void MyMesh::serviceFleetScheduleReply(',
+  ]
+  primary=primary.replace('@METHODS@','\n'.join(body(source,sig) for sig in signatures))
+  cli=(ROOT/'src/helpers/CommonCLI.cpp').read_text()
+  parsers='\n'.join(body(cli,sig) for sig in [
+   'static bool looksUnsignedInteger(', 'static const char* skipSpacesConst(',
+   'static bool parseUint32Strict(', 'static bool bwMatches(', 'static bool isValidLoRaBandwidth(',
+   'static int countSeparatedParts(', 'static bool parseScheduledRadioArgs(',
+  ])
+  parts=body((ROOT/'src/Utils.cpp').read_text(),'int Utils::parseTextParts(')
+  formatting='''\nstruct StrHelper {
+static void strncpy(char* dst,const char* src,size_t n){if(n){size_t len=strlen(src);if(len>=n)len=n-1;memcpy(dst,src,len);dst[len]=0;}}
+static const char* ftoa3(float value){static char text[32];snprintf(text,sizeof(text),"%.3f",double(value));return text;}
+};
+uint16_t rxPowerSavingPreambleForParams(uint8_t,float){return 32;}
+'''
+  parsing='\n#include <helpers/CLICommandUtils.h>\nnamespace mesh {\n'+parts+'\n}\n'+parsers
+  start=cli.index('  } else if (memcmp(config, "radioat ", 8) == 0)')
+  end=cli.index('  } else if (memcmp(config, "lat ", 4) == 0)',start)
+  primary_extra=PRIMARY_EXTRA.replace('@PRIMARY_SET@',cli[start:end])
+  physical,secondary=HARNESS.split('struct Fixture {',1)
+  secondary=('struct Fixture {'+secondary).replace(' assert(argc==2);const std::string scenario=argv[1];Fixture f;',
+   ' assert(argc==2);if(!strncmp(argv[1],"primary_",8))return primary(argc,argv);const std::string scenario=argv[1];Fixture f;')
+  fixture=work/"fixture.cpp";fixture.write_text('#define MESH_ENABLE_FLEET_CONTROL 1\n'+
+   prefix.replace("@TRANSPORT_METHODS@",methods)+physical+formatting+primary+parsing+primary_extra+secondary)
+  cls.binary=work/"schedule-test"
+  sanitizer=["-fsanitize=address,undefined","-fno-omit-frame-pointer","-no-pie"] if os.environ.get("MESHCORE_FLEET_SANITIZERS")=="1" else []
+  built=subprocess.run([compiler,"-std=c++17","-O2","-g","-DRP2040_PLATFORM",*sanitizer,
+    "-Wall","-Wextra","-Werror","-Wno-unused-function","-Wno-unused-parameter","-Wno-sign-compare","-Wno-misleading-indentation",
+    *(["-Wno-format-truncation"] if "g++" in compiler else []),
+    "-I",str(work),"-isystem",str(ROOT/"test/mocks"),"-I",str(ROOT/"src"),str(fixture),
+    str(ROOT/"src/helpers/FleetCommand.cpp"),str(ROOT/"src/helpers/FleetChannel.cpp"),
+    str(ROOT/"src/helpers/RadioProfileCLI.cpp"),"-lcrypto","-o",str(cls.binary)],capture_output=True,text=True,timeout=60)
+  if built.returncode:cls.work.cleanup();raise AssertionError(built.stdout+built.stderr)
+ @classmethod
+ def tearDownClass(cls):cls.work.cleanup()
+ def scenario(self,name):
+  result=subprocess.run([str(self.binary),name],capture_output=True,text=True,timeout=10)
+  self.assertEqual(result.returncode,0,result.stdout+result.stderr)
+ def test_permanent_schedule_waits_for_ack_then_persists_and_rejects_replay(self):self.scenario("permanent_ack")
+ def test_temporary_schedule_waits_for_ack_and_restores_at_original_end(self):self.scenario("temporary_ack")
+ def test_failed_ack_does_not_add_any_schedule(self):self.scenario("ack_fail")
+ def test_failed_queue_does_not_add_a_schedule(self):self.scenario("queue_fail")
+ def test_secondary_ack_timeout_removes_packet_and_cancels_schedule(self):self.scenario("ack_timeout")
+ def test_secondary_schedule_waits_for_both_physical_ack_copies(self):self.scenario("both_copies")
+ def test_delayed_ack_cannot_resurrect_an_expired_temporary_lease(self):self.scenario("expired_lease")
+ def test_due_permanent_schedule_keeps_old_radio_until_atomic_save_succeeds(self):self.scenario("save_fail")
+ def test_delete_active_or_future_schedule_is_delayed_until_ack_and_cancelled_on_failure(self):self.scenario("delete_ack")
+ def test_invalid_mac_never_reaches_actual_scheduler(self):self.scenario("unsigned_no_schedule")
+ def test_valid_mac_invalid_signature_cannot_reserve_or_schedule(self):self.scenario("bad_signature_no_schedule")
+ def test_backward_clock_correction_cannot_extend_temporary_lease(self):self.scenario("clock_backwards")
+ def test_get_all_and_individual_slot_reply_keeps_schedule_and_needs_no_barrier(self):self.scenario("get_slots")
+ def test_primary_permanent_schedule_runs_only_after_ack_and_deferred_loop(self):self.scenario("primary_permanent")
+ def test_primary_waits_for_both_physical_ack_copies_and_keeps_original_lease_end(self):self.scenario("primary_both_copies")
+ def test_primary_queue_failure_cancels_staged_schedule(self):self.scenario("primary_queue_failure")
+ def test_primary_failed_tx_cancels_staged_schedule(self):self.scenario("primary_tx_failure")
+ def test_primary_ack_timeout_removes_packet_and_cancels_schedule(self):self.scenario("primary_ack_timeout")
+ def test_primary_expired_window_is_discarded_after_late_ack(self):self.scenario("primary_expired_window")
+ def test_primary_local_schedule_edits_are_busy_until_fleet_ack_finishes(self):self.scenario("primary_local_busy")
+ def test_primary_active_delete_waits_for_ack_and_deferred_loop(self):self.scenario("primary_active_delete")
+ def test_primary_older_lease_expires_even_while_fleet_ack_is_pending(self):self.scenario("primary_expiry_during_pending")
+ def test_primary_permanent_radio_keeps_old_tuple_when_save_fails(self):self.scenario("primary_save_failure")
+ def test_primary_get_list_and_slot_do_not_mutate_schedule(self):self.scenario("primary_get_slots")
+ def test_primary_signed_handler_never_runs_for_forged_signature(self):self.scenario("primary_bad_signature")
+ def test_primary_direct_recovery_is_not_deleted_by_an_older_fleet_ack(self):self.scenario("primary_recovery_supersedes_delete")
+ def test_primary_return_to_saved_radio_cancels_older_pending_fleet_schedule(self):self.scenario("primary_recovery_supersedes_add")
+ def test_canceled_secondary_fleet_ack_does_not_release_new_direct_admin_mutation(self):self.scenario("replaced_secondary_ack")
+ def test_canceled_secondary_fleet_failure_does_not_cancel_new_direct_admin_mutation(self):self.scenario("replaced_secondary_fail")
+ def test_canceled_secondary_fleet_timeout_does_not_cancel_new_direct_admin_mutation(self):self.scenario("replaced_secondary_timeout")
+ def test_primary_only_fleet_ack_does_not_release_new_direct_secondary_mutation(self):self.scenario("primary_ack_ignores_secondary")
+ def test_primary_only_fleet_failure_does_not_cancel_new_direct_secondary_mutation(self):self.scenario("primary_fail_ignores_secondary")
+ def test_primary_only_fleet_timeout_does_not_cancel_new_direct_secondary_mutation(self):self.scenario("primary_timeout_ignores_secondary")
+
+if __name__=="__main__":unittest.main()

test/test_fleet_role_wiring.py

@@ -222,7 +222,7 @@ AUTHORIZATION = r'''
 static unsigned configurations = 0, allocations = 0;
 namespace mesh {
 struct FleetChannel {
-  explicit FleetChannel(void*) { ++allocations; }
+  explicit FleetChannel(void*, void* = nullptr) { ++allocations; }
   bool handleConfig(const char*, char* reply, size_t capacity) {
     assert(capacity == 160); ++configurations; strcpy(reply, "OK"); return true;
   }

test/test_radio_profile_cli.py

@@ -12,18 +12,22 @@ class RadioProfileCLITest(unittest.TestCase):
         compiler = shutil.which('g++') or shutil.which('clang++')
         self.assertIsNotNone(compiler)
         with tempfile.TemporaryDirectory() as work:
-            exe = Path(work) / 'profiles.exe'
-            subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra',
-                '-I', str(ROOT/'test/fixtures/radio_profiles/mocks'),
-                '-I', str(ROOT/'test/mocks'), '-I', str(ROOT/'src'),
-                str(ROOT/'src/helpers/RadioProfileCLI.cpp'),
-                str(ROOT/'test/fixtures/radio_profiles/cli_test.cpp'),
-                '-o', str(exe)], check=True, capture_output=True, text=True)
-            for scenario in (None, 'permanent_order', 'temporary_boundary', 'lease_creation',
-                             'save_failure', 'schedule_permutations', 'storage_isolation',
-                             'relative_schedule'):
-                with self.subTest(scenario=scenario or 'commands_persistence_and_timers'):
-                    subprocess.run([str(exe)] + ([scenario] if scenario else []), check=True)
+            # Compact STM32 images omit fleet staging. Both paths must preserve
+            # local schedules, persistence retries and temporary lease bounds.
+            for fleet_enabled in (False, True):
+                exe = Path(work) / f'profiles-{int(fleet_enabled)}.exe'
+                subprocess.run([compiler, '-std=c++17', '-Wall', '-Wextra',
+                    f'-DMESH_ENABLE_FLEET_CONTROL={int(fleet_enabled)}',
+                    '-I', str(ROOT/'test/fixtures/radio_profiles/mocks'),
+                    '-I', str(ROOT/'test/mocks'), '-I', str(ROOT/'src'),
+                    str(ROOT/'src/helpers/RadioProfileCLI.cpp'),
+                    str(ROOT/'test/fixtures/radio_profiles/cli_test.cpp'),
+                    '-o', str(exe)], check=True, capture_output=True, text=True)
+                for scenario in (None, 'permanent_order', 'temporary_boundary', 'lease_creation',
+                                 'save_failure', 'schedule_permutations', 'storage_isolation',
+                                 'relative_schedule'):
+                    with self.subTest(fleet=fleet_enabled, scenario=scenario or 'commands_persistence_and_timers'):
+                        subprocess.run([str(exe)] + ([scenario] if scenario else []), check=True)
 
 if __name__ == '__main__':
     unittest.main()