Run Honda native lifecycle tests under both CI runners

openpilot/selfdrive/car/tests/test_honda_aol.py

@@ -1,6 +1,5 @@
 """Honda AOL host qualification and serialized integration tests."""
 
-import time
 import unittest
 from unittest import mock
 
@@ -13,10 +12,6 @@ from opendbc.car.honda.interface import CarInterface
 from opendbc.car.honda.values import CAR, CarControllerParams, HondaFlags
 from openpilot.common.params import Params
 from openpilot.selfdrive.car.card import Car
-from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
-from openpilot.selfdrive.controls.controlsd import Controls
-from openpilot.starpilot.aol.intent import read_settings
-from openpilot.starpilot.aol.wire import SafetyState, decode_intent, encode_safety
 from openpilot.starpilot.car.honda.aol import CLASSIC_BOSCH_AOL_CARS, qualified_honda
 
 
@@ -141,7 +136,11 @@ class HondaFamilyQualificationTests(unittest.TestCase):
           expected = alpha_long or candidate != CAR.HONDA_ACCORD
           self.assertEqual(card.aol_qualified, expected)
           self.assertEqual(bool(card.CP.safetyConfigs[0].safetyParam & 32), alpha_long)
-          experience = 32 if expected and not alpha_long else 0
+          stock_owned = candidate != CAR.HONDA_ACCORD
+          experience = 32 if expected and stock_owned else 0
+          if alpha_long and stock_owned:
+            word = 2211 if candidate == CAR.ACURA_RDX_3G else 2082 | int(bool(card.CP.flags & HondaFlags.BOSCH_ALT_BRAKE))
+            self.assertEqual(card.CP.safetyConfigs[0].safetyParam, word)
           self.assertEqual(card.CP.alternativeExperience, experience)
           with car.CarParams.from_bytes(params.get('CarParams')) as published:
             self.assertEqual(bool(published.safetyConfigs[0].safetyParam & 32), alpha_long)
@@ -149,147 +148,24 @@ class HondaFamilyQualificationTests(unittest.TestCase):
 
 
 class HondaHostTests(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls):
+    super().setUpClass()
+    from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair
+    cls.honda_native_origins = cls.enterClassContext(honda_native_pair())
+
   def test_real_car_params_selfdrive_controls_chain_fails_closed(self):
-    class FakeHonda:
-      CC = object()
-      CS = object()
-
-      def __init__(self):
-        self.CP = CarInterface.get_params(CAR.HONDA_CIVIC_BOSCH, gen_empty_fingerprint(), [], True, False, False)
-
-      def update(self, _can):
-        state = car_state()
-        state.cruiseState.available = True
-        return state
-
-      def init(self, _cp, _recv, _send):
-        pass
-
-      def apply(self, _control, _now_ns):
-        return car.CarControl.Actuators(), []
-
-    class FakeRadar(RadarInterfaceBase):
-      def update(self, can_packets):
-        return None
-
-    with (OpenpilotPrefix(), mock.patch.dict('os.environ', {'AOL_REPLAY_RUNTIME': '1', 'SIMULATION': '1'}),
-          mock.patch('openpilot.selfdrive.selfdrived.selfdrived.SIMULATION', False)):
-      messaging.reset_context()
-      params = Params()
-      params.put_bool('OpenpilotEnabledToggle', True, block=True)
-      params.put_bool('AlwaysOnLateral', True, block=True)
-      honda = FakeHonda()
-      card = Car(honda, FakeRadar(honda.CP))
-      self.assertEqual(card.CP.safetyConfigs[-1].safetyParam & 0x20, 0x20)
-      with car.CarParams.from_bytes(params.get('CarParams')) as published:
-        self.assertEqual(published.safetyConfigs[-1].safetyParam & 0x20, 0x20)
-      selfdrive = SelfdriveD()  # reads the actual card-published CarParams
-      controls = Controls()    # reads the same serialized CarParams
-      self.assertTrue(selfdrive.aol_replay and controls.aol_replay)
-      can = messaging.PubMaster(['can'])
-      intent = messaging.SubMaster(['aolIntentWire'])
-      can.send('can', messaging.new_message('can', 1))
-      card.step()
-      intent.update(100)
-      self.assertTrue(intent.seen['aolIntentWire'])
-      self.assertTrue(decode_intent(intent['aolIntentWire']).allowedLatch)
-      selfdrive.step()
-      self.assertFalse(selfdrive.initialized)  # Native acknowledgment must precede bootstrap.
-      controls.sm.update(100)
-      cc, _ = controls.state_control()
-      self.assertFalse(cc.latActive or cc.longActive)  # no native capability/ack
-      self.assertEqual(selfdrive.aol_axis_decision.mode, 'off')
-
-      source = messaging.PubMaster(['aolSafetyWire', 'modelV2', 'extrinsicsCalibration', 'driverMonitoringState'])
-      time.sleep(0.05)  # allow native IPC subscriptions to connect before the first publication
-      model = messaging.new_message('modelV2')
-      model.valid = True
-      calibration = messaging.new_message('extrinsicsCalibration')
-      calibration.valid = True
-      calibration.extrinsicsCalibration.calStatus = log.ExtrinsicsCalibration.Status.calibrated
-      dm = messaging.new_message('driverMonitoringState')
-      dm.valid = True
-      def publish_native(*, requested_long=False):
-        native = messaging.new_message('aolSafetyWire', 0)
-        native.valid = True
-        observed = time.monotonic_ns()
-        native.aolSafetyWire = encode_safety(SafetyState(
-          1, True, observed, observed + 200_000_000, int(car.CarParams.SafetyModel.hondaBosch),
-          card.CP.safetyConfigs[-1].safetyParam, True, requested_long, True, requested_long,
-          'test-panda', selfdrive.aol_session_id))
-        source.send('aolSafetyWire', native)
-
-      for _ in range(3):
-        source.send('modelV2', model)
-        model.clear_write_flag()
-        source.send('extrinsicsCalibration', calibration)
-        calibration.clear_write_flag()
-        source.send('driverMonitoringState', dm)
-        dm.clear_write_flag()
-        publish_native()
-        selfdrive.sm.update(100)
-        controls.sm.update(0)
-      selfdrive.initialized = True
-      with mock.patch.object(selfdrive, 'update_events', side_effect=lambda _cs: selfdrive.events.clear()):
-        can.send('can', messaging.new_message('can', 1))
-        card.step()
-        selfdrive.step()
-      self.assertEqual(selfdrive.aol_axis_decision.mode, 'lateralOnly')
-      controls.sm.update(0)
-      cc, _ = controls.state_control()
-      self.assertTrue(cc.latActive)
-      self.assertFalse(cc.longActive)
-      # The native long request may still reflect a previous combined mode;
-      # this must not interrupt unchanged lateral permission.
-      publish_native(requested_long=True)
-      controls.sm.update(0)
-      cc, _ = controls.state_control()
-      self.assertTrue(cc.latActive)
-      self.assertFalse(cc.longActive)
-      # On the reverse transition, an old lateral-only acknowledgment keeps
-      # lateral active while the newly requested longitudinal axis waits.
-      publish_native()
-      axis_msg = messaging.new_message('aolAxisState')
-      axis_msg.valid = True
-      axis = axis_msg.aolAxisState
-      axis.sessionId = selfdrive.aol_session_id
-      axis.qualified = True
-      axis.desiredLateral = True
-      axis.desiredLongitudinal = True
-      axis.lateralActive = True
-      axis.longitudinalActive = False
-      axis.nativeAcknowledged = True
-      axis.observedMonoTime = time.monotonic_ns()
-      axis.validUntilMonoTime = axis.observedMonoTime + 30_000_000
-      selfdrive.pm.send('aolAxisState', axis_msg)
-      controls.sm.update(0)
-      cc, _ = controls.state_control()
-      self.assertTrue(cc.latActive)
-      self.assertFalse(cc.longActive)
-
-      dm.driverMonitoringState.lockout = True
-      source.send('driverMonitoringState', dm)
-      with mock.patch.object(selfdrive, 'update_events', side_effect=lambda _cs: selfdrive.events.clear()):
-        can.send('can', messaging.new_message('can', 1))
-        card.step()
-        selfdrive.step()
-      self.assertEqual(selfdrive.aol_axis_decision.mode, 'off')
-      controls.sm.update(100)
-      cc, _ = controls.state_control()
-      self.assertFalse(cc.latActive or cc.longActive)
-
-      # Saved AOL off retains the negotiated protocol while ending extra lateral autonomy.
-      params.put_bool('AlwaysOnLateral', False, block=True)
-      assert card.aol_card_intent is not None
-      card.aol_card_intent.settings = read_settings(params)
-      selfdrive.aol_settings = read_settings(params)
-      can.send('can', messaging.new_message('can', 1))
-      card.step()
-      selfdrive.step()
-      controls.sm.update(100)
-      cc, _ = controls.state_control()
-      self.assertFalse(cc.latActive or cc.longActive)
-      self.assertTrue(card.aol_qualified)
+    from openpilot.starpilot.aol.tests import test_honda_recovery_lifecycle as current
+    # The source-bound contract starts unarmed. Actual SET/CAN, C++ negotiation,
+    # native USB status and accepted controller TX establish the later grant.
+    # Fresh DM lockout retains an already armed LAT owner under shared policy;
+    # invalid DM, saved OFF, and unacknowledged LONG still fail closed.
+    caller = current.TestHondaCurrentCaller()
+    for scenario in ('driver_lockout', 'settings_off', 'long_reengage'):
+      with self.subTest(scenario=scenario):
+        caller.run_current_caller(
+          candidate=CAR.HONDA_CIVIC_BOSCH, word=2082, alpha=True,
+          host_scenario=scenario, serialized=True)
 
 
 class CardAxisSessionTests(unittest.TestCase):

openpilot/starpilot/aol/tests/honda_native_fixture.py

@@ -1,18 +1,27 @@
 """Compile current native safety and C++ negotiation for Honda caller tests."""
 
+from contextlib import contextmanager
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
 import subprocess
+import hashlib
+import json
+import os
 from pathlib import Path
 
 import capnproto
-import pytest
 
 from opendbc.safety.tests.libsafety import libsafety_py
 
 
-@pytest.fixture(scope='module')
-def honda_native_pair(tmp_path_factory):
+@contextmanager
+def honda_native_pair():
   root = Path(__file__).resolve().parents[4]
-  temporary = tmp_path_factory.mktemp('honda-native')
+  with TemporaryDirectory(prefix='honda-native-') as directory:
+    yield from _compiled_native_pair(root, Path(directory))
+
+
+def _compiled_native_pair(root, temporary):
   library = temporary / 'libsafety.so'
   bridge = temporary / 'pandad-order'
   subprocess.run(
@@ -34,6 +43,9 @@ def honda_native_pair(tmp_path_factory):
       '-fno-sanitize-recover=undefined',
       '-I',
       str(root / 'opendbc_repo'),
+      '-MD',
+      '-MF',
+      str(temporary / 'libsafety.d'),
       '-c',
       str(root / 'opendbc_repo/opendbc/safety/tests/libsafety/safety.c'),
       '-o',
@@ -74,13 +86,29 @@ def honda_native_pair(tmp_path_factory):
     check=True,
     timeout=35,
   )
+  inputs = (temporary / 'libsafety.d').read_text().replace('\\\n', '').split(':', 1)[1].split()
+  bridge_inputs = [root / name for name in ('openpilot/selfdrive/pandad/aol_protocol.h',
+                                           'openpilot/selfdrive/pandad/aol_wire.h',
+                                           'openpilot/starpilot/car/honda/aol_policy.h')]
+  origins = {
+    'root': str(root),
+    'native_library': str(library),
+    'bridge': str(bridge),
+    'inputs': {name: hashlib.sha256(Path(name).read_bytes()).hexdigest() for name in inputs},
+    'library_sha256': hashlib.sha256(library.read_bytes()).hexdigest(),
+    'bridge_inputs': {str(path): hashlib.sha256(path.read_bytes()).hexdigest() for path in bridge_inputs},
+    'bridge_source_sha256': hashlib.sha256(Path(__file__).with_name('honda_pandad_order.cc').read_bytes()).hexdigest(),
+    'schemas': {str(path): hashlib.sha256(path.read_bytes()).hexdigest() for path in generated},
+  }
+  (temporary / 'origins.json').write_text(json.dumps(origins, indent=2) + '\n')
   original = {key: vars(libsafety_py).get(key) for key in ('libsafety', 'libpath')}
-  with pytest.MonkeyPatch.context() as environment:
-    environment.setenv('HONDA_NATIVE_LIBRARY', str(library))
-    environment.setenv('HONDA_PANDAD_BRIDGE', str(bridge))
+  with patch.dict('os.environ', {'HONDA_NATIVE_LIBRARY': str(library), 'HONDA_PANDAD_BRIDGE': str(bridge)}):
     libsafety_py.load(library)
     try:
-      yield
+      assert Path(libsafety_py.libpath).resolve() == library.resolve()
+      assert Path(os.environ['HONDA_NATIVE_LIBRARY']).resolve() == library.resolve()
+      assert Path(os.environ['HONDA_PANDAD_BRIDGE']).resolve() == bridge.resolve()
+      yield origins
     finally:
       for key, value in original.items():
         if value is None:

openpilot/starpilot/aol/tests/honda_selfdrive_step.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 from openpilot.cereal import messaging
 
 
-def selfdrive_step(sd, published, native, previous_controls, cp, safety, now, *, calibration):
+def selfdrive_step(sd, published, native, previous_controls, cp, safety, now, *, calibration, driver_lockout=False, driver_monitoring_valid=True):
   peers = []
   for service in sd.sm.data:
     if service in ('alertDebug', 'lateralManeuverPlan', 'laneChangeAssistWire', 'starpilotLongitudinalPlan'):
@@ -28,6 +28,9 @@ def selfdrive_step(sd, published, native, previous_controls, cp, safety, now, *,
       value[0].alternativeExperience = cp.alternativeExperience
       value[0].controlsAllowed = bool(safety.get_controls_allowed())
       value[0].safetyRxChecksInvalid = not bool(safety.safety_config_valid())
+    elif service == 'driverMonitoringState':
+      value.lockout = driver_lockout
+      event.valid = driver_monitoring_valid
     elif service == 'extrinsicsCalibration':
       value.calStatus, value.rpyCalib = calibration, [0.0, 0.0, 0.0]
     elif service == 'vehicleParameters':

openpilot/starpilot/aol/tests/test_honda_alt_radar_lifecycle.py

@@ -1,10 +1,7 @@
 """Alternate radar stock Honda with current factory/parser/Card/SD/Controls/native."""
-import pytest
 from opendbc.car.honda.values import CAR
 from openpilot.starpilot.aol.tests.test_honda_recovery_lifecycle import TestHondaCurrentCaller as CurrentCaller
-from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair as honda_native_pair
 
-@pytest.mark.usefixtures('honda_native_pair')
 class TestHondaAlternateRadar(CurrentCaller):
   def test_alt_0_eps_reverse_recovery(self):
     self.run_current_caller(candidate=CAR.ACURA_RDX_3G_MMR, word=1, speed_mps=25.)

openpilot/starpilot/aol/tests/test_honda_canfd_lifecycle.py

@@ -1,10 +1,7 @@
 """Stock CAN-FD Honda caller, C++ negotiation, and native TX lifecycle."""
-import pytest
 from opendbc.car.honda.values import CAR
 from openpilot.starpilot.aol.tests.test_honda_recovery_lifecycle import TestHondaCurrentCaller as CurrentCaller
-from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair as honda_native_pair
 
-@pytest.mark.usefixtures('honda_native_pair')
 class TestHondaCanfdCaller(CurrentCaller):
   def test_canfd_0_eps_reverse_recovery(self):
     self.run_current_caller(candidate=CAR.HONDA_CRV_6G, word=16)

openpilot/starpilot/aol/tests/test_honda_classic_alpha_lifecycle.py

@@ -1,10 +1,7 @@
 """Normal-runtime classic Bosch alpha owner, real FW/parser/native TX lifecycle."""
-import pytest
 from opendbc.car.honda.values import CAR
 from openpilot.starpilot.aol.tests.test_honda_recovery_lifecycle import TestHondaCurrentCaller as CurrentCaller
-from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair as honda_native_pair
 
-@pytest.mark.usefixtures('honda_native_pair')
 class TestHondaClassicAlpha(CurrentCaller):
   def test_classic_0_eps_reverse_recovery(self):
     self.run_current_caller(candidate=CAR.HONDA_CIVIC_BOSCH, word=2082, alpha=True)

openpilot/starpilot/aol/tests/test_honda_pedal_lifecycle.py

@@ -1,10 +1,7 @@
 """Nidec pedal actual acquisition/parser/Card/Controls/native recovery."""
-import pytest
 from opendbc.car.honda.values import CAR
 from openpilot.starpilot.aol.tests.test_honda_recovery_lifecycle import TestHondaCurrentCaller as CurrentCaller
-from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair as honda_native_pair
 
-@pytest.mark.usefixtures('honda_native_pair')
 class TestHondaPedal(CurrentCaller):
   def test_pedal_0_eps_reverse_recovery(self):
     self.run_current_caller(candidate=CAR.HONDA_CIVIC, word=544, pedal=True)

openpilot/starpilot/aol/tests/test_honda_recovery_lifecycle.py

@@ -32,15 +32,17 @@ from openpilot.starpilot.aol.tests.honda_can import HondaCan
 from openpilot.starpilot.aol.tests.honda_pandad_order import PandadOrder
 from openpilot.starpilot.aol.tests.honda_selfdrive_step import selfdrive_step
 
-import pytest
-
 from openpilot.starpilot.aol.tests.honda_native_fixture import honda_native_pair as honda_native_pair
 
 TRACE = []
 
 
-@pytest.mark.usefixtures("honda_native_pair")
 class TestHondaCurrentCaller(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls):
+    super().setUpClass()
+    cls.honda_native_origins = cls.enterClassContext(honda_native_pair())
+
   def test_classic_civic_temporary_eps_reverse_return(self):
     self.run_current_caller(candidate=CAR.HONDA_CIVIC_BOSCH, word=0)
 
@@ -56,7 +58,8 @@ class TestHondaCurrentCaller(unittest.TestCase):
   def test_relay_fault_during_recovered_reverse(self):
     self.run_current_caller(candidate=CAR.HONDA_CIVIC_BOSCH, word=0, fault='relay')
 
-  def run_current_caller(self, *, candidate, word, fault=None, alternate_brake=False, alpha=False, speed_mps=12.5, pedal=False):
+  def run_current_caller(self, *, candidate, word, fault=None, alternate_brake=False, alpha=False, speed_mps=12.5,
+                         pedal=False, host_scenario=None, serialized=False):
     safety = libsafety_py.libsafety
     with (
       OpenpilotPrefix(),
@@ -104,6 +107,7 @@ class TestHondaCurrentCaller(unittest.TestCase):
         io.enter_context(patch.object(car_helpers, 'can_fingerprint', return_value=(None, fp)))
         card = Car()
       cp, ci = card.CP, card.CI
+      assert card.aol_card_intent is not None
       self.assertEqual(cp.fingerprintSource, structs.CarParams.FingerprintSource.fw)
       self.assertFalse(cp.fuzzyFingerprint)
       self.assertEqual(cp.carVin, vin)
@@ -117,7 +121,9 @@ class TestHondaCurrentCaller(unittest.TestCase):
       self.assertEqual(alt_brake, alternate_brake or bool(candidate.config.flags & HondaFlags.BOSCH_ALT_BRAKE))
       with structs.CarParams.from_bytes(saved.get('CarParams')) as published:
         self.assertEqual(published.to_dict(), cp.to_dict())
-      sd, controls = SelfdriveD(CP=cp), Controls()
+      sd, controls = SelfdriveD() if serialized else SelfdriveD(CP=cp), Controls()
+      self.assertEqual(sd.CP.to_dict(), cp.to_dict())
+      self.assertEqual(controls.CP.to_dict(), cp.to_dict())
       stream = HondaCan(ci)
       safety.init_tests()
       safety.set_alternative_experience(32)
@@ -143,10 +149,16 @@ class TestHondaCurrentCaller(unittest.TestCase):
       try:
         for tick in range(145):
           now = base + tick * 10_000_000
+          reengage = host_scenario == 'long_reengage' and tick >= 131
+          if host_scenario == 'settings_off' and tick == 135:
+            from openpilot.starpilot.aol.intent import read_settings
+            saved.put_bool('AlwaysOnLateral', False, block=True)
+            card.aol_card_intent.settings = read_settings(saved)
+            sd.aol_settings = read_settings(saved)
           # SET press/release is the real first ordinary native grant, preceding engaged HB.
           incoming = stream.frames(
-            cruise_button=3 if tick == 20 else 0,
-            cruise=int(21 <= tick < 50),
+            cruise_button=3 if tick == 20 or host_scenario == 'long_reengage' and tick == 130 else 0,
+            cruise=int(21 <= tick < 50 or reengage),
             brake=50 <= tick < 55,
             eps=6 if 80 <= tick < 95 else 0,
             gear=2 if 110 <= tick < 125 else None,
@@ -175,7 +187,7 @@ class TestHondaCurrentCaller(unittest.TestCase):
           panda.pandaStates[0].safetyModel = cp.safetyConfigs[0].safetyModel
           panda.pandaStates[0].safetyParam, panda.pandaStates[0].alternativeExperience = word, 32
           panda.pandaStates[0].ignitionLine = True
-          if tick >= 110 and fault:
+          if tick >= 110 and fault in ('relay', 'diagnostic'):
             panda.pandaStates[0].faults = ['relayMalfunction' if fault == 'relay' else 'interruptRateCan2']
           panda.pandaStates[0].controlsAllowed = bool(safety.get_controls_allowed())
           panda.pandaStates[0].safetyRxChecksInvalid = not bool(safety.safety_config_valid())
@@ -245,13 +257,16 @@ class TestHondaCurrentCaller(unittest.TestCase):
           self.assertEqual(cs.gearShifter, structs.CarState.GearShifter.reverse if 110 <= tick < 125 else structs.CarState.GearShifter.drive)
           self.assertEqual(cs.brakePressed, 50 <= tick < 55)
           self.assertTrue(cs.cruiseState.available)
-          self.assertEqual(cs.cruiseState.enabled, 21 <= tick < 50)
+          self.assertEqual(cs.cruiseState.enabled, 21 <= tick < 50 or reengage)
           intent = decode_intent(published['aolIntentWire'].aolIntentWire)
           self.assertIsNotNone(intent)
           # Initial receipt is unavailable until actual negotiator completes its zero acknowledgment.
           absent = messaging.new_message('aolSafetyWire', 0, valid=False, logMonoTime=now)
           current_receipt = receipt if receipt is not None else absent
-          driven_new = selfdrive_step(sd, published, current_receipt, controls_published, cp, safety, now, calibration='calibrated')
+          driven_new = selfdrive_step(
+            sd, published, current_receipt, controls_published, cp, safety, now, calibration='calibrated',
+            driver_lockout=host_scenario == 'driver_lockout' and tick >= 135,
+            driver_monitoring_valid=not (host_scenario == 'driver_lockout' and tick >= 140))
           retained = dict(driven)
           driven.update(driven_new)
           for service, event in retained.items():
@@ -355,7 +370,7 @@ class TestHondaCurrentCaller(unittest.TestCase):
                 self.assertFalse(command.longActive or row['controls'] or row['permission'] & 2, row)
               else:
                 self.assertTrue(accepted or fault == "relay" and tick >= 110, row)
-              if alpha and address == 0x1DF and tick >= 52:
+              if alpha and address == 0x1DF and tick >= 52 and not reengage:
                 gas = int.from_bytes(data[:2], 'big', signed=True)
                 accel = (data[3] << 3) | (data[4] >> 5)
                 accel = accel - 0x800 if accel & 0x400 else accel
@@ -363,7 +378,7 @@ class TestHondaCurrentCaller(unittest.TestCase):
             row['TX'].append(tx)
           row['events'] = [str(e.name) for e in driven['onroadEvents'].onroadEvents]
           row['fault_reason'] = str(driven['aolAxisState'].aolAxisState.faultReason)
-          if tick in (45, 79, 105) or tick == 135 and fault != "relay":
+          if tick in (45, 79, 105) or tick == 135 and fault != "relay" and host_scenario not in ("driver_lockout", "settings_off"):
             self.assertTrue(command.latActive, row)
             self.assertTrue(any(t.get('accepted') and t.get('request') and t.get('torque') for t in row['TX']), row)
             self.assertTrue(intent.allowedLatch, row)
@@ -372,7 +387,36 @@ class TestHondaCurrentCaller(unittest.TestCase):
             self.assertTrue(intent.lateralArmed, row)
             self.assertFalse(command.latActive or command.longActive, row)
             self.assertFalse(row['permission'] & 1, row)
-          if tick >= 55:
+          if tick < 20:
+            self.assertFalse(intent.allowedLatch or command.latActive or command.longActive, row)
+          if host_scenario == "driver_lockout" and 135 <= tick < 140:
+            # Current shared policy retains an already active, source-bound LAT
+            # owner during fresh DM lockout, with LONG inhibited.
+            self.assertTrue(sd.sm['driverMonitoringState'].lockout, row)
+            self.assertTrue(sd.aol_availability_long_inhibit, row)
+            self.assertTrue(command.latActive and row['permission'] & 1, row)
+            self.assertFalse(command.longActive or row['permission'] & 2, row)
+          withdraw = host_scenario == "settings_off" and tick >= 137 or host_scenario == "driver_lockout" and tick >= 142
+          if withdraw:
+            self.assertFalse(command.latActive or command.longActive, row)
+            if tick == 144:
+              if host_scenario == "settings_off":
+                self.assertFalse(row["permission"] & 3, row)
+              else:
+                # A short availability pause retains the native LAT owner.
+                # Host commands still withdraw, including the actual CAN TX.
+                self.assertEqual(row["fault_reason"], "availabilityPause", row)
+                self.assertEqual(row["permission"], 1, row)
+                steer = [tx for tx in row['TX'] if 'torque' in tx]
+                self.assertTrue(steer, row)
+                self.assertTrue(all(tx['accepted'] and tx['torque'] == 0 and not tx['request'] for tx in steer), row)
+          if host_scenario == "long_reengage" and tick == 135:
+            self.assertTrue(command.latActive, row)
+            self.assertFalse(command.longActive or row["permission"] & 2, row)
+          if host_scenario == "long_reengage" and tick == 144:
+            self.assertTrue(command.latActive and command.longActive, row)
+            self.assertEqual(row["permission"], 3, row)
+          if tick >= 55 and not reengage:
             self.assertFalse(command.longActive, row)
           if fault == 'relay' and tick >= 120:
             self.assertFalse(card.aol_card_intent.allowed_latch or intent.lateralArmed, row)

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "efd5f01b16f5629c00da62d3e832b576f70f4a212ffc264ee783c393ed8f5545", "schema_version": 1}
+{"protected_inventory_sha256": "461764647a55c7de9da88172eba75d01ff214e0de95fe9f805940797c7cc63f4", "schema_version": 1}