Keep nRF52 GPS discovery from blocking USB setup

src/helpers/sensors/EnvironmentSensorManager.cpp

@@ -348,6 +348,10 @@ static bool serialHasValidGpsSentence(Stream& serial, uint32_t timeout_ms) {
   const uint32_t started = millis();
   do {
     while (serial.available()) {
+      // Continuous non-NMEA input must not bypass the existing probe window.
+      // Sample inside the byte loop as well as after an empty FIFO; unsigned
+      // elapsed time preserves the deadline across millisecond rollover.
+      if (static_cast<uint32_t>(millis() - started) >= timeout_ms) return false;
       if (probe.ingest(static_cast<uint8_t>(serial.read()))) return true;
     }
     delay(5);
@@ -1113,11 +1117,11 @@ const char* EnvironmentSensorManager::getSettingValue(int i) const {
 bool EnvironmentSensorManager::setSettingValue(const char* name, const char* value) {
   #if ENV_INCLUDE_GPS
   bool gps_setting_available = gps_detected;
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
   gps_setting_available = gps_setting_available || gps_discovery_pending;
 #endif
   if (gps_setting_available && strcmp(name, "gps") == 0) {
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
     // Preferences load before a cold receiver finishes its bounded probe.
     // Retain intent so late discovery cannot replace a saved off/on choice.
     gps_discovery_preference_known = true;
@@ -1181,7 +1185,7 @@ void EnvironmentSensorManager::setPowerSavingEnabled(bool enabled) {
 }
 
 #if ENV_INCLUDE_GPS
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
 void EnvironmentSensorManager::finishBasicGpsDiscovery(bool found) {
   gps_discovery_pending = false;
   gps_detected = found;
@@ -1238,7 +1242,7 @@ void EnvironmentSensorManager::initBasicGPS() {
   resetGpsTelemetryTransportState();
   gps_serial_transport = false;
   gps_serial_transport_blocked = false;
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
   gps_detected = false;
   gps_active = false;
 #endif
@@ -1259,7 +1263,7 @@ void EnvironmentSensorManager::initBasicGPS() {
     MESH_DEBUG_PRINTLN("No GPS wake/reset pin found for this board. Continuing on...");
   #endif
 
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
   // Reserve the configured UART immediately, including while presence is
   // still unknown. Continue discovery from loop() so an absent optional GPS
   // cannot consume the website's entire five-second USB bootstrap deadline.
@@ -1540,7 +1544,7 @@ void EnvironmentSensorManager::stop_gps() {
 
 bool EnvironmentSensorManager::gpsUsesSerialUart(uint8_t uart) const {
 #if ENV_INCLUDE_GPS
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
   // The pending probe owns this UART before presence has been established.
   return uart == 1 && gps_serial_transport;
 #else
@@ -1603,14 +1607,14 @@ bool EnvironmentSensorManager::setGpsSerialTransportBlocked(uint8_t uart,
     // Cancel remote-query acquisition/hold state before releasing the UART.
     setGpsTelemetryTransportAvailable(false);
     gps_serial_transport_blocked = true;
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
     if (gps_discovery_pending) _location->stop();
 #endif
     if (gps_active) stop_gps();
     Serial1.end();
   } else {
     gps_serial_transport_blocked = false;
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
     if (gps_discovery_pending) {
       // The bridge's time cannot count as a GPS absence timeout. Reclaim only
       // after release and give the newly awakened receiver its full window.
@@ -1637,7 +1641,7 @@ void EnvironmentSensorManager::loop() {
   #if ENV_INCLUDE_GPS
   static unsigned long next_gps_update = 0;
   unsigned long now = millis();
-#if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+#if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && !defined(RAK_WISBLOCK_GPS)
   serviceBasicGpsDiscovery();
 #endif
   loopGpsTelemetry(now);

src/helpers/sensors/EnvironmentSensorManager.h

@@ -26,7 +26,9 @@ protected:
   bool     gps_serial_transport_blocked = false;
 
   #if ENV_INCLUDE_GPS
-  #if defined(ESP32_PLATFORM) && !defined(RAK_WISBLOCK_GPS)
+  #if (defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) \
+    && !defined(RAK_WISBLOCK_GPS) \
+    && !(defined(RAK_BOARD) && !defined(RAK_WISMESH_TAG))
   bool gps_discovery_pending = false;
   bool gps_discovery_preference_known = false;
   uint32_t gps_discovery_started_at = 0;

test/test_esp32_gps_discovery.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Execute cooperative ESP32 GPS discovery with real preference/UART policy.
+"""Execute cooperative ESP32/nRF52 GPS discovery with preference/UART policy.
 
 Only the UART, clock and GPS hardware are peripheral doubles. Discovery,
 completion, loop servicing, settings, power transitions and UART handoff come
@@ -7,7 +7,6 @@ from EnvironmentSensorManager.cpp; SensorManager.cpp and LocationProvider's
 power/timing policy compile unchanged. This does not qualify physical GPS/USB.
 """
 from pathlib import Path
-import re
 import shutil
 import subprocess
 import sys
@@ -73,8 +72,6 @@ struct EnvironmentSensorManager : SensorManager {
   bool userEnabled() const { return isGpsTelemetryUserEnabled(); }
   bool transportAvailable() const { return isGpsTelemetryTransportAvailable(); }
   void initBasicGPS();
-  void finishBasicGpsDiscovery(bool found);
-  void serviceBasicGpsDiscovery();
   void armGpsPowerSavingCycle();
   void start_gps();
   void stop_gps();
@@ -263,15 +260,14 @@ int main(int argc, char** argv) {
 def harness(*, negative=None):
     source = SOURCE.read_text(encoding="utf-8")
     header = HEADER.read_text(encoding="utf-8")
-    names = ("gps_detected", "gps_active", "gps_serial_transport",
-             "gps_serial_transport_blocked", "gps_discovery_pending",
-             "gps_discovery_preference_known", "gps_discovery_started_at")
-    fields = "\n".join(re.search(r"^\s*(?:bool|uint32_t)\s+" + name + r"\s*=[^;]*;",
-                                 header, re.MULTILINE)[0] for name in names)
+    # Preserve the actual header's platform/provider guards and declarations.
+    # Extracting each field without its guard could invent nRF52 state that
+    # the production header never made available to that platform.
+    fields = header[header.index("  bool     gps_detected"):
+                    header.index("  LocationProvider* _location")]
+    fields += "\n#endif // ENV_INCLUDE_GPS\n"
     signatures = (
         "void EnvironmentSensorManager::initBasicGPS()",
-        "void EnvironmentSensorManager::finishBasicGpsDiscovery(bool found)",
-        "void EnvironmentSensorManager::serviceBasicGpsDiscovery()",
         "void EnvironmentSensorManager::armGpsPowerSavingCycle()",
         "void EnvironmentSensorManager::start_gps()",
         "void EnvironmentSensorManager::stop_gps()",
@@ -283,7 +279,15 @@ def harness(*, negative=None):
         "bool EnvironmentSensorManager::setGpsSerialTransportBlocked(",
         "void EnvironmentSensorManager::loop()",
     )
-    methods = "\n".join(extract_braced(source, signature) for signature in signatures)
+    # These two methods live under the production platform/provider guard.
+    # Compile that guard unchanged as well, instead of calling an extracted
+    # unconditional definition that would mask a missing nRF52 implementation.
+    begin = source.rindex("#if ENV_INCLUDE_GPS\n", 0, source.index(
+        "void EnvironmentSensorManager::finishBasicGpsDiscovery(bool found)"))
+    end = source.index("void EnvironmentSensorManager::initBasicGPS()", begin)
+    guarded_discovery = source[begin:end] + "\n#endif // ENV_INCLUDE_GPS\n"
+    methods = guarded_discovery + "\n" + "\n".join(
+        extract_braced(source, signature) for signature in signatures)
     mutations = {
         "blocking": ("  gps_discovery_started_at = millis();",
                      "  gps_discovery_started_at = millis();\n  delay(5000);"),
@@ -301,14 +305,15 @@ def harness(*, negative=None):
     return HARNESS.replace("@FIELDS@", fields).replace("@METHODS@", methods)
 
 
-class Esp32GpsDiscoveryTests(unittest.TestCase):
+class CooperativeGpsDiscoveryTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.compiler = shutil.which("g++") or shutil.which("clang++")
         if cls.compiler is None:
             raise unittest.SkipTest("a host C++17 compiler is required")
 
-    def compile_run(self, cases, *, defines=(), negative=None):
+    def compile_run(self, cases, *, defines=(), negative=None,
+                    platform="ESP32_PLATFORM"):
         with tempfile.TemporaryDirectory(prefix="meshcore-esp32-gps-") as directory:
             work = Path(directory)
             for name, text in {
@@ -319,7 +324,7 @@ class Esp32GpsDiscoveryTests(unittest.TestCase):
             cpp, binary = work / "discovery.cpp", work / "discovery"
             cpp.write_text(harness(negative=negative), encoding="utf-8")
             command = [self.compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
-                       "-Wno-unused-parameter", "-DENV_INCLUDE_GPS=1", "-DESP32_PLATFORM=1",
+                       "-Wno-unused-parameter", "-DENV_INCLUDE_GPS=1", "-D" + platform,
                        *["-D" + define for define in defines],
                        *(["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
                           "-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else []),
@@ -328,7 +333,8 @@ class Esp32GpsDiscoveryTests(unittest.TestCase):
             compiled = subprocess.run(command, capture_output=True, text=True, timeout=60)
             self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
             for case in cases:
-                with self.subTest(case=case, defines=defines, negative=negative):
+                with self.subTest(case=case, defines=defines, negative=negative,
+                                  platform=platform):
                     result = subprocess.run([str(binary), case], capture_output=True,
                                             text=True, timeout=10)
                     if negative:
@@ -363,6 +369,48 @@ class Esp32GpsDiscoveryTests(unittest.TestCase):
     def test_loaded_preference_must_be_retained_negative_control(self):
         self.compile_run(("latest",), defines=("PERSISTANT_GPS=1",), negative="preference_loss")
 
+    def test_nrf52_basic_gps_uses_production_guarded_state_and_methods(self):
+        for defines in (("GPS_BAUD_RATE=38400",), ("PERSISTANT_GPS=1",),
+                        ("RAK_BOARD=1", "RAK_WISMESH_TAG=1")):
+            self.compile_run(("absent", "late", "latest", "awake", "wrap", "blocked", "no_enable"),
+                             defines=defines, platform="NRF52_PLATFORM")
+        for defines in (("ENV_SKIP_GPS_DETECT=1",),
+                        ("ENV_SKIP_GPS_DETECT=1", "PERSISTANT_GPS=1")):
+            self.compile_run(("awake",), defines=defines, platform="NRF52_PLATFORM")
+
+    def test_nrf52_guards_preserve_distinct_wisblock_provider(self):
+        # Unlike WisMesh Tag, ordinary WisBlock must not acquire basic GPS
+        # discovery members. Preprocess the real header; the complete RAK
+        # provider has its own behavior and tests.
+        header = HEADER.read_text(encoding="ascii")
+        with tempfile.TemporaryDirectory(prefix="meshcore-nrf52-gps-header-") as directory:
+            work = Path(directory)
+            for relative in ("Mesh.h", "helpers/SensorManager.h",
+                             "helpers/sensors/LocationProvider.h"):
+                path = work / relative
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_text("\n", encoding="ascii")
+            cpp = work / "header.cpp"
+            cpp.write_text(header, encoding="ascii")
+            for defines, expected in ((("NRF52_PLATFORM",), True),
+                                      (("NRF52_PLATFORM", "RAK_BOARD"), False),
+                                      (("NRF52_PLATFORM", "RAK_BOARD", "RAK_WISMESH_TAG"), True),
+                                      (("RP2040_PLATFORM",), False)):
+                with self.subTest(defines=defines):
+                    result = subprocess.run([self.compiler, "-E", "-P", "-x", "c++",
+                        "-DENV_INCLUDE_GPS=1", *["-D" + value for value in defines],
+                        "-I" + str(work), str(cpp)], text=True, capture_output=True, timeout=10)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual("gps_discovery_pending" in result.stdout, expected)
+
+    def test_nrf52_runtime_negative_controls(self):
+        for cases, negative, defines in ((("absent",), "blocking", ()),
+                                        (("blocked",), "blocked_poll", ()),
+                                        (("blocked",), "cancellation", ()),
+                                        (("latest",), "preference_loss", ("PERSISTANT_GPS=1",))):
+            self.compile_run(cases, negative=negative, defines=defines,
+                             platform="NRF52_PLATFORM")
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

test/test_gps_upstream_adaptations.py

@@ -295,6 +295,73 @@ int main(){
 }
 '''
 
+RAK_UART_PROBE = r'''
+#include <cassert>
+#include <cstring>
+#include <string>
+#include <Arduino.h>
+#include <helpers/sensors/NmeaSentenceProbe.h>
+@TIMEOUT@
+@PROBE@
+struct Uart : Stream {
+ bool continuous_noise=false;
+ uint32_t noise_before=0, reads=0;
+ std::string bytes;
+ int available() override {
+   return continuous_noise || noise_before || !bytes.empty();
+ }
+ int read() override {
+   ++reads; ++now_ms;  // A real peripheral clock advances while bytes arrive.
+   if (continuous_noise) return '!';
+   if (noise_before) { --noise_before; return '!'; }
+   if (bytes.empty()) return -1;
+   const int value=bytes.front(); bytes.erase(0,1); return value;
+ }
+ void validSentence() {
+   const std::string body="GPRMC,123";
+   uint8_t checksum=0;
+   for(uint8_t byte:body) checksum^=byte;
+   char hex[3]; std::snprintf(hex,sizeof(hex),"%02X",checksum);
+   bytes="$"+body+"*"+hex;
+ }
+};
+int main(int argc,char** argv) {
+ assert(argc==2);
+ Uart uart;
+ bool expected=false;
+ if(!strcmp(argv[1],"noise")) uart.continuous_noise=true;
+ else if(!strcmp(argv[1],"wrap_noise")) {
+   now_ms=UINT32_MAX-400U; uart.continuous_noise=true;
+ } else if(!strcmp(argv[1],"valid")) {
+   uart.validSentence(); expected=true;
+ } else if(!strcmp(argv[1],"noisy_valid")) {
+   uart.noise_before=600; uart.validSentence(); expected=true;
+ } else if(!strcmp(argv[1],"wrap_valid")) {
+   now_ms=UINT32_MAX-400U; uart.noise_before=600;
+   uart.validSentence(); expected=true;
+ } else if(!strcmp(argv[1],"invalid_checksum")) {
+   uart.validSentence(); uart.bytes.back()='Z';
+ } else if(!strcmp(argv[1],"too_late")) {
+   uart.noise_before=RAK_UART_GPS_PROBE_TIMEOUT_MS-2; uart.validSentence();
+ } else if(!strcmp(argv[1],"truncated")) {
+   uart.bytes="$GPRMC,123*";
+ } else assert(!strcmp(argv[1],"empty"));
+ const uint32_t started=millis();
+ assert(serialHasValidGpsSentence(uart,RAK_UART_GPS_PROBE_TIMEOUT_MS)==expected);
+ const uint32_t elapsed=millis()-started;
+ if(expected) {
+   assert(elapsed<RAK_UART_GPS_PROBE_TIMEOUT_MS && delay_count==0);
+ } else {
+   assert(elapsed>=RAK_UART_GPS_PROBE_TIMEOUT_MS);
+   assert(elapsed<RAK_UART_GPS_PROBE_TIMEOUT_MS+5);
+ }
+ if(uart.continuous_noise) {
+   assert(elapsed==RAK_UART_GPS_PROBE_TIMEOUT_MS && delay_count==0);
+   assert(uart.reads==RAK_UART_GPS_PROBE_TIMEOUT_MS);
+ }
+}
+'''
+
 
 class GpsUpstreamAdaptationsTest(unittest.TestCase):
     @classmethod
@@ -356,6 +423,47 @@ int main(){SensorManager sensors;assert(!sensors.requestGpsTelemetryTimeSync(UIN
             with self.subTest(defines=defines):
                 self.compile_run(COLD.replace("@INIT@", init), defines)
 
+    def run_rak_uart_probe(self, *, negative=False):
+        source = (ROOT / "src/helpers/sensors/EnvironmentSensorManager.cpp").read_text()
+        timeout = re.search(r"#ifndef RAK_UART_GPS_PROBE_TIMEOUT_MS\s*[\s\S]*?#endif", source).group()
+        probe = extract_braced(source, "static bool serialHasValidGpsSentence(")
+        if negative:
+            # Reproduce the original continuously nonempty FIFO loop. The
+            # process watchdog below bounds this deliberately infinite host
+            # control; it does not change the production's existing deadline.
+            deadline = "      if (static_cast<uint32_t>(millis() - started) >= timeout_ms) return false;"
+            self.assertEqual(probe.count(deadline), 1)
+            probe = probe.replace(deadline, "")
+        with tempfile.TemporaryDirectory(prefix="meshcore-rak-uart-probe-") as directory:
+            work = Path(directory)
+            (work / "Arduino.h").write_text(ARDUINO, encoding="ascii")
+            cpp, binary = work / "probe.cpp", work / "probe"
+            cpp.write_text(RAK_UART_PROBE.replace("@TIMEOUT@", timeout).replace("@PROBE@", probe),
+                           encoding="ascii")
+            result = subprocess.run([self.compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                "-Wno-unused-parameter",
+                *(["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                   "-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else []),
+                "-I", str(work), "-I", str(ROOT / "src"), str(cpp), "-o", str(binary)],
+                capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            if negative:
+                with self.assertRaises(subprocess.TimeoutExpired):
+                    subprocess.run([str(binary), "noise"], capture_output=True, text=True, timeout=1)
+                return
+            for scenario in ("noise", "wrap_noise", "valid", "noisy_valid", "wrap_valid",
+                             "invalid_checksum", "too_late", "truncated", "empty"):
+                with self.subTest(scenario=scenario):
+                    result = subprocess.run([str(binary), scenario], capture_output=True,
+                                            text=True, timeout=10)
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_rak_uart_probe_deadline_survives_continuous_noise_and_rollover(self):
+        self.run_rak_uart_probe()
+
+    def test_rak_original_nonempty_fifo_loop_negative_control(self):
+        self.run_rak_uart_probe(negative=True)
+
     def test_ble_led_optout_preserves_default_and_begin_order(self):
         source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text()
         begin = extract_braced(source, "bool SerialBLEInterface::begin(")

test/test_official_config_gui.py

@@ -67,8 +67,121 @@ int main() {
 '''
 
 
-def cli_source(*, lf_reply=False, reject_cr=False):
-    source = console.firmware_harness(lose_crlf=lf_reply)
+def nrf52_cli_source():
+    # Reuse the hardware boundary from the native nRF52 USB transport tests.
+    # UsbLogging.cpp and its TinyUSB callbacks compile unchanged; only FIFO,
+    # endpoint and virtual time are supplied by the host peripheral fixture.
+    fixture = ROOT / "test/fixtures/nrf52_usb_console/test_console.cpp"
+    source = fixture.read_text().split("static void service()", 1)[0]
+    source = source.replace("static bool mounted = true, dtr = true, dfu = false;",
+                            "static bool mounted = true, dtr = false, dfu = false;")
+    source = source.replace("return 0; }\nvoid tud_cdc_n_read_flush",
+                            "return nrfInputAvailable(); }\nvoid tud_cdc_n_read_flush")
+    source = source.replace("void tud_cdc_n_read_flush(uint8_t n) { assert(n == 0); }",
+                            "void tud_cdc_n_read_flush(uint8_t n) { assert(n == 0); nrf_input.clear(); }")
+    source = source.replace("MockSerial Serial;", r'''
+#include <deque>
+#include <iostream>
+#include <sstream>
+#include <cstdlib>
+#include <cstring>
+static std::deque<uint8_t> nrf_input;
+int nrfInputAvailable() { return nrf_input.size(); }
+int nrfInputRead() {
+  if (nrf_input.empty()) return -1;
+  const int value = nrf_input.front(); nrf_input.pop_front(); return value;
+}
+MockSerial Serial;
+''')
+    pump = body((ROOT / "examples/simple_repeater/main.cpp").read_text(),
+                "static void __attribute__((noinline)) serviceCommandInterfaces()")
+    return source + r'''
+struct MyMesh {
+  void handleUsbCommand(const char*, char* reply) { strcpy(reply, "Unknown"); }
+  void cancelPendingSerialOutput() {}
+  bool hasPendingSerialOutput() { return false; }
+} the_mesh;
+struct Board { void loop() {} } board;
+static constexpr size_t LOCAL_SERIAL_COMMAND_MAX = 159;
+static char command[LOCAL_SERIAL_COMMAND_MAX + 2] = {};
+static bool command_overflow = false;
+''' + pump + r'''
+static void advance() {
+  ++g_mock_millis;
+  serviceCommandInterfaces();
+  mesh::serviceUsbTerminalPort();
+  if (!fifo.empty()) {
+    host += fifo; fifo.clear(); tud_cdc_tx_complete_cb(0);
+  }
+}
+static void emit() {
+  for (unsigned tick = 0; tick < 100; ++tick) advance();
+  if (!host.empty()) {
+    std::cout << "DATA ";
+    for (uint8_t value : host) printf("%02x", value);
+    std::cout << '\n'; host.clear();
+  }
+  std::cout << "DONE\n" << std::flush;
+}
+int main() {
+  static_assert(MESH_NRF52_USB_CONSOLE_COOPERATIVE,
+                "Use nRF52's actual TinyUSB console gate");
+  mesh::setUsbLoggingEnabled(false);
+  mesh::prepareUsbLoggingPort();
+  mesh::beginUsbLoggingPort();
+  std::string line;
+  while (std::getline(std::cin, line)) {
+    if (line.rfind("WRITE ", 0) == 0) {
+      for (size_t offset = 6; offset < line.size(); offset += 2) {
+        unsigned byte; assert(offset + 1 < line.size());
+        assert(sscanf(line.c_str() + offset, "%2x", &byte) == 1);
+        nrf_input.push_back(uint8_t(byte));
+      }
+    } else if (line == "CLOSE") {
+      // The unchanged GUI never requests DTR. The production gate must admit
+      // real RX as reader proof and revoke it at this low-DTR host close.
+      tud_cdc_line_state_cb(0, false, false);
+      mounted = false; tud_umount_cb();
+    } else { assert(false && "Unknown peripheral operation"); }
+    emit();
+    assert(nrf_input.empty());
+  }
+}
+'''
+
+
+def nrf52_headers(work):
+    # Add Print formatting and USB input to the existing peripheral mocks.
+    # These are SDK boundaries, not copies of firmware transport behavior.
+    fixture = ROOT / "test/fixtures/nrf52_usb_console/mocks"
+    stream = (ROOT / "test/mocks/Stream.h").read_text().replace(
+        "#include <string.h>", "#include <string.h>\n#include <cstdarg>\n#include <cstdio>\n#include <cassert>")
+    stream = stream.replace("    virtual void flush()", r'''
+    size_t println(const char* text) { return write(text) + write("\r\n"); }
+    size_t printf(const char* format, ...) {
+        char buffer[256]; va_list args; va_start(args, format);
+        int length = vsnprintf(buffer, sizeof(buffer), format, args); va_end(args);
+        assert(length >= 0 && size_t(length) < sizeof(buffer));
+        return write(reinterpret_cast<const uint8_t*>(buffer), length);
+    }
+    virtual void flush()''')
+    (work / "Stream.h").write_text(stream, encoding="ascii")
+    arduino = (ROOT / "test/mocks/Arduino.h").read_text()
+    mock_serial = (fixture / "Arduino.h").read_text().split("#include <cassert>", 1)[1]
+    mock_serial = mock_serial.replace("return 0;", "return nrfInputAvailable();", 1)
+    mock_serial = mock_serial.replace("int read() override { return -1; }",
+                                      "int read() override { return nrfInputRead(); }")
+    arduino += "\n#include <cassert>\nint nrfInputAvailable();\nint nrfInputRead();\n" + mock_serial
+    (work / "Arduino.h").write_text(arduino, encoding="ascii")
+    return fixture
+
+
+def cli_source(*, lf_reply=False, reject_cr=False, platform="ESP32_PLATFORM"):
+    source = (nrf52_cli_source() if platform == "NRF52_PLATFORM"
+              else console.firmware_harness(lose_crlf=lf_reply))
+    if platform == "NRF52_PLATFORM" and lf_reply:
+        source = source.replace('console.printf("  -> %s\\r\\n", reply);',
+                                'console.printf("  -> %s\\n", reply);')
     common = (ROOT / "src/helpers/CommonCLI.cpp").read_text()
     time_handler = body(common, 'if (memcmp(command, "time ", 5) == 0)')
     support = r'''
@@ -105,7 +218,7 @@ class OfficialConfigGuiTests(unittest.TestCase):
         cls.sources = official_sources()
 
     def run_contract(self, *, blocking=False, lf_reply=False, reject_cr=False,
-                     extra_boot_ms=0):
+                     extra_boot_ms=0, platform="ESP32_PLATFORM"):
         with tempfile.TemporaryDirectory(prefix="official-config-gui-") as directory:
             work = Path(directory)
             gui = self.sources["src/gui.js"].read_text()
@@ -123,7 +236,7 @@ class OfficialConfigGuiTests(unittest.TestCase):
             boot_cpp, boot = work / "gps.cpp", work / "gps"
             boot_cpp.write_text(gps_boot_source(blocking=blocking), encoding="ascii")
             command = [self.compiler, "-std=c++17", "-Wall", "-Wextra", "-Wno-unused-parameter",
-                       "-Wno-unused-function", "-DENV_INCLUDE_GPS=1", "-DESP32_PLATFORM=1",
+                       "-Wno-unused-function", "-DENV_INCLUDE_GPS=1", "-D" + platform,
                        *sanitize, "-I" + str(work), "-I" + str(ROOT / "src"), str(boot_cpp),
                        str(ROOT / "src/helpers/SensorManager.cpp"), "-o", str(boot)]
             built = subprocess.run(command, text=True, capture_output=True, timeout=60)
@@ -133,13 +246,23 @@ class OfficialConfigGuiTests(unittest.TestCase):
             self.assertGreaterEqual(extra_boot_ms, 0)
             boot_ms = int(measured.stdout.strip()) + extra_boot_ms
             cli_cpp, cli = work / "cli.cpp", work / "cli"
-            cli_cpp.write_text(cli_source(lf_reply=lf_reply, reject_cr=reject_cr), encoding="ascii")
+            cli_cpp.write_text(cli_source(lf_reply=lf_reply, reject_cr=reject_cr,
+                                         platform=platform), encoding="ascii")
+            if platform == "NRF52_PLATFORM":
+                peripheral = nrf52_headers(work)
+                cli_flags = ["-DARDUINO", "-DNRF52_PLATFORM", "-DUSE_TINYUSB",
+                             "-DMESH_DEBUG=1", "-I", str(work), "-I", str(peripheral)]
+                cli_sources = [str(ROOT / "src/helpers" / relative) for relative in (
+                    "UsbLogging.cpp", "UsbLoggingClientActivity.cpp", "UsbLoggingLineStateOverride.cpp")]
+            else:
+                cli_flags = ["-DARDUINO_USB_MODE=1", "-DARDUINO_USB_CDC_ON_BOOT=1",
+                             "-DESP32_PLATFORM=1", "-DMESH_ESP32_USB_CONSOLE_COOPERATIVE=1",
+                             "-DMESH_USB_CONSOLE_COOPERATIVE=1", "-DMESH_USB_LOGGING_AVAILABLE=1",
+                             "-DCONFIG_TINYUSB_ENABLED=0"]
+                cli_sources = []
             command = [self.compiler, "-std=c++17", "-Wall", "-Wextra",
-                       "-DARDUINO_USB_MODE=1", "-DARDUINO_USB_CDC_ON_BOOT=1",
-                       "-DESP32_PLATFORM=1", "-DMESH_ESP32_USB_CONSOLE_COOPERATIVE=1",
-                       "-DMESH_USB_CONSOLE_COOPERATIVE=1", "-DMESH_USB_LOGGING_AVAILABLE=1",
-                       "-DCONFIG_TINYUSB_ENABLED=0", *sanitize, "-I", str(ROOT / "src"),
-                       str(cli_cpp), "-o", str(cli)]
+                       *cli_flags, *sanitize, "-I", str(ROOT / "src"),
+                       str(cli_cpp), *cli_sources, "-o", str(cli)]
             built = subprocess.run(command, text=True, capture_output=True, timeout=60)
             self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
             result = subprocess.run([self.node, str(FIXTURE / "contract.mjs"),
@@ -189,6 +312,29 @@ class OfficialConfigGuiTests(unittest.TestCase):
         self.assertNotEqual(result.returncode, 0, "Negative control did not fail")
         self.assertIn("initial time missed the stock 5000 ms deadline", result.stderr)
 
+    def test_nrf52_basic_gps_preserves_stock_gui_initial_time_deadline(self):
+        # Use nRF52's compiled GPS guards, actual USB transport/callbacks and
+        # production repeater CR command pump/time reply with DTR low. Add
+        # setup's one-second settle; filesystem/radio timing is not invented.
+        result = self.run_contract(platform="NRF52_PLATFORM", extra_boot_ms=1000)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_nrf52_old_blocking_gps_breaks_stock_gui_initial_time(self):
+        # Reinsert the former five-second GPS wait without changing the
+        # pinned client's actual timer, retry policy, or first command.
+        result = self.run_contract(platform="NRF52_PLATFORM", blocking=True,
+                                   extra_boot_ms=1000)
+        self.assertNotEqual(result.returncode, 0, "Negative control did not fail")
+        self.assertIn("initial time missed the stock 5000 ms deadline", result.stderr)
+
+    def test_nrf52_cr_framing_and_crlf_reply_are_required_by_stock_gui(self):
+        for options in ({"reject_cr": True}, {"lf_reply": True}):
+            with self.subTest(options=options):
+                result = self.run_contract(platform="NRF52_PLATFORM", extra_boot_ms=1000,
+                                           **options)
+                self.assertNotEqual(result.returncode, 0, "Negative control did not fail")
+                self.assertIn("initial time missed the stock 5000 ms deadline", result.stderr)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)