Keep updater USB connections clear of boot and modem controls

test/test_host_kiss_ingress.py

@@ -177,8 +177,8 @@ class HostKissIngressTest(unittest.TestCase):
             signature = ('static void __attribute__((noinline)) serviceCommandInterfaces()'
                          if role == 'simple_repeater' else 'void loop()')
             pump = body(source, signature)
-            self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'),
-                            pump.index('console.read()'))
+            read = 'const int value = console_read()' if role == 'simple_repeater' else 'console.read()'
+            self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'), pump.index(read))
 
 
 if __name__ == '__main__':

test/test_repeater_ota_serial_control.py

@@ -15,12 +15,16 @@ HARNESS = r'''
 #include <helpers/ota/MotaSourceSerial.h>
 #include <helpers/ota/MotaSeederProto.h>
 using namespace mesh::ota;
+namespace mesh {
+bool kiss_owned=false;
+bool isUsbBinaryStreamOwned(){return kiss_owned;}
+}
 struct Wire:Stream {
   std::deque<uint8_t> input;
   std::vector<uint8_t> next_response;
-  int writes=0;
+  int writes=0,reads=0;
   int available()override{return input.size();}
-  int read()override{if(input.empty())return -1;auto b=input.front();input.pop_front();return b;}
+  int read()override{++reads;if(input.empty())return -1;auto b=input.front();input.pop_front();return b;}
   size_t write(const uint8_t* data,size_t n)override{
     assert(n>=4&&data[0]=='M'&&data[1]=='S');++writes;
     input.insert(input.end(),next_response.begin(),next_response.end());next_response.clear();return n;
@@ -39,24 +43,33 @@ static std::string controls(SerialMotaSource& source){
   }
   return output;
 }
+static bool kissClaimAllowed(SerialMotaSource& control_source){
+  @KISS_GATE@
+  return kiss_can_read();
+}
 int main(){
   const std::string command="ota.peer cmd "+std::string(64,'1')+" ota key add "+std::string(64,'A')+"\r";
   const std::string dangerous="\rota.peer cmd "+std::string(64,'2')+" reboot\r\nota folder off\r";
-  const std::vector<uint8_t> payload(dangerous.begin(),dangerous.end());
+  std::vector<uint8_t> payload={0xc0,0,0x41,0xc0};
+  payload.insert(payload.end(),dangerous.begin(),dangerous.end());
   const auto response=frame(payload);
   for(size_t prefix=0;prefix<response.size();++prefix){
     g_mock_millis=0;Wire wire;
     SerialMotaSource source(wire,MotaStreamWritePolicy::NoFlush,5);
     source.enableSharedTextControl();
+    assert(kissClaimAllowed(source));
     wire.next_response.assign(response.begin(),response.begin()+prefix);
     std::vector<uint8_t> target(payload.size());
     assert(!source.read(0,0,target.data(),target.size()));
     assert(source.hasPendingResponse());
+    assert(!kissClaimAllowed(source));
     for(size_t i=prefix;i<response.size();++i){
       wire.input.push_back(response[i]);
       assert(controls(source).empty());
+      if(i+1<response.size())assert(!kissClaimAllowed(source));
     }
     assert(!source.hasPendingResponse());
+    assert(kissClaimAllowed(source));
     wire.push(command);
     assert(controls(source)==command);
   }
@@ -68,7 +81,9 @@ int main(){
   wire.push(command);
   wire.next_response=frame({42});
   uint8_t value=0;assert(source.read(0,0,&value,1)&&value==42);
+  assert(!kissClaimAllowed(source)); // queued complete CLI still belongs to console
   assert(controls(source)==command);
+  assert(kissClaimAllowed(source));
   // ASCII records can arrive after the binary request has started. Passwords
   // and command arguments containing the response magic remain whole text,
   // including when the console already consumed the beginning of the line.
@@ -78,11 +93,13 @@ int main(){
     SerialMotaSource shared(racing,MotaStreamWritePolicy::NoFlush,5);shared.enableSharedTextControl();
     racing.push(login.substr(0,split));
     const auto prefix=controls(shared);
+    if(split && login[split-1]!='\r' && login[split-1]!='\n')assert(!kissClaimAllowed(shared));
     racing.next_response.assign(login.begin()+split,login.end());
     const auto binary=frame({44});
     racing.next_response.insert(racing.next_response.end(),binary.begin(),binary.end());
     assert(shared.read(0,0,&value,1)&&value==44);
     assert(prefix+controls(shared)==login);
+    assert(kissClaimAllowed(shared));
   }
   // Over-capacity controls invalidate the existing prefix and discard to its
   // delimiter, so the suffix never becomes a second mutating command.
@@ -95,6 +112,21 @@ int main(){
   wire.push("ota.peer result\r");
   assert(controls(source)=="ota.peer result\r");
   source.resetSessionState();assert(!source.hasPendingResponse());
+  // An attached shared USB folder pauses immediately while KISS owns the
+  // port; neither stale-drain reads nor a new MS request may touch its frame.
+  Wire owned;SerialMotaSource paused(owned,MotaStreamWritePolicy::NoFlush,600);
+  paused.enableSharedTextControl();owned.input={0xc0,0,0x41,0xc0};
+  mesh::kiss_owned=true;const auto before_clock=g_mock_millis;
+  assert(!paused.read(0,0,&value,1));
+  assert(!owned.reads&&!owned.writes&&owned.input.size()==4&&g_mock_millis==before_clock);
+  assert(!paused.hasPendingResponse());
+  owned.input.clear();mesh::kiss_owned=false;owned.next_response=frame({45});
+  assert(paused.read(0,0,&value,1)&&value==45); // same attached source resumes
+  // Dedicated UART/BLE sources do not share the physical KISS stream.
+  Wire dedicated;SerialMotaSource separate(dedicated,MotaStreamWritePolicy::NoFlush,5);
+  dedicated.next_response=frame({46});mesh::kiss_owned=true;
+  assert(separate.read(0,0,&value,1)&&value==46);
+  mesh::kiss_owned=false;
 }
 '''
 
@@ -104,11 +136,16 @@ class RepeaterOtaSerialControlTest(unittest.TestCase):
         with tempfile.TemporaryDirectory(prefix='meshcore-repeater-mota-control-') as folder:
             cpp = Path(folder)/'test.cpp'
             binary = Path(folder)/('test.exe' if os.name == 'nt' else 'test')
-            cpp.write_text(HARNESS)
+            main = (ROOT/'examples/simple_repeater/main.cpp').read_text()
+            start = main.index('const auto kiss_can_read = [&]()')
+            end = main.index('\n  };', start)+len('\n  };')
+            cpp.write_text(HARNESS.replace('@KISS_GATE@', main[start:end]))
             sanitizers = [] if os.name == 'nt' else [
                 '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie']
             built=subprocess.run([
-                'g++','-std=c++17','-DESP32_PLATFORM=1',
+                'g++','-std=c++17','-DESP32_PLATFORM=1','-DMESH_ENABLE_DIRECT_OTA_ADMIN=1',
+                '-DMESH_OTA_FOLDER_SERIAL_SHARED_CONSOLE=1',
+                '-DMESH_ENABLE_HOST_KISS=1','-DARDUINO=1',
                 *sanitizers,
                 '-I'+str(ROOT/'test/mocks'),'-I'+str(ROOT/'src'),str(cpp),
                 str(ROOT/'src/helpers/ota/MotaSourceSerial.cpp'),'-o',str(binary)],

tools/lora_ota/ota_wizard.py

@@ -24,6 +24,7 @@ from ota_wizard_packages import (find_builder, find_running_image, install_build
 from ota_wizard_session import DirectSession
 from ota_wizard_bundle import (load_bundle, load_ready_application, load_request,
                                save_bundle, save_request)
+from ota_wizard_usb import open_serial_session
 
 ROOT = Path(__file__).resolve().parents[2]
 
@@ -139,10 +140,8 @@ def detect_mode(port: str) -> str:
     """A harmless text probe avoids feeding Binary commands to a repeater CLI."""
     import serial
     connection = serial.Serial(port=None, baudrate=115200, timeout=.1, write_timeout=2)
-    connection.dtr, connection.rts = True, False
-    connection.port = port
     try:
-        connection.open()
+        open_serial_session(connection, port)
         time.sleep(.75)
         connection.write(b"\rver\rota status\r")
         deadline = time.monotonic() + 3

tools/lora_ota/ota_wizard_companion.py

@@ -68,6 +68,52 @@ def _deflate(data):
     return compressor.compress(data) + compressor.flush()
 
 
+async def _create_usb_sdk(port, baudrate=115200, *, only_error=True,
+                          default_timeout=15, auto_reconnect=False):
+    """Use SDK framing with control lines selected before the USB port opens."""
+    import serial
+    import serial_asyncio_fast
+    from meshcore import MeshCore, SerialConnection
+    from ota_wizard_usb import open_serial_session
+
+    class ConfiguredConnection(SerialConnection):
+        async def connect(self, timeout=10):
+            self._connected_event.clear()
+            wire = serial.Serial(port=None, baudrate=self.baudrate, write_timeout=2)
+            try:
+                # The pinned SDK factory cannot configure DTR before opening.
+                # Adopt an already configured connection through the serial
+                # transport's public API; keep the SDK protocol and parser.
+                await asyncio.to_thread(open_serial_session, wire, self.port)
+                loop = asyncio.get_running_loop()
+                transport, _ = await serial_asyncio_fast.connection_for_serial(
+                    loop, lambda: self.MCSerialClientProtocol(self), wire)
+                self.transport = transport
+                await asyncio.wait_for(self._connected_event.wait(), timeout)
+                return self.port
+            except BaseException:
+                if self.transport is not None:
+                    self.transport.close()
+                    self.transport = None
+                await asyncio.to_thread(wire.close)
+                raise
+
+    connection = ConfiguredConnection(port, baudrate, cx_dly=.1)
+    sdk = MeshCore(connection, only_error=only_error, default_timeout=default_timeout,
+                   auto_reconnect=auto_reconnect)
+    try:
+        if await sdk.connect() is None:
+            await sdk.disconnect()
+            return None
+        return sdk
+    except BaseException:
+        try:
+            await sdk.disconnect()
+        except Exception:
+            pass
+        raise
+
+
 class CompanionTransport:
     def __init__(self, port, baud=115200, *, command_timeout=15,
                  remote_timeout=60, discovery_seconds=4, sdk_factory=None,
@@ -126,10 +172,10 @@ class CompanionTransport:
             return self
         if self._sdk_factory is None:
             try:
-                from meshcore import MeshCore, EventType
+                from meshcore import EventType
             except ImportError as exc:
                 raise CompanionError("Install the meshcore Python package for USB Companion support") from exc
-            self._sdk_factory, self._event_types = MeshCore.create_serial, EventType
+            self._sdk_factory, self._event_types = _create_usb_sdk, EventType
         if self._event_types is None:
             raise CompanionError("Companion event definitions are unavailable")
         try:

tools/lora_ota/ota_wizard_repeater.py

@@ -15,6 +15,7 @@ import time
 import zlib
 
 import lora_ota as ota
+from ota_wizard_usb import open_serial_session
 
 
 class RepeaterError(ota.OtaError):
@@ -178,9 +179,7 @@ class RepeaterTransport:
         def create():
             connection = factory(port=None, baudrate=self.baud, timeout=.1,
                                  write_timeout=3, rtscts=False, dsrdtr=False)
-            connection.dtr, connection.rts = True, False
-            connection.port = self.port
-            connection.open()
+            open_serial_session(connection, self.port)
             return connection
         try:
             self._serial = await asyncio.to_thread(create)

tools/lora_ota/ota_wizard_usb.py

@@ -0,0 +1,74 @@
+"""Select USB control-line states before opening a radio's serial port."""
+from __future__ import annotations
+
+import os
+import re
+
+
+def _device_path(value) -> str:
+    text = str(value)
+    # Windows callers may use either COM7 or the equivalent \\.\COM7.
+    match = re.fullmatch(r"(?:\\\\\.\\)?COM(\d+)", text, re.I)
+    if match:
+        return "com" + str(int(match[1]))
+    return os.path.realpath(text).casefold()
+
+
+def serial_line_states(port, *, ports=None) -> tuple[bool, bool]:
+    """Return (DTR, RTS), resolving aliases against the selected USB device.
+
+    ESP32 Serial/JTAG 303A:1001 and UART bridges must leave DTR released
+    across firmware reboots. Other Espressif CDC, Seeed and Adafruit/nRF52
+    CDC devices need DTR asserted. Unknown ports use the released state.
+    """
+    if ports is None:
+        try:
+            from serial.tools import list_ports
+            ports = list_ports.comports()
+        except (ImportError, OSError):
+            ports = ()
+    selected = _device_path(port)
+    for info in ports:
+        device = getattr(info, "device", None)
+        if device is None or _device_path(device) != selected:
+            continue
+        vid, pid = getattr(info, "vid", None), getattr(info, "pid", None)
+        return vid in (0x303A, 0x2886, 0x239A) and (vid, pid) != (0x303A, 0x1001), False
+    return False, False
+
+
+def configure_serial_session(connection, port=None, *, ports=None) -> tuple[bool, bool]:
+    """Configure a closed pySerial-compatible object without toggling a device."""
+    if getattr(connection, "is_open", False):
+        raise ValueError("USB control lines must be configured before opening the port")
+    if port is not None:
+        connection.port = port
+    dtr, rts = serial_line_states(connection.port, ports=ports)
+    connection.rts = rts
+    connection.dtr = dtr
+    return dtr, rts
+
+
+def open_serial_session(connection, port=None, *, ports=None) -> None:
+    """Open configured lines without a POSIX USB-UART bridge reset pulse."""
+    configure_serial_session(connection, port, ports=ports)
+    # POSIX pySerial normally applies DTR before RTS. Releasing DTR while
+    # the driver still asserts RTS can pulse ESP32 EN. Defer DTR until RTS
+    # has been released, matching tools/hil/serial_session.py.
+    defer_dtr = os.name == "posix" and not connection.dtr
+    previous_dsrdtr = getattr(connection, "dsrdtr", False)
+    try:
+        if defer_dtr:
+            connection.dsrdtr = True
+        connection.open()
+        if defer_dtr:
+            connection.rts = False
+            connection.dtr = False
+            connection.dsrdtr = previous_dsrdtr
+    except BaseException:
+        try:
+            connection.close()
+        finally:
+            if defer_dtr:
+                connection.dsrdtr = previous_dsrdtr
+        raise

tools/lora_ota/test_ota_wizard_companion.py

@@ -3,6 +3,7 @@ import asyncio
 from copy import deepcopy
 from enum import Enum
 from types import SimpleNamespace
+import sys
 import unittest
 from unittest import mock
 import time
@@ -460,5 +461,112 @@ class PureHelpers(unittest.TestCase):
                            companion.packet_airtime(20, (910, 62.5, 7, 5)))
 
 
+class UsbConstructionTests(unittest.IsolatedAsyncioTestCase):
+    async def create_sdk(self, dtr, *, handshake=True):
+        import ota_wizard_usb as usb
+        history = []
+
+        class Wire:
+            def __init__(self, port, baudrate, write_timeout):
+                self.port, self.baudrate = port, baudrate
+                self.is_open, self.dsrdtr = False, False
+                self._dtr = self._rts = True
+                history.append(("construct", port))
+
+            @property
+            def dtr(self):
+                return self._dtr
+
+            @dtr.setter
+            def dtr(self, value):
+                history.append(("dtr", value, self.is_open))
+                self._dtr = value
+
+            @property
+            def rts(self):
+                return self._rts
+
+            @rts.setter
+            def rts(self, value):
+                history.append(("rts", value, self.is_open))
+                self._rts = value
+
+            def open(self):
+                history.append(("open", self.dtr, self.rts))
+                self.is_open = True
+
+            def close(self):
+                history.append(("close",))
+                self.is_open = False
+
+        class SerialConnection:
+            def __init__(self, port, baudrate, cx_dly):
+                self.port, self.baudrate = port, baudrate
+                self.transport = None
+                self._connected_event = asyncio.Event()
+
+            class MCSerialClientProtocol:
+                def __init__(self, connection):
+                    self.connection = connection
+
+                def connection_made(self, transport):
+                    self.connection.transport = transport
+                    transport.serial.rts = False
+                    self.connection._connected_event.set()
+
+        class MeshCore:
+            last = None
+            def __init__(self, connection, **settings):
+                MeshCore.last = self
+                self.connection, self.settings = connection, settings
+
+            async def connect(self):
+                await self.connection.connect()
+                return event() if handshake else None
+
+            async def disconnect(self):
+                if self.connection.transport:
+                    self.connection.transport.close()
+                    self.connection.transport = None
+
+        async def adopt(loop, protocol_factory, wire):
+            self.assertTrue(wire.is_open)
+            self.assertEqual((wire.dtr, wire.rts), (dtr, False))
+            history.append(("adopt", wire.dtr, wire.rts))
+            protocol = protocol_factory()
+            transport = SimpleNamespace(serial=wire, close=wire.close)
+            protocol.connection_made(transport)
+            return transport, protocol
+
+        modules = {"meshcore": SimpleNamespace(MeshCore=MeshCore, SerialConnection=SerialConnection),
+                   "serial": SimpleNamespace(Serial=Wire),
+                   "serial_asyncio_fast": SimpleNamespace(connection_for_serial=adopt)}
+        with mock.patch.dict(sys.modules, modules), \
+                mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)):
+            sdk = await companion._create_usb_sdk("USB-TEST", default_timeout=.1)
+        return sdk, MeshCore.last, history
+
+    async def test_native_usb_dtr_is_false_before_sdk_adopts_connection(self):
+        sdk, instance, history = await self.create_sdk(False)
+        self.assertIs(sdk, instance)
+        self.assertLess(history.index(("dtr", False, False)), history.index(("open", False, False)))
+        self.assertFalse(any(item[0] == "dtr" and item[1] for item in history))
+        self.assertFalse(sdk.settings["auto_reconnect"])
+        await sdk.disconnect()
+        self.assertEqual(history[-1], ("close",))
+
+    async def test_tinyusb_nrf_dtr_is_true_before_open(self):
+        sdk, _, history = await self.create_sdk(True)
+        self.assertLess(history.index(("dtr", True, False)), history.index(("open", True, False)))
+        await sdk.disconnect()
+        self.assertEqual(history[-1], ("close",))
+
+    async def test_failed_binary_handshake_closes_the_configured_port(self):
+        sdk, instance, history = await self.create_sdk(False, handshake=False)
+        self.assertIsNone(sdk)
+        self.assertIsNone(instance.connection.transport)
+        self.assertEqual(history[-1], ("close",))
+
+
 if __name__ == "__main__":
     unittest.main()

tools/lora_ota/test_ota_wizard_session.py

@@ -262,6 +262,9 @@ class SessionTests(unittest.IsolatedAsyncioTestCase):
             await self.session.cleanup()
 
     async def test_destination_temporarily_missing_after_reboot_is_probed_again(self):
+        import itertools
+        from types import SimpleNamespace
+        from unittest import mock
         from ota_wizard_companion import CompanionError
         before = await self.session.inspect()
         original = self.link.remote_command
@@ -273,7 +276,11 @@ class SessionTests(unittest.IsolatedAsyncioTestCase):
                 raise CompanionError("No directly reachable destination replied yet")
             return await original(key, password, text)
         self.link.remote_command = reboot_discovery
-        after = await self.session.install(wizard.full_package(application(), 1024), before)
+        # Advance only this module's deadline clock; compiler load must not
+        # exhaust the tiny test reboot budget between three fake probes.
+        clock = SimpleNamespace(monotonic=itertools.count(0, .001).__next__)
+        with mock.patch.object(wizard, "time", clock):
+            after = await self.session.install(wizard.full_package(application(), 1024), before)
         self.assertEqual(after.version, NEW)
         self.assertEqual(missing, 0)
         self.assertEqual(len(self.link.reset_calls), 3)

tools/lora_ota/test_ota_wizard_usb.py

@@ -0,0 +1,186 @@
+"""USB enumeration policy and pre-open line ordering without hardware access."""
+import sys
+from types import ModuleType, SimpleNamespace
+import unittest
+from unittest import mock
+
+import ota_wizard as wizard
+import ota_wizard_usb as usb
+from ota_wizard_repeater import RepeaterTransport
+from test_ota_wizard_repeater import FakeSerial
+
+
+def metadata(device="COM7", vid=0x303A, pid=0x1001):
+    return SimpleNamespace(device=device, vid=vid, pid=pid)
+
+
+class SerialFixture:
+    def __init__(self, **kwargs):
+        self.is_open = False
+        self.events = []
+        self.port = None
+        self._rts = self._dtr = True
+        self._dsrdtr = False
+        self.open_error = None
+
+    @property
+    def dtr(self):
+        return self._dtr
+
+    @dtr.setter
+    def dtr(self, value):
+        self.events.append(("dtr", value, self.is_open))
+        self._dtr = value
+
+    @property
+    def rts(self):
+        return self._rts
+
+    @rts.setter
+    def rts(self, value):
+        self.events.append(("rts", value, self.is_open))
+        self._rts = value
+
+    @property
+    def dsrdtr(self):
+        return self._dsrdtr
+
+    @dsrdtr.setter
+    def dsrdtr(self, value):
+        self.events.append(("dsrdtr", value, self.is_open))
+        self._dsrdtr = value
+
+    def open(self):
+        self.events.append(("open", self.dtr, self.rts, self.dsrdtr, self.port))
+        self.is_open = True
+        if self.open_error:
+            raise self.open_error
+
+    def close(self):
+        self.events.append(("close",))
+        self.is_open = False
+
+    def write(self, data):
+        self.events.append(("write", data))
+        return len(data)
+
+    def read(self, length):
+        return b"  -> v1.18.0.1\r\n"
+
+
+class UsbPolicyTests(unittest.TestCase):
+    def test_vid_pid_matrix_dtr_with_rts_always_released(self):
+        cases = ((0x303A, 0x1001, False), (0x303A, 0x4001, True),
+                 (0x2886, 0x0044, True), (0x239A, 0x8029, True),
+                 (0x10C4, 0xEA60, False), (0x1A86, 0x7523, False),
+                 (0x0403, 0x6001, False), (None, None, False))
+        for vid, pid, dtr in cases:
+            with self.subTest(vid=vid, pid=pid):
+                ports = [metadata(vid=vid, pid=pid)]
+                self.assertEqual(usb.serial_line_states("COM7", ports=ports), (dtr, False))
+        self.assertEqual(usb.serial_line_states("COM8", ports=[metadata(vid=0x239A)]),
+                         (False, False))
+
+    def test_windows_com_aliases_and_posix_by_id_resolve_same_device(self):
+        self.assertEqual(usb.serial_line_states(r"\\.\COM7", ports=[metadata(vid=0x239A)]),
+                         (True, False))
+        self.assertEqual(usb.serial_line_states("com7", ports=[metadata(vid=0x2886)]),
+                         (True, False))
+        def resolved(value):
+            return "/dev/ttyACM0" if value == "/dev/serial/by-id/radio" else value
+        with mock.patch.object(usb.os.path, "realpath", side_effect=resolved):
+            self.assertEqual(usb.serial_line_states("/dev/serial/by-id/radio",
+                ports=[metadata("/dev/ttyACM0", vid=0x2886)]), (True, False))
+
+    def test_lazy_enumeration_and_missing_metadata_default_to_released(self):
+        serial, tools = ModuleType("serial"), ModuleType("serial.tools")
+        listing = mock.Mock(return_value=[metadata(vid=0x239A)])
+        tools.list_ports = SimpleNamespace(comports=listing)
+        serial.tools = tools
+        with mock.patch.dict(sys.modules, {"serial": serial, "serial.tools": tools}):
+            self.assertEqual(usb.serial_line_states("COM7"), (True, False))
+            listing.assert_called_once_with()
+            listing.side_effect = OSError("USB device disappeared")
+            self.assertEqual(usb.serial_line_states("COM7"), (False, False))
+
+    def test_configuration_happens_before_open_and_rejects_open_object(self):
+        port = SerialFixture()
+        self.assertEqual(usb.configure_serial_session(port, "COM7", ports=[metadata()]),
+                         (False, False))
+        self.assertEqual(port.events, [("rts", False, False), ("dtr", False, False)])
+        self.assertEqual(port.port, "COM7")
+        port.is_open = True
+        before = list(port.events)
+        with self.assertRaisesRegex(ValueError, "before opening"):
+            usb.configure_serial_session(port, "COM8", ports=[])
+        self.assertEqual(port.events, before)
+        self.assertEqual(port.port, "COM7")
+
+    def test_posix_release_defers_dtr_until_after_rts_without_handshaking_left_on(self):
+        port = SerialFixture()
+        with mock.patch.object(usb, "os", SimpleNamespace(name="posix")), \
+                mock.patch.object(usb, "serial_line_states", return_value=(False, False)):
+            usb.open_serial_session(port, "/dev/ttyACM0")
+        self.assertEqual(port.events, [
+            ("rts", False, False), ("dtr", False, False), ("dsrdtr", True, False),
+            ("open", False, False, True, "/dev/ttyACM0"),
+            ("rts", False, True), ("dtr", False, True), ("dsrdtr", False, True)])
+        self.assertFalse(port.dsrdtr)
+
+    def test_native_cdc_and_windows_open_without_posix_bridge_sequence(self):
+        for platform, dtr in (("posix", True), ("nt", True), ("nt", False)):
+            port = SerialFixture()
+            with self.subTest(platform=platform, dtr=dtr), \
+                    mock.patch.object(usb, "os", SimpleNamespace(name=platform)), \
+                    mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)):
+                usb.open_serial_session(port, "COM7")
+            self.assertEqual(port.events, [("rts", False, False), ("dtr", dtr, False),
+                                          ("open", dtr, False, False, "COM7")])
+
+    def test_open_failure_closes_device_and_restores_handshaking(self):
+        port = SerialFixture()
+        port.open_error = OSError("open failed")
+        with mock.patch.object(usb, "os", SimpleNamespace(name="posix")), \
+                mock.patch.object(usb, "serial_line_states", return_value=(False, False)), \
+                self.assertRaisesRegex(OSError, "open failed"):
+            usb.open_serial_session(port, "/dev/ttyUSB0")
+        self.assertFalse(port.is_open)
+        self.assertFalse(port.dsrdtr)
+        self.assertIn(("close",), port.events)
+
+    def test_detect_mode_obeys_shared_policy_before_first_probe(self):
+        serial = ModuleType("serial")
+        connection = SerialFixture()
+        serial.Serial = mock.Mock(return_value=connection)
+        with mock.patch.dict(sys.modules, {"serial": serial}), \
+                mock.patch.object(usb, "serial_line_states", return_value=(False, False)), \
+                mock.patch.object(wizard.time, "sleep"), \
+                mock.patch.object(wizard.time, "monotonic", side_effect=[0, 0, 4]):
+            self.assertEqual(wizard.detect_mode("COM7"), "repeater")
+        opened = next(event for event in connection.events if event[0] == "open")
+        self.assertEqual(opened[1:3], (False, False))
+        self.assertFalse(connection.is_open)
+
+
+class RepeaterUsbPolicyTests(unittest.IsolatedAsyncioTestCase):
+    async def test_repeater_opens_with_shared_policy_for_both_device_types(self):
+        for dtr in (False, True):
+            connection = FakeSerial()
+            observed = []
+            def open_port():
+                observed.append((connection.dtr, connection.rts, connection.port))
+                connection.is_open = True
+            connection.open = open_port
+            link = RepeaterTransport("COM7", open_settle=0,
+                                     serial_factory=lambda **kwargs: connection)
+            with self.subTest(dtr=dtr), \
+                    mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)):
+                try:
+                    await link.open()
+                    self.assertEqual(observed, [(dtr, False, "COM7")])
+                finally:
+                    await link.close()
+
+
+if __name__ == "__main__":
+    unittest.main()