Avoid reset pulses when serial test collectors reopen UART bridges

test/test_hil_esp32_self_serve_ota.py

@@ -25,7 +25,9 @@ def load_module(name, path):
     return module
 
 
-SOAK = load_module("self_serve_test_soak", HIL_DIR / "s3_memory_soak.py")
+SESSION = load_module("self_serve_test_serial_session", HIL_DIR / "serial_session.py")
+with mock.patch.dict(sys.modules, {"serial_session": SESSION}):
+    SOAK = load_module("self_serve_test_soak", HIL_DIR / "s3_memory_soak.py")
 with mock.patch.dict(sys.modules, {"s3_memory_soak": SOAK}):
     HIL = load_module("self_serve_test_harness", HIL_DIR / "esp32_self_serve_ota.py")
 

test/test_hil_esp32_usb_host_reboot.py

@@ -25,7 +25,9 @@ def load_module(name, path):
     return module
 
 
-SOAK = load_module("usb_reboot_test_soak", HIL_DIR / "s3_memory_soak.py")
+SESSION = load_module("usb_reboot_test_serial_session", HIL_DIR / "serial_session.py")
+with mock.patch.dict(sys.modules, {"serial_session": SESSION}):
+    SOAK = load_module("usb_reboot_test_soak", HIL_DIR / "s3_memory_soak.py")
 with mock.patch.dict(sys.modules, {"s3_memory_soak": SOAK}):
     OTA = load_module("usb_reboot_test_ota", HIL_DIR / "esp32_self_serve_ota.py")
 with mock.patch.dict(sys.modules, {"esp32_self_serve_ota": OTA}):

tools/hil/contact_cache_serial_stress.py

@@ -19,6 +19,8 @@ import struct
 import time
 
 import serial
+from serial_session import open_configured_session
+from profile_switch import configure_session
 
 from esp32_companion_serial_stress import (
     DeviceFrameReader,
@@ -43,12 +45,10 @@ def secret_counters(reply):
 
 def run(args):
     port = serial.Serial(None, 115200, timeout=0.05, write_timeout=10)
-    port.dtr = False
-    port.rts = False
     port.port = args.port
-    port.open()
+    configure_session(port)
+    open_configured_session(port)
     try:
-        port.dtr = True
         time.sleep(4)
         reader = DeviceFrameReader(TransportCounters())
 

tools/hil/esp32_companion_serial_stress.py

@@ -30,6 +30,7 @@ import struct
 import sys
 import time
 from typing import Any, Callable, Dict, Optional, Sequence, Tuple
+from serial_session import open_configured_session
 
 
 HOST_FRAME_MARKER = ord("<")
@@ -466,9 +467,9 @@ def _make_pyserial_factory() -> Callable[[StressConfig], Any]:
         ) from exc
 
     def factory(config: StressConfig) -> Any:
-        # Native USB/JTAG devices may require DTR for the CDC session. Set both
-        # control lines while closed and keep RTS deasserted, including when
-        # opting into DTR, to avoid deliberately driving the ESP reset line.
+        # Native USB/JTAG devices may require DTR for the CDC session. POSIX
+        # pySerial opens bridges by applying DTR before RTS; defer DTR there
+        # so a kernel's initial asserted lines cannot pulse the ESP reset line.
         port = serial.Serial()
         port.port = config.port
         port.baudrate = config.baudrate
@@ -476,7 +477,7 @@ def _make_pyserial_factory() -> Callable[[StressConfig], Any]:
         port.write_timeout = config.write_timeout
         port.dtr = config.dtr
         port.rts = False
-        port.open()
+        open_configured_session(port)
         return port
 
     return factory

tools/hil/fem_noise_floor_test.py

@@ -42,6 +42,7 @@ import time
 from typing import Any, Callable, Dict, List, Optional, Sequence
 
 import esp32_companion_serial_stress as companion
+from serial_session import open_configured_session
 
 
 CMD_GET_FEM_RX_GAIN = 0x42
@@ -141,16 +142,17 @@ def _radio_stats_spec() -> companion.RequestSpec:
     )
 
 
-def _resolve_dtr(config: FemTestConfig) -> bool:
+def _resolve_dtr(config: FemTestConfig, port: Any) -> bool:
     if config.dtr == "on":
         return True
     if config.dtr == "off":
         return False
-    # Text consoles require an asserted host session on both nRF52 USB CDC and
-    # ESP32-S3 USB-Serial/JTAG. Companion framing on V4 remains usable without
-    # DTR and avoids changing the device's host-session state during a binary
-    # protocol test.
-    return config.protocol == "cli" or "HT-n5262" in config.port
+    # AUTO follows the actual USB transport, not the CLI/framing protocol.
+    # A bridge's DTR may hold GPIO0; native CDC uses it as a host session.
+    # Import only after open() has requested the optional serial dependency.
+    from profile_switch import configure_session
+    configure_session(port)
+    return bool(port.dtr)
 
 
 class CompanionFemTransport:
@@ -176,9 +178,9 @@ class CompanionFemTransport:
             self.config.read_poll_timeout, self.config.response_timeout
         )
         port.write_timeout = self.config.write_timeout
-        port.dtr = _resolve_dtr(self.config)
+        port.dtr = _resolve_dtr(self.config, port)
         port.rts = False
-        port.open()
+        open_configured_session(port)
         self.counters.ports_opened += 1
         self.port = port
         try:
@@ -292,9 +294,9 @@ class CliFemTransport:
             self.config.read_poll_timeout, self.config.response_timeout
         )
         port.write_timeout = self.config.write_timeout
-        port.dtr = _resolve_dtr(self.config)
+        port.dtr = _resolve_dtr(self.config, port)
         port.rts = False
-        port.open()
+        open_configured_session(port)
         self.port = port
         try:
             if self.config.open_delay:

tools/hil/profile_four_tx_deploy.py

@@ -7,7 +7,7 @@ import subprocess
 import time
 import zipfile
 from profile_four_tx_fixture import TXS,RX,resolve
-from profile_switch import exchange,configure_session
+from profile_switch import exchange,open_session
 import serial
 from serial.tools import list_ports
 
@@ -52,7 +52,7 @@ def ready(t):
         try:
             name=resolve(t)
             p=serial.Serial();p.port=name;p.baudrate=115200;p.timeout=.2;p.write_timeout=2
-            configure_session(p);p.open()
+            open_session(p)
             try:
                 info=exchange(p,'info','board',3)
                 if info.get('bench')!='production-profile-switch-v8' or not info.get('ready'):

tools/hil/profile_four_tx_package.py

@@ -41,7 +41,7 @@ for name in ('firmware.bin','partitions.bin','bootloader.bin'):
     add('rx-'+name,(ROOT/'.pio/build'/RX['env']/name).read_bytes())
 add('rx-boot_app0.bin',(Path.home()/'.platformio/packages/framework-arduinoespressif32/tools/partitions/boot_app0.bin').read_bytes())
 for name in ('profile_four_tx_fixture.py','profile_four_tx_deploy.py','profile_four_tx.py',
-             'profile_switch.py','profile_switch_channels.py','profile_switch_packets.py',
+             'profile_switch.py','serial_session.py','profile_switch_channels.py','profile_switch_packets.py',
              'profile_switch_sweep.py'):
     add(name,(ROOT/'tools/hil'/name).read_bytes())
 for name in ('profile_fixed_tx.cpp','profile_fixed_tx.ini','profile_switch.cpp','profile_switch_channels.h',

tools/hil/profile_four_tx_repeat_run.py

@@ -5,6 +5,7 @@ is the reviewed collector; dependencies and image provenance are hash-checked
 against the previous bench bundle. No flashing or hub reset is performed.
 """
 import base64
+import ast
 import hashlib
 import json
 from pathlib import Path
@@ -24,6 +25,11 @@ if subprocess.run(['systemctl','is-active','--quiet','meshcore-memory-soak.servi
     raise RuntimeError('V4 soak active; do not interrupt another test')
 dependencies=('profile_four_tx_fixture.py','profile_switch.py','profile_switch_channels.py',
               'profile_switch_packets.py','profile_switch_sweep.py')
+# Copy the exact dependency closure of the immutable previous snapshot.
+if any((isinstance(node,ast.ImportFrom) and node.module=='serial_session')
+       or (isinstance(node,ast.Import) and any(alias.name=='serial_session' for alias in node.names))
+       for node in ast.walk(ast.parse((previous/'profile_switch.py').read_bytes()))):
+    dependencies+=('serial_session.py',)
 files={name:(previous/name).read_bytes() for name in dependencies}
 for name,data in files.items():
     if hashlib.sha256(data).hexdigest()!=manifest['files'][name]['sha256']:

tools/hil/profile_pair.py

@@ -14,7 +14,7 @@ import time
 import serial
 from serial.tools import list_ports
 from profile_four_tx_fixture import TXS,RX,resolve
-from profile_switch import configure_session,exchange,read_packet_response
+from profile_switch import open_session,exchange,read_packet_response
 
 TARGETS=(TXS[0],TXS[3])
 PLAN=({'sf':7,'bw_khz':62.5,'freq_khz':909500},
@@ -28,7 +28,7 @@ def require(row,expected,label):
 
 def open_radio(target,pair=True):
     port=serial.Serial();port.port=resolve(target);port.baudrate=115200
-    port.timeout=.2;port.write_timeout=2;configure_session(port);port.open()
+    port.timeout=.2;port.write_timeout=2;open_session(port)
     try:
         port.reset_input_buffer()
         info=exchange(port,'info','board',5)

tools/hil/profile_pair_4p6_manage.py

@@ -37,7 +37,7 @@ if args.action=='prepare':
         manifest['files'][name]={'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest()}
     for name in ('firmware.bin','partitions.bin'):add('rx-'+name,root/'.pio/build/profile_four_tx_v4_rx'/name)
     for name in ('profile_pair.py','profile_pair_4p6_collect.py','profile_pair_run.py','profile_pair_4p6_deploy.py',
-                 'profile_switch.py','profile_four_tx_fixture.py','ProfilePairPlan.h','ProfileSwitchUsb.h',
+                 'profile_switch.py','serial_session.py','profile_four_tx_fixture.py','ProfilePairPlan.h','ProfileSwitchUsb.h',
                  'profile_switch.cpp','profile_switch.ini','profile_switch_channels.h','profile_switch_experiments.h',
                  'profile_stationary_baseline.h','ProfileChannelVisitClock.h','ProfileChannelTrace.h','ProfileFrequencyOffset.h'):
         add(name,root/'tools/hil'/name)

tools/hil/profile_pair_launch.py

@@ -1,5 +1,6 @@
 """Launch the bounded pair collector only once; inspect status separately."""
 import hashlib
+import ast
 import json
 from pathlib import Path
 import subprocess
@@ -7,7 +8,14 @@ root=Path.home()/'hwtest/runs/pair-sf8-32-20260914'
 if (root/'launch.json').exists() or (root/'results.json').exists():raise RuntimeError('Already launched')
 if not json.loads((root/'deployment.json').read_text()).get('complete'):raise RuntimeError('Deployment incomplete')
 manifest=json.loads((root/'manifest.json').read_text())
-for name in ('profile_pair.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py'):
+dependencies=('profile_pair.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py')
+# Historical bundles remain immutable. Require the helper only when their
+# own shipped profile_switch imports it, then verify its recorded hash too.
+if any((isinstance(node,ast.ImportFrom) and node.module=='serial_session')
+       or (isinstance(node,ast.Import) and any(alias.name=='serial_session' for alias in node.names))
+       for node in ast.walk(ast.parse((root/'profile_switch.py').read_bytes()))):
+    dependencies+=('serial_session.py',)
+for name in dependencies:
     if hashlib.sha256((root/name).read_bytes()).hexdigest()!=manifest['files'][name]['sha256']:
         raise RuntimeError('Collector/dependency changed')
 with (root/'collector.log').open('x') as log:

tools/hil/profile_pair_package.py

@@ -30,7 +30,7 @@ for target in (TXS[0],TXS[3]):
     add(target['env']+'.zip',data)
     manifest['files'][target['env']+'.zip'].update(app_base=min(addresses),image_end=max(addresses)+1)
 for name in ('firmware.bin','partitions.bin'):add('rx-'+name,(root/'.pio/build'/RX['env']/name).read_bytes())
-for name in ('profile_pair.py','profile_pair_deploy.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py',
+for name in ('profile_pair.py','profile_pair_deploy.py','profile_pair_run.py','profile_switch.py','serial_session.py','profile_four_tx_fixture.py',
              'ProfilePairPlan.h','ProfileSwitchUsb.h','profile_fixed_tx.cpp','profile_fixed_tx.ini',
              'profile_switch.cpp','profile_switch.ini','profile_switch_channels.h','profile_switch_experiments.h',
              'profile_stationary_baseline.h','ProfileChannelVisitClock.h','ProfileChannelTrace.h','ProfileFrequencyOffset.h'):

tools/hil/profile_switch.py

@@ -10,6 +10,7 @@ import time
 
 import serial
 from serial.tools import list_ports
+from serial_session import open_configured_session
 
 _run_sequences = itertools.count(secrets.randbelow(0x3fffffff) + 1)
 
@@ -24,6 +25,12 @@ def configure_session(port):
                    for info in list_ports.comports())
 
 
+def open_session(port):
+    """Open a configured native session or a bridge without pulsing reset."""
+    configure_session(port)
+    open_configured_session(port)
+
+
 def read_response(port, key, timeout, expected=None):
     deadline = time.monotonic() + timeout
     pending = bytearray()
@@ -113,9 +120,8 @@ def main():
     port.baudrate = 115200
     port.timeout = 0.25
     port.write_timeout = 2
-    configure_session(port)
     try:
-        port.open()
+        open_session(port)
         port.reset_input_buffer()
         info = exchange(port, "info", "bench", 5)
         if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"):

tools/hil/profile_switch_dwell.py

@@ -11,7 +11,7 @@ import sys
 import time
 
 import serial
-from profile_switch import configure_session
+from profile_switch import open_session
 
 
 DWELLS = tuple(value / 10 for value in range(41, 82, 5))
@@ -49,9 +49,8 @@ def reboot_boards(names):
     for name in names:
         port = serial.Serial()
         port.port, port.baudrate, port.timeout, port.write_timeout = name, 115200, .25, 2
-        configure_session(port)
         try:
-            port.open()
+            open_session(port)
             port.write(b"reboot\n")
             port.flush()
         finally:

tools/hil/profile_switch_packets.py

@@ -6,7 +6,7 @@ import secrets
 from pathlib import Path
 
 import serial
-from profile_switch import exchange, exchange_ready, read_packet_response, configure_session
+from profile_switch import exchange, exchange_ready, read_packet_response, open_session
 from profile_switch_sweep import CASES, BULK_CASES, PRODUCTION_CASES
 
 
@@ -14,8 +14,7 @@ def open_port(name):
     port = serial.Serial()
     port.port, port.baudrate, port.timeout = name, 115200, 0.25
     port.write_timeout = 2
-    configure_session(port)
-    port.open()
+    open_session(port)
     port.reset_input_buffer()
     info = exchange(port, "info", "bench", 5)
     if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"):

tools/hil/profile_switch_sweep.py

@@ -3,7 +3,7 @@ import argparse
 import json
 from pathlib import Path
 import serial
-from profile_switch import exchange, exchange_ready, configure_session
+from profile_switch import exchange, exchange_ready, open_session
 
 CASES = [
     ("production_batch", 0, 2), ("manual_full_control", 128, 2),
@@ -45,9 +45,8 @@ def main():
     port = serial.Serial()
     port.port, port.baudrate, port.timeout = args.port, 115200, 0.25
     port.write_timeout = 2
-    configure_session(port)
     try:
-        port.open()
+        open_session(port)
         port.reset_input_buffer()
         info = exchange(port, "info", "bench", 5)
         if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"):

tools/hil/s3_memory_soak.py

@@ -16,6 +16,7 @@ import socketserver
 import subprocess
 import threading
 import time
+from serial_session import open_configured_session
 
 
 class Device:
@@ -68,7 +69,7 @@ class Device:
             # Keep RTS low; a held handle avoids repeated reset transitions.
             stream.dtr = os.name == 'nt'
             stream.rts = False
-            stream.open()
+            open_configured_session(stream)
             self.stream = stream
             self.read(time.monotonic()+20)
             self.write(b'\r')

tools/hil/serial_session.py

@@ -0,0 +1,28 @@
+"""Open an already configured serial session without a bridge reset pulse."""
+import os
+
+
+def open_configured_session(port):
+    # POSIX pySerial normally applies DTR before RTS when opening. Some
+    # bridge drivers initially assert both lines; clearing DTR first pulses
+    # ESP32 EN. Defer DTR until RTS is released. Windows configures both in
+    # one DCB operation and uses dsrdtr for real handshaking, so leave it alone.
+    # The caller selects DTR and supplies the already configured port; no USB
+    # inventory or pySerial import is needed here.
+    defer_dtr = os.name == "posix" and not port.dtr
+    previous_dsrdtr = port.dsrdtr
+    try:
+        if defer_dtr:
+            port.dsrdtr = True
+        port.open()
+        if defer_dtr:
+            port.rts = False
+            port.dtr = False
+            port.dsrdtr = previous_dsrdtr
+    except BaseException:
+        try:
+            port.close()
+        finally:
+            if defer_dtr:
+                port.dsrdtr = previous_dsrdtr
+        raise

tools/hil/test_contact_cache_serial_stress.py

@@ -0,0 +1,63 @@
+"""Offline checks for the contact cache collector's serial opening boundary."""
+import os
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+
+import contact_cache_serial_stress as hil
+
+
+class ContactCacheOpenTests(unittest.TestCase):
+    @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend")
+    def test_actual_posix_open_avoids_reset_and_keeps_requested_session(self):
+        from test_profile_switch_host import PosixModemSerial
+
+        class BeforeProtocol(Exception):
+            pass
+
+        master, slave = os.openpty()
+        try:
+            for vid, expected in ((0x10C4, False), (0x303A, True)):
+                with self.subTest(vid=vid):
+                    ports = []
+                    def factory(*args, **kwargs):
+                        port = PosixModemSerial(*args, **kwargs)
+                        ports.append(port)
+                        return port
+
+                    with mock.patch.object(hil.serial, "Serial", side_effect=factory), mock.patch.object(
+                            hil.time, "sleep", side_effect=BeforeProtocol), mock.patch(
+                            "profile_switch.list_ports.comports", return_value=[
+                                SimpleNamespace(device=os.ttyname(slave), vid=vid)]):
+                        # Stop at the existing initial settle, before any radio
+                        # or protocol request. Exercise the actual collector open.
+                        with self.assertRaises(BeforeProtocol):
+                            hil.run(SimpleNamespace(port=os.ttyname(slave)))
+                    self.assertEqual(len(ports), 1)
+                    port = ports[0]
+                    self.assertFalse(port.has_reset_transition(), port.modem_edges)
+                    self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False})
+                    self.assertFalse(port.dsrdtr)
+                    self.assertFalse(port.is_open)
+        finally:
+            os.close(master)
+            os.close(slave)
+
+    def test_collector_open_failure_closes_and_restores_flow(self):
+        from test_profile_switch_host import FailingSessionPort
+
+        for phase in ("open", "rts", "dtr", "restore"):
+            with self.subTest(phase=phase):
+                port = FailingSessionPort(phase)
+                with mock.patch.object(hil.serial, "Serial", return_value=port), mock.patch(
+                        "serial_session.os.name", "posix"), mock.patch(
+                        "profile_switch.list_ports.comports", return_value=[]):
+                    with self.assertRaisesRegex(RuntimeError, phase + " failed"):
+                        hil.run(SimpleNamespace(port="test"))
+                self.assertFalse(port.is_open)
+                self.assertFalse(port.dsrdtr)
+                self.assertEqual(port.close_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/hil/test_esp32_companion_serial_stress.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 from contextlib import redirect_stdout
 import io
 import json
+import os
 import struct
 import sys
 import types
@@ -214,18 +215,42 @@ class PayloadValidationTest(unittest.TestCase):
 
 
 class StressModesTest(unittest.TestCase):
+    @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend")
+    def test_actual_posix_open_avoids_companion_reset(self) -> None:
+        try:
+            import serial
+            from test_profile_switch_host import PosixModemSerial
+        except ImportError:
+            self.skipTest("pySerial is required for the real open-order regression")
+        master, slave = os.openpty()
+        try:
+            with mock.patch.object(serial, "Serial", PosixModemSerial):
+                port = hil._make_pyserial_factory()(fast_config(port=os.ttyname(slave)))
+            try:
+                self.assertFalse(port.has_reset_transition(), port.modem_edges)
+                self.assertEqual(port.modem_lines, {"dtr": False, "rts": False})
+                self.assertFalse(port.dsrdtr)
+            finally:
+                port.close()
+        finally:
+            os.close(master)
+            os.close(slave)
+
     def test_selected_control_lines_are_set_while_port_is_closed(self) -> None:
         class Serial:
             def __init__(self):
                 self.is_open = False
+                self.dsrdtr = False
+                self.events = []
 
             def __setattr__(self, name, value):
                 if name in ("dtr", "rts"):
-                    assert not self.is_open
+                    self.events.append((name, value))
                 object.__setattr__(self, name, value)
 
             def open(self):
                 assert self.rts is False
+                self.events.append(("open", self.dtr))
                 self.is_open = True
 
         with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=Serial)}):
@@ -235,6 +260,43 @@ class StressModesTest(unittest.TestCase):
                     port = factory(fast_config(dtr=dtr))
                     self.assertIs(port.dtr, dtr)
                     self.assertTrue(port.is_open)
+                    opened = port.events.index(("open", dtr))
+                    self.assertLess(port.events.index(("dtr", dtr)), opened)
+                    self.assertLess(port.events.index(("rts", False)), opened)
+                    self.assertFalse(port.dsrdtr)
+
+    def test_other_backend_keeps_manual_companion_lines_without_dsr_handoff(self) -> None:
+        from test_profile_switch_host import SessionPort
+
+        class OtherBackendPort(SessionPort):
+            def __init__(self):
+                super().__init__("test")
+
+            def __setattr__(self, name, value):
+                if name == "dsrdtr" and value:
+                    raise AssertionError("Other backend entered DSR handshaking")
+                super().__setattr__(name, value)
+
+        with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=OtherBackendPort)}), mock.patch(
+                "serial_session.os.name", "nt"):
+            port = hil._make_pyserial_factory()(fast_config())
+        self.assertTrue(port.is_open)
+        self.assertEqual(port.opened_control_lines, (False, False))
+        self.assertFalse(port.dsrdtr)
+
+    def test_factory_open_or_release_failure_closes_and_restores_flow(self) -> None:
+        from test_profile_switch_host import FailingSessionPort
+
+        for phase in ("open", "rts", "dtr", "restore"):
+            with self.subTest(phase=phase):
+                port = FailingSessionPort(phase)
+                with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=lambda: port)}), mock.patch(
+                        "serial_session.os.name", "posix"):
+                    with self.assertRaisesRegex(RuntimeError, phase + " failed"):
+                        hil._make_pyserial_factory()(fast_config())
+                self.assertFalse(port.is_open)
+                self.assertFalse(port.dsrdtr)
+                self.assertEqual(port.close_count, 1)
 
     def test_dtr_cli_opt_in_preserves_default_and_rejects_non_boolean_config(self) -> None:
         parser = hil.build_argument_parser()

tools/hil/test_fem_noise_floor_test.py

@@ -4,8 +4,11 @@
 from __future__ import annotations
 
 import struct
+import os
 import time
 import unittest
+from unittest import mock
+from types import SimpleNamespace
 
 import esp32_companion_serial_stress as companion
 import fem_noise_floor_test as hil
@@ -62,17 +65,71 @@ def fast_config(**overrides):
 
 
 class PayloadValidationTest(unittest.TestCase):
+    @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend")
+    def test_actual_posix_open_preserves_selected_fem_dtr_without_reset(self):
+        try:
+            import serial
+            from test_profile_switch_host import PosixModemSerial
+        except ImportError:
+            self.skipTest("pySerial is required for the real open-order regression")
+        master, slave = os.openpty()
+        try:
+            for transport_type in (hil.CompanionFemTransport, hil.CliFemTransport):
+                for selected, vid, expected in (("off", 0x303A, False), ("on", 0x10C4, True),
+                                                ("auto", 0x10C4, False), ("auto", 0x1A86, False),
+                                                ("auto", 0x303A, True), ("auto", 0x2886, True),
+                                                ("auto", 0x239A, True)):
+                    with self.subTest(transport=transport_type.__name__, dtr=selected, vid=vid):
+                        transport = transport_type(fast_config(
+                            port=os.ttyname(slave), dtr=selected, open_delay=0))
+                        try:
+                            with mock.patch.object(serial, "Serial", PosixModemSerial), mock.patch(
+                                    "profile_switch.list_ports.comports", return_value=[
+                                        SimpleNamespace(device=os.ttyname(slave), vid=vid)]):
+                                transport.open()
+                            port = transport.port
+                            self.assertFalse(port.has_reset_transition(), port.modem_edges)
+                            self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False})
+                            self.assertFalse(port.dsrdtr)
+                        finally:
+                            transport.close()
+        finally:
+            os.close(master)
+            os.close(slave)
+
+    def test_fem_open_or_release_failure_closes_unpublished_port(self):
+        from test_profile_switch_host import FailingSessionPort
+        import serial
+
+        for transport_type in (hil.CompanionFemTransport, hil.CliFemTransport):
+            for phase in ("open", "rts", "dtr", "restore"):
+                with self.subTest(transport=transport_type.__name__, phase=phase):
+                    port = FailingSessionPort(phase)
+                    transport = transport_type(fast_config(dtr="off", open_delay=0))
+                    with mock.patch.object(serial, "Serial", return_value=port), mock.patch(
+                            "serial_session.os.name", "posix"):
+                        with self.assertRaisesRegex(RuntimeError, phase + " failed"):
+                            transport.open()
+                    self.assertIsNone(transport.port)
+                    self.assertFalse(port.is_open)
+                    self.assertFalse(port.dsrdtr)
+                    self.assertEqual(port.close_count, 1)
+
     def test_automatic_dtr_matches_transport_session_requirements(self):
-        self.assertTrue(hil._resolve_dtr(fast_config(protocol="cli")))
-        self.assertTrue(hil._resolve_dtr(fast_config(
-            port="/dev/serial/by-id/usb-Heltec_HT-n5262G_test-if00",
-        )))
-        self.assertFalse(hil._resolve_dtr(fast_config(
-            port="/dev/serial/by-id/usb-Espressif_USB_JTAG_test-if00",
-        )))
-        self.assertFalse(hil._resolve_dtr(fast_config(
-            protocol="cli", dtr="off",
-        )))
+        from test_profile_switch_host import SessionPort
+
+        for protocol in ("cli", "companion"):
+            for vid, expected in ((0x303A, True), (0x2886, True), (0x239A, True),
+                                  (0x10C4, False), (0x1A86, False), (None, False)):
+                with self.subTest(protocol=protocol, vid=vid):
+                    port = SessionPort("test")
+                    with mock.patch("profile_switch.list_ports.comports", return_value=[
+                            SimpleNamespace(device="test", vid=vid)]):
+                        self.assertEqual(hil._resolve_dtr(fast_config(protocol=protocol), port), expected)
+        with mock.patch("profile_switch.list_ports.comports") as inventory:
+            for selected, expected in (("on", True), ("off", False)):
+                self.assertEqual(hil._resolve_dtr(fast_config(dtr=selected), SessionPort("test")), expected)
+            inventory.assert_not_called()
 
     def test_fem_get_and_set_payloads(self):
         self.assertTrue(hil.validate_fem_get_response(b"\x00\x01")["enabled"])

tools/hil/test_profile_four_tx.py

@@ -1,6 +1,7 @@
 import ast
 import copy
 import json
+import importlib.util
 from pathlib import Path
 import re
 import subprocess
@@ -15,6 +16,44 @@ from profile_four_tx import check_scan,check_sent,parse_args
 from profile_four_tx_fixture import TXS,RX,resolve
 
 class FourTransmitterTests(unittest.TestCase):
+    def test_current_collector_opens_with_legacy_and_current_native_dependencies(self):
+        from test_profile_switch_host import SessionPort
+
+        class NativePort(SessionPort):
+            def __init__(self):
+                super().__init__('test')
+
+            def reset_input_buffer(self):
+                pass
+
+        def legacy_configure(port):
+            port.dtr=True
+            port.rts=False
+
+        # The repeat harness injects today's collector into an immutable old
+        # bundle. That native-only snapshot exports configure_session but not
+        # open_session. Exercise actual module import and its opening function.
+        legacy=SimpleNamespace(configure_session=legacy_configure,exchange=lambda *args:None,
+                               exchange_ready=lambda *args:None,read_packet_response=lambda *args:None)
+        current=sys.modules['profile_switch']
+        path=Path(__file__).with_name('profile_four_tx.py')
+        for dependency in (legacy,current):
+            with self.subTest(snapshot='legacy' if dependency is legacy else 'current'):
+                port=NativePort()
+                with patch.dict(sys.modules,{'profile_switch':dependency}), patch.object(
+                        dependency,'exchange',return_value={'bench':'production-profile-switch-v8','ready':True}), patch(
+                        'serial.Serial',return_value=port), patch('serial.tools.list_ports.comports',return_value=[
+                            SimpleNamespace(device='/dev/native_fixture',vid=0x303A)]):
+                    spec=importlib.util.spec_from_file_location('four_tx_repeat_import_test',path)
+                    collector=importlib.util.module_from_spec(spec)
+                    spec.loader.exec_module(collector)
+                    with patch.object(collector,'resolve',return_value='/dev/native_fixture'):
+                        opened,info=collector.open_radio(RX)
+                self.assertIs(opened,port)
+                self.assertTrue(info['ready'])
+                self.assertEqual(port.opened_control_lines,(True,False))
+                port.close()
+
     def test_dwell_override_changes_only_requested_visit_time(self):
         default=parse_args([])
         self.assertEqual((default.dwell_symbols,default.dwell_us),(5.1,41780))

tools/hil/test_profile_switch_host.py

@@ -1,10 +1,44 @@
 """Regression tests for the serial HIL result reader."""
 import unittest
+import os
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
 from types import SimpleNamespace
-from profile_switch import read_response, exchange_ready, configure_session, exchange
+from profile_switch import read_response, exchange_ready, configure_session, exchange, open_session
+
+if os.name == "posix":
+    import serial.serialposix
+
+    class PosixModemSerial(serial.serialposix.Serial):
+        """Actual pySerial on a PTY, with only modem ioctls simulated.
+
+        The CP2102 kernel-open state observed in HIL asserts both lines. The
+        board's reset circuit is driven when DTR is cleared before RTS.
+        """
+        def __init__(self, *args, **kwargs):
+            self.modem_lines = {"dtr": True, "rts": True}
+            self.modem_edges = []
+            self.flow_at_reconfigure = []
+            super().__init__(*args, **kwargs)
+
+        def _reconfigure_port(self, force_update=False):
+            self.flow_at_reconfigure.append(self.dsrdtr)
+            super()._reconfigure_port(force_update)
+
+        def _set_modem_line(self, name, value):
+            self.modem_lines[name] = value
+            self.modem_edges.append((name, value, self.modem_lines.copy()))
+
+        def _update_dtr_state(self):
+            self._set_modem_line("dtr", self._dtr_state)
+
+        def _update_rts_state(self):
+            self._set_modem_line("rts", self._rts_state)
+
+        def has_reset_transition(self):
+            return any(not lines["dtr"] and lines["rts"]
+                       for _, _, lines in self.modem_edges)
 
 
 class Port:
@@ -27,14 +61,106 @@ class SessionPort:
         self.is_open = False
         self.dtr = True
         self.rts = True
+        self.dsrdtr = False
         self.opened_control_lines = None
+        self.close_count = 0
 
     def open(self):
         self.opened_control_lines = (self.dtr, self.rts)
         self.is_open = True
 
+    def close(self):
+        self.close_count += 1
+        self.is_open = False
+
+
+class FailingSessionPort(SessionPort):
+    def __init__(self, phase):
+        super().__init__("test")
+        self.failure_phase = phase
+
+    def __setattr__(self, name, value):
+        if getattr(self, "is_open", False):
+            phase = getattr(self, "failure_phase", None)
+            if name == phase or (phase == "restore" and name == "dsrdtr" and not value):
+                raise RuntimeError(phase + " failed")
+        object.__setattr__(self, name, value)
+
+    def open(self):
+        if self.failure_phase == "open":
+            raise RuntimeError("open failed")
+        super().open()
+
 
 class ResultReaderTests(unittest.TestCase):
+    @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend")
+    def test_actual_posix_open_avoids_reset_and_preserves_native_dtr(self):
+        master, slave = os.openpty()
+        try:
+            device = os.ttyname(slave)
+            for vid, expected in ((0x10C4, False), (0x1A86, False),
+                                  (0x303A, True), (0x2886, True), (0x239A, True)):
+                with self.subTest(vid=vid):
+                    port = PosixModemSerial()
+                    port.port = device
+                    try:
+                        with patch("profile_switch.list_ports.comports", return_value=[
+                                SimpleNamespace(device=device, vid=vid)]):
+                            open_session(port)
+                        self.assertFalse(port.has_reset_transition(), port.modem_edges)
+                        self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False})
+                        self.assertFalse(port.dsrdtr)
+                        if expected:
+                            self.assertNotIn(True, port.flow_at_reconfigure,
+                                             "Native USB entered the bridge handoff")
+                    finally:
+                        port.close()
+        finally:
+            os.close(master)
+            os.close(slave)
+
+    @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend")
+    def test_actual_posix_old_open_order_is_a_reset_negative_control(self):
+        master, slave = os.openpty()
+        port = PosixModemSerial()
+        port.port = os.ttyname(slave)
+        try:
+            with patch("profile_switch.list_ports.comports", return_value=[]):
+                configure_session(port)
+            port.open()  # Former caller sequence, using actual pySerial open.
+            self.assertTrue(port.has_reset_transition(), port.modem_edges)
+        finally:
+            port.close()
+            os.close(master)
+            os.close(slave)
+
+    def test_other_backend_does_not_use_dsr_handshaking(self):
+        class OtherBackendPort(SessionPort):
+            def __setattr__(self, name, value):
+                if name == "dsrdtr" and value:
+                    raise AssertionError("Other backend entered DSR handshaking")
+                super().__setattr__(name, value)
+
+        port = OtherBackendPort("test")
+        with patch("profile_switch.os.name", "nt"), patch(
+                "profile_switch.list_ports.comports", return_value=[]):
+            open_session(port)
+        self.assertTrue(port.is_open)
+        self.assertEqual(port.opened_control_lines, (False, False))
+        self.assertFalse(port.dsrdtr)
+
+    def test_failed_open_or_line_release_closes_and_restores_flow(self):
+        for phase in ("open", "rts", "dtr", "restore"):
+            with self.subTest(phase=phase):
+                port = FailingSessionPort(phase)
+                with patch("profile_switch.os.name", "posix"), patch(
+                        "profile_switch.list_ports.comports", return_value=[]):
+                    with self.assertRaisesRegex(RuntimeError, phase + " failed"):
+                        open_session(port)
+                self.assertFalse(port.is_open)
+                self.assertFalse(port.dsrdtr)
+                self.assertEqual(port.close_count, 1)
+
     def test_timing_runs_get_correlated_cached_replies(self):
         port = Port([])
         with patch("profile_switch._run_sequences", iter([99])), patch(

tools/hil/test_s3_memory_soak.py

@@ -12,6 +12,28 @@ soak=importlib.util.module_from_spec(spec)
 spec.loader.exec_module(soak)
 
 class TelemetryTests(unittest.TestCase):
+    @unittest.skipUnless(os.name == 'posix', 'Requires the POSIX pySerial backend')
+    def test_actual_posix_soak_open_avoids_board_reset(self):
+        import serial
+        from test_profile_switch_host import PosixModemSerial
+
+        master,slave=os.openpty()
+        device=soak.Device('test',os.ttyname(slave),'test')
+        try:
+            with patch.object(serial,'Serial',PosixModemSerial), patch.object(
+                    device,'read',return_value=''), patch.object(
+                    device,'_command',return_value='test'):
+                device.connect()
+            port=device.stream
+            self.assertFalse(port.has_reset_transition(),port.modem_edges)
+            self.assertEqual(port.modem_lines,{'dtr':False,'rts':False})
+            self.assertFalse(port.dsrdtr)
+            self.assertEqual(device.connections,1)
+        finally:
+            device.close()
+            os.close(master)
+            os.close(slave)
+
     def test_diagnostics_without_key_override_need_no_private_header(self):
         # Preprocess a clean copy: the developer's ignored key header must not
         # satisfy an accidental unconditional include during this check.

tools/hil/test_serial_session.py

@@ -0,0 +1,62 @@
+"""Verify the shared serial opener keeps optional SDK dependencies optional."""
+from pathlib import Path
+import ast
+import hashlib
+import subprocess
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+
+
+class SerialSessionImportTests(unittest.TestCase):
+    def test_immutable_collectors_require_only_their_shipped_dependencies(self):
+        base = Path(__file__).parent
+        fixture_names = ("profile_pair.py", "profile_pair_run.py", "profile_four_tx_fixture.py",
+                         "profile_switch.py", "profile_switch_channels.py",
+                         "profile_switch_packets.py", "profile_switch_sweep.py")
+        for script, end in (("profile_pair_launch.py", "with (root/'collector.log')"),
+                            ("profile_four_tx_repeat_run.py", "code=base64.b64decode")):
+            source = (base / script).read_text()
+            # Execute the actual dependency-selection and hash-validation
+            # block, without launch, subprocess, firmware or output operations.
+            block = source[source.index("dependencies="):source.index(end)]
+            for current, helper in ((False, "absent"), (True, "valid"),
+                                    (True, "absent"), (True, "tampered")):
+                with self.subTest(script=script, current=current, helper=helper), TemporaryDirectory() as tmp:
+                    root = Path(tmp)
+                    for name in fixture_names:
+                        (root / name).write_text("")
+                    (root / "profile_switch.py").write_text(
+                        "from serial_session import open_configured_session\n" if current else "# Legacy snapshot\n")
+                    if helper != "absent":
+                        (root / "serial_session.py").write_text("# Shipped helper\n")
+                    manifest = {"files": {file.name: {"sha256": hashlib.sha256(file.read_bytes()).hexdigest()}
+                                          for file in root.iterdir()}}
+                    if helper == "tampered":
+                        (root / "serial_session.py").write_text("# Changed after packaging\n")
+                    namespace = dict(root=root, previous=root, manifest=manifest,
+                                     ast=ast, hashlib=hashlib)
+                    if current and helper == "absent":
+                        with self.assertRaises(FileNotFoundError):
+                            exec(block, namespace)
+                    elif helper == "tampered":
+                        with self.assertRaises(RuntimeError):
+                            exec(block, namespace)
+                    else:
+                        exec(block, namespace)
+                        self.assertEqual("serial_session.py" in namespace["dependencies"], current)
+
+    def test_helpers_import_with_only_the_standard_library(self):
+        # -S removes installed site packages. The companion/FEM/soak modules
+        # must remain importable until their caller actually requests serial.
+        program = ("import sys; sys.path.insert(0, " + repr(str(Path(__file__).parent)) + "); "
+                   "import serial_session, esp32_companion_serial_stress, "
+                   "fem_noise_floor_test, s3_memory_soak; "
+                   "assert 'serial' not in sys.modules")
+        result = subprocess.run([sys.executable, "-I", "-S", "-c", program],
+                                text=True, capture_output=True, timeout=10)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()