Use C6L native USB and recover diagnostics-only Companion sessions

examples/companion_radio/main.cpp

@@ -3546,6 +3546,15 @@ void setup() {
 #endif
 }
 
+static void serviceUsbLoggingOnlySession() {
+#if MESH_ESP32_USB_CONSOLE_COOPERATIVE && !defined(ENABLE_USB_INTERFACE)
+  // BLE/WiFi-only Companions still own native USB diagnostics. Mark setup
+  // complete and finish bus-reset cleanup even without a USB protocol owner.
+  (void)mesh::takeUsbTerminalSessionReset();
+  (void)mesh::tryCompleteUsbTerminalSessionReset();
+#endif
+}
+
 static bool usbLoggingRecoverySafe(void*) {
   if (board.isOTAUpdateRunning() || board.isRadioTestActive()
       || radio_driver.isWatchdogObserving() || radio_driver.isCalibratingNoiseFloor()
@@ -3581,6 +3590,7 @@ void loop() {
 #if defined(NRF52_PLATFORM)
   board.feedWatchdog();
 #endif
+  serviceUsbLoggingOnlySession();
 #if defined(ENABLE_USB_INTERFACE)
   serviceUsbTerminalHostSessionReset();
   serviceUsbAsciiSessionDefault();

test/test_esp32_hwcdc_recipes.py

@@ -52,14 +52,13 @@ LEGACY_DIAGNOSTIC_CDC_OVERRIDES = {
     "xiao_s3_partition_migrator_test_usb_diagnostic",
 }
 
-# Pin the unaffected family/transport counts rather than copying a machine's
+# Pin reviewed family/transport counts rather than copying a machine's
 # environment inventory. Mixed families have explicit role rules below.
 UNCHANGED_COUNTS = {
     ("ESP32-S3-WROOM-1-N4", "uart"): 21,
     ("ebyte_eora-s3", "hwcdc"): 8,
     ("esp32-c3-devkitm-1", "uart"): 19,
-    ("esp32-c6-devkitm-1", "hwcdc"): 12,
-    ("esp32-c6-devkitm-1", "uart"): 4,
+    ("esp32-c6-devkitm-1", "hwcdc"): 16,
     ("esp32-s3-devkitc-1", "hwcdc"): 22,
     ("esp32-s3-devkitc-1", "uart"): 55,
     ("esp32-s3-zero", "hwcdc"): 17,
@@ -149,16 +148,12 @@ def bool_option(value):
     return str(value).lower() in ("true", "1", "yes")
 
 
-def expected_unchanged_transport(board, name):
+def expected_other_transport(board, name):
     choices = {kind for family, kind in UNCHANGED_COUNTS if family == board}
     if len(choices) == 1:
         return next(iter(choices))
     if board == "esp32-s3-devkitc-1":
         return "hwcdc" if name.startswith(("Heltec_Wireless_Tracker_", "RAK_3112_")) else "uart"
-    if board == "esp32-c6-devkitm-1":
-        if name.startswith("M5Stack_Unit_C6L_"):
-            return "hwcdc" if name == "M5Stack_Unit_C6L_companion_radio_usb" else "uart"
-        return "hwcdc"
     if board in ("heltec_v4_migrator", "seeed_xiao_esp32s3_migrator"):
         return "hwcdc" if name.endswith("_test_usb_diagnostic") else "uart"
     raise AssertionError(f"unreviewed ESP32 board: {board}")
@@ -266,15 +261,49 @@ class Esp32HwcdcRecipeTests(unittest.TestCase):
                 self.assertEqual(self.sdk["hwcdc_primary_enabled"](row["env"]),
                                  row["kind"] == "hwcdc")
                 self.assertFalse(self.sdk["native_primary_enabled"](row["env"]))
-        self.assertEqual(counts, {"hwcdc": 319, "uart": 122, "classic_uart": 61})
+        self.assertEqual(counts, {"hwcdc": 323, "uart": 118, "classic_uart": 61})
 
-    def test_uart_existing_hwcdc_and_newer_mcu_recipes_keep_their_transports(self):
+    def test_all_m5stack_native_usb_roles_select_hwcdc_and_its_startup_patch(self):
+        # M5's documented USB-C wiring exposes C6 Serial/JTAG, including when
+        # the Companion uses BLE. Resolve real SDK/compiler flags so a UART
+        # console regression cannot hide in the unchanged-transport counts.
+        names = {name for name in self.rows if name.startswith("M5Stack_Unit_C6L_")}
+        self.assertEqual(names, {
+            "M5Stack_Unit_C6L_repeater", "M5Stack_Unit_C6L_room_server",
+            "M5Stack_Unit_C6L_companion_radio_ble",
+            "M5Stack_Unit_C6L_companion_radio_usb", "M5Stack_Unit_C6L_kiss_modem",
+        })
+        for name in names:
+            with self.subTest(environment=name):
+                row = self.rows[name]
+                self.assertEqual(row["mcu"], "esp32c6")
+                self.assertEqual(row["kind"], "hwcdc")
+                self.assertEqual(row["values"], {USB_MODE: "1", USB_CDC: "1"})
+                self.assertTrue(self.sdk["hwcdc_primary_enabled"](row["env"]))
+                self.assertFalse(self.sdk["native_primary_enabled"](row["env"]))
+                self.assertEqual(self.config.get(f"env:{name}", "extra_scripts", []).count(
+                    "pre:scripts/esp32_usb_session_fix.py"), 1)
+                self.assertEqual(self.config.get(f"env:{name}", "platform"),
+                                 "https://github.com/pioarduino/platform-espressif32/releases/"
+                                 "download/53.03.13-1/platform-espressif32.zip")
+
+    def test_m5stack_selector_removal_reproduces_the_old_uart_backend(self):
+        for name in ("M5Stack_Unit_C6L_repeater", "M5Stack_Unit_C6L_room_server",
+                     "M5Stack_Unit_C6L_companion_radio_ble", "M5Stack_Unit_C6L_kiss_modem"):
+            with self.subTest(environment=name):
+                env = flag_environment(build_flags=self.config.get(f"env:{name}", "build_flags"),
+                                       build_unflags=["-DARDUINO_USB_MODE=1",
+                                                      "-DARDUINO_USB_CDC_ON_BOOT=1"])
+                self.assertEqual(usb_values(env), {USB_MODE: "0", USB_CDC: "0"})
+                self.assertFalse(self.sdk["hwcdc_primary_enabled"](env))
+
+    def test_uart_existing_hwcdc_and_newer_mcu_recipes_match_reviewed_transports(self):
         counts = Counter()
         for name, row in self.rows.items():
             if not row["esp32"] or row["board"] in MIGRATED_BOARD_COUNTS:
                 continue
             with self.subTest(environment=name):
-                expected = expected_unchanged_transport(row["board"], name)
+                expected = expected_other_transport(row["board"], name)
                 self.assertEqual(row["kind"], expected)
                 counts[row["board"], row["kind"]] += 1
         self.assertEqual(counts, UNCHANGED_COUNTS)

test/test_hwcdc_startup_session.py

@@ -24,6 +24,7 @@ EXTRA = r'''
 #include <iostream>
 #include <string>
 #define MESH_ESP32_HWCDC_SESSION_GUARD 1
+#define MESH_ESP32_USB_CONSOLE_COOPERATIVE 1
 #define MESH_HWCDC_PINNED_TX_BACKPORT 1
 #define portENTER_CRITICAL(m) portENTER_CRITICAL_SAFE(m)
 #define portEXIT_CRITICAL(m) portEXIT_CRITICAL_SAFE(m)
@@ -154,6 +155,19 @@ int main(int argc, char** argv) {
     assert(tryCompleteUsbTerminalSessionReset());
     assert(phy_detaches == 1 && host_rx.empty() && ring.data.empty());
     assert(canAccessEsp32Hwcdc(nullptr) && mesh_hwcdc_tx_allowed);
+  } else if (name == "logging_only") {
+    // A BLE-only Companion has no USB parser to close the startup exemption.
+    // Its real logging-only service must do so before the next host reset.
+    enumerate(); deliverEvents(); serviceUsbLoggingOnlySession();
+    assert(meshEsp32HwcdcShouldReportBusReset());
+    assert(phy_detaches == 0 && canAccessEsp32Hwcdc(nullptr));
+    ring.data.push_back({'o', 'l', 'd', '\r', '\n'});
+    host_rx = {'o', 'l', 'd', '\r'};
+    enumerate(); deliverEvents();
+    assert(!canAccessEsp32Hwcdc(nullptr));
+    serviceUsbLoggingOnlySession();
+    assert(phy_detaches == 1 && host_rx.empty() && ring.data.empty());
+    assert(canAccessEsp32Hwcdc(nullptr));
   } else { assert(false && "Unknown startup case"); }
   std::cout << "PASS " << name << " phy_detaches=" << phy_detaches
             << " retained_rx=" << host_rx.size() << '\n';
@@ -162,7 +176,8 @@ int main(int argc, char** argv) {
 
 
 def application_sources(*, remove_boot_gate=False, remove_owner_hook=False,
-                        boot_traffic_arms=False, keep_startup_forever=False):
+                        boot_traffic_arms=False, keep_startup_forever=False,
+                        remove_logging_only_service=False):
     source = (ROOT / "src/helpers/UsbLogging.cpp").read_text()
     signatures = (
         "static bool canAccessEsp32Hwcdc(void*)",
@@ -175,6 +190,15 @@ def application_sources(*, remove_boot_gate=False, remove_owner_hook=False,
         "bool tryCompleteUsbTerminalSessionReset()",
     )
     functions = "\n".join(body(source, signature) for signature in signatures)
+    main = (ROOT / "examples/companion_radio/main.cpp").read_text()
+    logging_service = body(main, "static void serviceUsbLoggingOnlySession()")
+    # The production session methods are extracted into this flat host shell.
+    logging_service = logging_service.replace("mesh::", "")
+    if remove_logging_only_service:
+        logging_service = logging_service.replace(
+            "  (void)takeUsbTerminalSessionReset();\n", "").replace(
+                "  (void)tryCompleteUsbTerminalSessionReset();\n", "")
+    functions += "\n" + logging_service
     owner = body(source, 'extern "C" bool meshEsp32HwcdcShouldReportBusReset()')
     # The extracted application methods live in a flat test shell. Only their
     # namespace qualification changes; state and the actual owner hook remain.
@@ -238,6 +262,67 @@ class HwcdcStartupSessionTests(unittest.TestCase):
             with self.subTest(case=case):
                 self.run_case(case)
 
+    def test_logging_only_companion_ends_startup_and_cleans_active_reset(self):
+        self.run_case("logging_only")
+
+    def test_missing_logging_only_service_keeps_startup_pending_negative_control(self):
+        self.run_case("logging_only", expect_success=False, remove_logging_only_service=True)
+
+    def test_logging_only_companion_scope_uses_actual_transport_macros(self):
+        main = (ROOT / "examples/companion_radio/main.cpp").read_text()
+        helper = body(main, "static void serviceUsbLoggingOnlySession()")
+        loop = body(main[main.rindex("\nvoid loop()"):], "void loop()")
+        self.assertLess(loop.index("serviceUsbLoggingOnlySession();"),
+                        loop.index("the_mesh.loop();"))
+        self.assertLess(loop.index("serviceUsbLoggingOnlySession();"),
+                        loop.index("mesh::serviceUsbLoggingPort();"))
+        harness = r'''
+#include <helpers/UsbLogging.h>
+#include <cassert>
+namespace mesh {
+static unsigned calls = 0;
+bool takeUsbTerminalSessionReset() { ++calls; return true; }
+bool tryCompleteUsbTerminalSessionReset() { ++calls; return true; }
+}
+@HELPER@
+int main() {
+  serviceUsbLoggingOnlySession();
+  assert(mesh::calls == EXPECT_SERVICE * 2);
+}
+'''.replace("@HELPER@", helper)
+        profiles = {
+            "hwcdc": ["ESP32=1", "ARDUINO_USB_MODE=1", "ARDUINO_USB_CDC_ON_BOOT=1"],
+            "tinyusb": ["ESP32=1", "ARDUINO_USB_MODE=0", "ARDUINO_USB_CDC_ON_BOOT=1"],
+            "new_uart": ["ESP32=1", "ARDUINO_USB_MODE=1", "ARDUINO_USB_CDC_ON_BOOT=0"],
+            "classic_uart": ["ESP32=1"],
+            "nrf52": ["NRF52_PLATFORM=1", "USE_TINYUSB=1"],
+            "rp2040": ["RP2040_PLATFORM=1"],
+        }
+        with tempfile.TemporaryDirectory(prefix="companion-logging-session-") as directory:
+            directory = Path(directory)
+            cpp = directory / "scope.cpp"
+            cpp.write_text(harness, encoding="ascii")
+            for profile, flags in profiles.items():
+                for usb_protocol in (False, True):
+                    with self.subTest(profile=profile, usb_protocol=usb_protocol):
+                        binary = directory / (profile + str(usb_protocol))
+                        defines = ["-DARDUINO", *("-D" + flag for flag in flags),
+                                   "-DEXPECT_SERVICE=" + str(int(
+                                       profile in ("hwcdc", "tinyusb") and not usb_protocol))]
+                        if usb_protocol:
+                            defines.append("-DENABLE_USB_INTERFACE=1")
+                        compiled = subprocess.run([
+                            "g++", "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                            "-isystem", str(ROOT / "test/mocks"), "-I" + str(ROOT / "src"),
+                            *defines, str(cpp), "-o", str(binary)],
+                            capture_output=True, text=True, timeout=60)
+                        self.assertEqual(compiled.returncode, 0,
+                                         compiled.stdout + compiled.stderr)
+                        executed = subprocess.run([str(binary)], capture_output=True,
+                                                  text=True, timeout=10)
+                        self.assertEqual(executed.returncode, 0,
+                                         executed.stdout + executed.stderr)
+
     def test_original_boot_gate_failure_negative_control(self):
         self.run_case("early", expect_success=False, remove_boot_gate=True)
 

variants/m5stack_unit_c6l/platformio.ini

@@ -7,6 +7,10 @@ build_flags =
   ${esp32c6_base.build_flags}
   ${sensor_base.build_flags}
   -I variants/m5stack_unit_c6l
+  ; USB-C connects directly to the C6 USB Serial/JTAG peripheral on GPIO12/13.
+  ; Every role needs the native console backend, including BLE and KISS.
+  -D ARDUINO_USB_CDC_ON_BOOT=1
+  -D ARDUINO_USB_MODE=1
   -D P_LORA_TX_LED=15
   -D P_LORA_SCLK=20
   -D P_LORA_MISO=22
@@ -100,8 +104,6 @@ build_flags = ${M5Stack_Unit_C6L.build_flags}
   -D MAX_CONTACTS=350
   -D MAX_GROUP_CHANNELS=40
   -D OFFLINE_QUEUE_SIZE=256
-  -D ARDUINO_USB_CDC_ON_BOOT=1
-  -D ARDUINO_USB_MODE=1
   -D ENABLE_USB_INTERFACE
 build_src_filter = ${M5Stack_Unit_C6L.build_src_filter}
   +<helpers/esp32/*.cpp>