Document and test 1.17.1.5 USB logging sleep workaround

test/README.md

@@ -30,6 +30,7 @@ python3 test/test_esp32_full_partition.py       # Full partition-preservation po
 python3 test/test_esp32_dram.py                 # Classic ESP32 static limits and cached-image build gates
 python3 test/test_esp32_usb_serial_hygiene.py   # Single-TTY diagnostics/NVS contract
 python3 test/test_esp32_usb_sleep.py            # USB/logging sleep guards, G3 button wake and two-minute hold/rollover
+python3 test/test_logging_sleep_contract.py    # USB enable examples and real Repeater/Room MQTT sleep gates
 python3 test/test_esp32_tinyusb_role_hygiene.py # G2/room USB write coverage and bounded-list contracts
 python3 test/test_esp32_tinyusb_cooperative_output.py # Real role dump/list pumps with host C++ stubs
 python3 test/test_esp32_tinyusb_nonblocking.py # Native CDC stalled-host/64-byte-FIFO simulation (C++17 compiler)

test/test_firmware_picker.js

@@ -877,6 +877,7 @@ const controlledReleases = [release(liveFamily, '2026-09-01T00:00:00Z', [
   asset('heltec_v4_repeater_observer_mqtt-full-usb-wifi-ota-' + liveFamily + '.bin'),
 ]), release('repeater-room-' + liveFamily, '2026-09-01T00:00:00Z', [
   asset('RAK_4631_repeater-' + liveFamily + '.uf2'),
+  asset('Station_G3_ESP32_repeater-' + liveFamily + '.bin'),
 ])];
 const controlled = picker.buildCatalog(controlledReleases, controls);
 const findControlled = name => controlled.profiles.find(p => p.target === name);
@@ -885,10 +886,24 @@ const observer = findControlled('heltec_v4_repeater_observer_mqtt-full-usb-wifi'
 for (const mode of ['none', 'usb', 'wifi', 'both']) {
   const directions = picker.runtimeDirections(observer, {logging: mode});
   assert.deepStrictEqual(directions[0].actions[0].commands,
-    ['set logging.output ' + (mode === 'none' ? 'off' : mode), 'get logging.output']);
+    (mode === 'usb' || mode === 'both' ? ['powersaving off'] : []).concat(
+      ['set logging.output ' + (mode === 'none' ? 'off' : mode), 'get logging.output']));
 }
 const mqttCompanion = findControlled('heltec_v4_2_v4_3_companion_radio_full_femon');
 assert.deepStrictEqual(mqttCompanion.loggingModes, ['none', 'usb', 'wifi', 'both']);
+for (const mode of ['usb', 'both']) {
+  assert.deepStrictEqual(picker.runtimeDirections(mqttCompanion, {logging: mode})[0].actions[0].commands,
+    ['powersaving off', 'set usb.logging on']);
+}
+assert(picker.installSteps(mqttCompanion, 'bin').some(step => step.includes('powersaving off')));
+// The workaround belongs to the affected release, not every future ESP32 build.
+for (const releaseFamily of ['v1.17.1.4-old', 'v1.17.1.5-halo-keymind-cascade-dev-newfix',
+                             'v1.17.1.6-next', 'v1.17.1.50-next', '']) {
+  const otherRelease = {...mqttCompanion, releaseFamily};
+  assert.deepStrictEqual(picker.runtimeDirections(otherRelease, {logging: 'usb'})[0].actions[0].commands,
+    ['set usb.logging on']);
+  assert(!picker.installSteps(otherRelease, 'bin').some(step => step.includes('powersaving off')));
+}
 const companionWifi = picker.runtimeDirections(mqttCompanion, {logging: 'wifi'});
 assert.deepStrictEqual(companionWifi[0].actions[0].commands, ['set usb.logging off']);
 assert(companionWifi[0].actions[0].text.includes('enable the desired MQTT'));
@@ -900,6 +915,11 @@ const nrf = findControlled('RAK_4631_companion_radio_full');
 assert.deepStrictEqual(picker.runtimeDirections(nrf, {logging: 'usb'})[0].actions[0].commands,
   ['set usb.logging on reboot']);
 const rakRepeater = findControlled('RAK_4631_repeater');
+const g3Repeater = findControlled('Station_G3_ESP32_repeater');
+assert.deepStrictEqual(picker.runtimeDirections(g3Repeater, {logging: 'usb'})[0].actions[0].commands,
+  ['powersaving off', 'set usb.logging on']);
+assert.deepStrictEqual(picker.runtimeDirections(g3Repeater, {logging: 'none'})[0].actions[0].commands,
+  ['set usb.logging off']);
 const rakCommands = picker.runtimeDirections(rakRepeater, {logging: 'usb', mode: 'standard'});
 assert.deepStrictEqual(rakCommands[0].actions[0].commands, ['set usb.logging on']);
 assert.strictEqual(rakCommands.find(s => s.title.startsWith('RS232')).actions[0].label, 'Off');

test/test_logging_sleep_contract.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""Check USB workaround instructions and execute the real MQTT sleep gates."""
+
+from pathlib import Path
+import os
+import re
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def method(path, signature):
+    text = (ROOT / path).read_text()
+    start = text.index(signature)
+    end = text.index("{", start) + 1
+    depth = 1
+    while depth:
+        depth += (text[end] == "{") - (text[end] == "}")
+        end += 1
+    return text[start:end]
+
+
+HARNESS = r'''
+#include <cstdint>
+#include <stdexcept>
+#include <iostream>
+#define WITH_BRIDGE 1
+struct AbstractBridge { virtual bool isRunning() const = 0; };
+struct MQTTBridge : AbstractBridge {
+  bool _initialized = false;
+  @RUNNING@
+};
+struct Radio {
+  bool isWatchdogObserving() const { return false; }
+  bool isCalibratingNoiseFloor() const { return false; }
+};
+struct MyMesh {
+  MQTTBridge* bridge = nullptr;
+  const AbstractBridge* activeBridge() const { return bridge; }
+  struct { bool pending = false; } deferred_cli_command;
+  struct { bool hasActiveUserGpioTimer() const { return false; } } _cli;
+  struct { int getNumClients() const { return 0; } } acl;
+  struct { bool battery_alert_enabled = false; } _prefs;
+  Radio radio_driver;
+  bool pending_self_advert = false;
+  bool saved_radio_apply_pending = false, temp_radio_applied = false;
+  unsigned long next_flood_advert = 0, next_local_advert = 0;
+  unsigned long dirty_contacts_expiry = 0, next_recent_repeater_sweep = 0;
+  unsigned long next_battery_alert_check = 0, next_push = 0;
+  unsigned long radio_apply_retry_at = 0, set_radio_at = 0, revert_radio_at = 0;
+  bool hasPendingOtaApply() const { return false; }
+  bool hasQueuedWorkDue() const { return false; }
+  bool hasRetryWorkDue() const { return false; }
+  bool hasScheduledRadioWorkDue() const { return false; }
+  bool isMillisTimerDue(unsigned long) const { return false; }
+  bool millisHasNowPassed(unsigned long) const { return false; }
+  bool getNextQueueWakeDelay(uint32_t&) const { return false; }
+  bool getNextRetryWakeDelay(uint32_t&) const { return false; }
+  uint32_t limitSleepToMillisTimer(unsigned long, uint32_t secs) const { return secs; }
+  uint32_t limitSleepToScheduledRadioWork(uint32_t secs) const { return secs; }
+  bool hasPendingWork() const;
+  uint32_t getPowerSaveSleepSeconds(uint32_t) const;
+};
+@METHODS@
+static void require(bool ok, const char* why) {
+  if (!ok) throw std::runtime_error(why);
+}
+int main() {
+  try {
+    MyMesh node;
+    require(node.getPowerSaveSleepSeconds(30) == 30, "quiet node cannot sleep");
+    MQTTBridge logger;
+    node.bridge = &logger;
+    require(node.getPowerSaveSleepSeconds(30) == 30, "stopped MQTT blocked sleep");
+    logger._initialized = true;
+    require(node.hasPendingWork(), "running MQTT lost its sleep blocker");
+    require(node.getPowerSaveSleepSeconds(30) == 0,
+            "WiFi-only logging can enter device sleep without a USB guard");
+    logger._initialized = false;
+    require(node.getPowerSaveSleepSeconds(30) == 30,
+            "stopping MQTT did not release its sleep blocker");
+    require(node.getPowerSaveSleepSeconds(0) == 0, "zero sleep limit changed");
+  } catch (const std::exception& e) {
+    std::cerr << e.what() << '\n';
+    return 1;
+  }
+}
+'''
+
+
+class LoggingSleepContractTest(unittest.TestCase):
+    def test_usb_enable_examples_include_workaround_first(self):
+        for path in (
+            "docs/cli_commands.md", "docs/role_feature_switches.md",
+            "docs/full_companion_features.md", "docs/companion_radio_full.md",
+            "docs/firmware_picker.md", "docs/releases/1.17.1.5.md",
+        ):
+            with self.subTest(path=path):
+                text = (ROOT / path).read_text()
+                blocks = re.findall(r"```text\n(.*?)```", text, re.S)
+                self.assertTrue(any("powersaving off\nset usb.logging on\n" in block
+                                    for block in blocks), path)
+                self.assertNotRegex(text, r"powersaving off\nset usb.logging on reboot")
+
+    def test_wifi_exception_is_documented_with_running_state_check(self):
+        for path in ("docs/cli_commands.md", "docs/role_feature_switches.md",
+                     "docs/releases/1.17.1.5.md"):
+            with self.subTest(path=path):
+                text = " ".join((ROOT / path).read_text().split())
+                self.assertRegex(text, r"WiFi/MQTT-only logging.*?does not need")
+                self.assertIn("get bridge.running", text)
+
+    def test_running_mqtt_prevents_sleep_in_repeater_and_room(self):
+        running = method("src/helpers/bridges/MQTTBridge.h",
+                         "bool isRunning() const override")
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-sleep-") as temp:
+            for role in ("simple_repeater", "simple_room_server"):
+                with self.subTest(role=role):
+                    source = f"examples/{role}/MyMesh.cpp"
+                    methods = "\n".join(method(source, signature) for signature in (
+                        "bool MyMesh::hasPendingWork() const",
+                        "uint32_t MyMesh::getPowerSaveSleepSeconds(uint32_t max_secs) const",
+                    ))
+                    cpp = Path(temp) / f"{role}.cpp"
+                    binary = Path(temp) / role
+                    cpp.write_text(HARNESS.replace("@RUNNING@", running)
+                                   .replace("@METHODS@", methods))
+                    built = subprocess.run([
+                        os.environ.get("CXX", "c++"), "-std=c++17", "-Wall", "-Wextra",
+                        str(cpp), "-o", str(binary),
+                    ], capture_output=True, text=True)
+                    self.assertEqual(built.returncode, 0, built.stderr)
+                    result = subprocess.run([str(binary)], capture_output=True, text=True)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()