Preserve armed lateral intent through temporary warnings

opendbc_repo/opendbc/car/mazda/tests/test_native_engagement_callers.py

@@ -90,7 +90,7 @@ def test_eps_recovery_actual_callers(identity, lateral_only, monkeypatch, armed_
 
 
 def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake', cancel_action=3, armed_intent=None,
-                                    lateral_only=False):
+                                    lateral_only=False, warning=None, unbelted=False):
   from openpilot.cereal import messaging
   from openpilot.selfdrive.car import card as card_module
   from openpilot.starpilot.aol.wire import PandaSlot, SafetyState, encode_safety
@@ -107,6 +107,7 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
     now, tick, queued = [10_000_000_000], 0, []
     received, previous_sd = {}, {}
     first_engagement_ns = None
+    belt_off = [False]
     try:
       sd, controls, drive = case._authority_callers(False, cp=card.CP.as_reader().as_builder(), genuine_events=True)
       if scenario in ('cancel', 'long_cancel', 'mapped_cancel', 'temporary', 'startup', 'startup_stopped'):
@@ -114,6 +115,14 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
         sd.aol_settings = read_settings(saved)
         assert sd.aol_settings.cancel_actions == (cancel_action, 0, 0)
         assert card.aol_card_intent.settings == sd.aol_settings
+      soft_warning = [False]
+      original_events = sd.update_events
+      def warning_events(*args, **kwargs):
+        original_events(*args, **kwargs)
+        if soft_warning[0]:
+          from openpilot.selfdrive.selfdrived.events import EventName
+          sd.events.add(getattr(EventName, warning))
+      local_patch.setattr(sd, 'update_events', warning_events)
       assert sd.CP.to_dict() == card.CP.to_dict() == ci.CP.to_dict() == controls.CP.to_dict()
       safety.init_tests()
       safety.set_alternative_experience(int(card.CP.alternativeExperience))
@@ -146,6 +155,9 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
         frames = complete_frames(packer, card.CP, tick, main=main, engaged=engaged, brake=brake, cancel=cancel, reverse=reverse,
                                  speed_kph=speed_kph, permanent=permanent, temporary=temporary)
         frames.append(packer.make_can_msg('STEER', 0, {'STEER_ANGLE': 1.0}))
+        if belt_off[0]:
+          belt = packer.make_can_msg('SEATBELT', 0, {'DRIVER_SEATBELT': 0})
+          frames = [row for row in frames if not (row[0] == belt[0] and row[2] == 0)] + [belt]
         safety.set_timer(now[0] // 1000)
         prior_state = previous_sd.get('selfdriveState')
         prior_axis = previous_sd.get('aolAxisState')
@@ -242,6 +254,10 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
           assert any(frame.address == 0x243 for frame in outgoing.sendcan)
           for frame in outgoing.sendcan:
             assert safety.safety_tx_hook(make_msg(frame.src, frame.address, dat=frame.dat))
+        if unbelted and belt_off[0]:
+          assert state.seatbeltUnlatched
+          assert not axis.desiredLongitudinal and not axis.longitudinalActive and not command.longActive
+          assert not safety.aol_get_permission_mask() & 2
         return state, axis, host_state, command
 
       for _ in range(120):
@@ -285,6 +301,16 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
         state, axis, host_state, command = step(engaged=True)
       assert state.canValid and host_state.enabled and command.latActive
       assert card.aol_card_intent.allowed_latch
+      if unbelted:
+        for _ in range(30):
+          state, axis, host_state, command = step(brake=True)
+        assert card.aol_card_intent.allowed_latch and axis.lateralActive and not axis.longitudinalActive
+        belt_off[0] = True
+        for _ in range(5):
+          state, axis, host_state, command = step()
+          assert state.seatbeltUnlatched and card.aol_card_intent.allowed_latch
+          assert axis.lateralActive and command.latActive and not command.longActive
+          assert safety.aol_get_permission_mask() == 1
       if lateral_only:
         for _ in range(30):
           state, axis, host_state, command = step(brake=True)
@@ -292,6 +318,56 @@ def _exercise_actual_family_callers(identity, monkeypatch, *, scenario='brake',
         for _ in range(30):
           state, axis, host_state, command = step()
         assert card.aol_card_intent.allowed_latch and axis.lateralActive and not axis.longitudinalActive
+      if scenario == 'warning':
+        from openpilot.selfdrive.selfdrived.events import EventName
+        for _ in range(30):
+          state, axis, host_state, command = step(brake=True)
+        for _ in range(30):
+          state, axis, host_state, command = step()
+        assert card.aol_card_intent.allowed_latch and axis.lateralActive and not command.longActive
+        soft_warning[0] = True
+        for paused_tick in range(350):
+          state, axis, host_state, command = step()
+          if any(event.softDisable for event in card.sm['onroadEvents']):
+            assert card.aol_transport_feedback.ordinary_warning_qualified
+          if paused_tick > 20:
+            assert safety.aol_get_request_mask() == safety.aol_get_permission_mask() == 0
+          assert card.aol_card_intent.allowed_latch, 'Ordinary warning erased armed Mazda lateral intent'
+          assert not command.latActive and not command.longActive
+          assert not axis.lateralActive and not axis.longitudinalActive
+          assert getattr(EventName, warning) in sd.events.names
+        soft_warning[0] = False
+        for _ in range(40):
+          state, axis, host_state, command = step()
+          assert card.aol_card_intent.allowed_latch
+          assert not command.longActive and not axis.longitudinalActive
+          if not any(event.softDisable for event in card.sm['onroadEvents']):
+            assert not card.aol_transport_feedback.ordinary_warning_qualified
+        assert axis.lateralActive and command.latActive, 'Healthy warning recovery failed to restore retained lateral'
+        return
+      if scenario == 'warning_active':
+        from openpilot.selfdrive.selfdrived.events import EventName
+        assert card.aol_card_intent.allowed_latch and host_state.enabled and command.latActive
+        soft_warning[0] = True
+        for paused_tick in range(350):
+          state, axis, host_state, command = step(engaged=True)
+          if any(event.softDisable for event in card.sm['onroadEvents']):
+            assert card.aol_transport_feedback.ordinary_warning_qualified
+          if paused_tick > 20:
+            assert safety.aol_get_request_mask() == safety.aol_get_permission_mask() == 0
+          assert card.aol_card_intent.allowed_latch, 'Ordinary warning erased armed Mazda lateral intent'
+          assert not command.latActive and not command.longActive
+          assert not axis.lateralActive and not axis.longitudinalActive
+          assert getattr(EventName, warning) in sd.events.names
+        soft_warning[0] = False
+        for _ in range(40):
+          state, axis, host_state, command = step()
+          assert card.aol_card_intent.allowed_latch
+          assert not command.longActive and not axis.longitudinalActive
+          if not any(event.softDisable for event in card.sm['onroadEvents']):
+            assert not card.aol_transport_feedback.ordinary_warning_qualified
+        assert axis.lateralActive and command.latActive, 'Healthy warning recovery failed to restore retained lateral'
+        return
       if scenario == 'mapped_cancel':
         for _ in range(5):
           state, axis, host_state, command = step(cancel=True)
@@ -483,3 +559,21 @@ def test_main_on_startup_disabled_aol_actual_card_controls_native(identity, monk
       safety.set_safety_hooks(structs.CarParams.SafetyModel.noOutput, 0)
 
   exercise_startup(identity, False, False, False, monkeypatch, probe=probe)
+
+
+@pytest.mark.parametrize('identity', (CAR.MAZDA_CX5_2022, CAR.MAZDA_CX9_2021))
+@pytest.mark.parametrize('warning', ('calibrationInvalid', 'lowMemory', 'processNotRunning', 'radarTempUnavailable'))
+def test_ordinary_warning_retains_intent_actual_callers(identity, warning, monkeypatch, armed_intent_decoder):
+  _exercise_actual_family_callers(identity, monkeypatch, scenario='warning', warning=warning, armed_intent=armed_intent_decoder)
+
+
+@pytest.mark.parametrize('identity', (CAR.MAZDA_CX5_2022, CAR.MAZDA_CX9_2021))
+@pytest.mark.parametrize('scenario', ('mapped_cancel', 'temporary'))
+def test_unbelted_lateral_pause_recovery_actual_callers(identity, scenario, monkeypatch, armed_intent_decoder):
+  _exercise_actual_family_callers(identity, monkeypatch, scenario=scenario, cancel_action=3,
+                                 armed_intent=armed_intent_decoder, unbelted=True)
+
+@pytest.mark.parametrize('identity', (CAR.MAZDA_CX5_2022, CAR.MAZDA_CX9_2021))
+@pytest.mark.parametrize('warning', ('calibrationInvalid', 'lowMemory', 'processNotRunning', 'radarTempUnavailable'))
+def test_active_ordinary_warning_retains_intent_actual_callers(identity, warning, monkeypatch, armed_intent_decoder):
+  _exercise_actual_family_callers(identity, monkeypatch, scenario='warning_active', warning=warning, armed_intent=armed_intent_decoder)

openpilot/selfdrive/car/card.py

@@ -658,7 +658,8 @@ class Car:
           native,
           latched=self.aol_card_intent.allowed_latch,
           panda_ready=angle_panda_ready,
-          restricted=(restrict(self.CP, CS) or
+          restricted=(getattr(self.aol_transport_feedback, 'ordinary_warning_qualified', False) or
+            restrict(self.CP, CS) or
             ((angle_aol or mazda_aol or preap_aol or rivian_aol) and self.aol_card_intent.pause_lateral) or
             (angle_aol and CS.brakePressed and (not CS.standstill) and (CS.vEgo < self.aol_settings.pause_brake_mps)) or
             (volkswagen_aol and (self.aol_card_intent.pause_lateral or
@@ -877,6 +878,8 @@ class Car:
       getattr(self, '_aol_can_observation', None), now_ns)
 
   def aol_disarming_fault(self, CS, event_ns: int, now_ns: int, *, transport_observed_healthy=False) -> bool:
+    if not aol_policy_for(self.CP).intent_supported:
+      return True
     from openpilot.starpilot.aol.vehicle import intent_disarming_fault
     if getattr(self, 'aol_transport_feedback', None) is None:
       self.aol_transport_feedback = TransportPauseFeedback(self.CP)

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -1977,9 +1977,8 @@ class SelfdriveD:
       cold_park_wait = self.cold_gm_park_readiness(CS, intent, now_ns)
       self.aol_fault_reason = ('none' if cold_park_wait else
                                'availabilityPause' if ordinary_availability or cold_main_wait else
-                               'critical' if self.aol_authority_lost or self.events.contains(ET.IMMEDIATE_DISABLE) or
-                               aol_hard_soft_fault and not availability else
-                               'availabilityPause' if availability else 'transportPause' if pause else 'none')
+                               'critical' if self.aol_authority_lost or self.events.contains(ET.IMMEDIATE_DISABLE) else
+                               'availabilityPause' if availability or aol_hard_soft_fault else 'transportPause' if pause else 'none')
       if self.aol_authority_lost or pause:
         self.events.add(EventName.controlsMismatch)
         lost_active_aol |= pause
@@ -2050,7 +2049,7 @@ class SelfdriveD:
       self.aol_lag_takeover_alert = True
     if (self.aol_replay and source_current and not was_enabled and not self.enabled and
         previous.lateral_active and previous.native_acknowledged and not self.aol_axis_decision.lateral_active and
-        self.aol_fault_reason == 'critical' and not self.aol_authority_lost and
+        self.aol_fault_reason in ('critical', 'availabilityPause') and not self.aol_authority_lost and
         intent is not None and intent.allowedLatch and self.events.contains(ET.SOFT_DISABLE) and
         not self.events.contains(ET.IMMEDIATE_DISABLE)):
       self.aol_soft_takeover_alert = True

openpilot/starpilot/aol/intent.py

@@ -122,16 +122,10 @@ def disarming_fault(events, CS, *, temporary_ui_process_failure: bool = False,
       if event.name == log.OnroadEvent.EventName.controlsMismatch and temporary_source_pause:
         continue
       return True
-    if event.softDisable:
-      if event.name == log.OnroadEvent.EventName.selfdrivedLagging and temporary_selfdrive_lagging:
-        continue
-      if event.name == log.OnroadEvent.EventName.processNotRunning and temporary_ui_process_failure:
-        continue
-      if event.name in (log.OnroadEvent.EventName.steerTempUnavailable, log.OnroadEvent.EventName.seatbeltNotLatched):
-        continue
-      if event.name == log.OnroadEvent.EventName.wrongGear and CS.gearShifter != GearShifter.unknown:
-        continue
+    if event.softDisable and event.name == log.OnroadEvent.EventName.wrongGear and CS.gearShifter == GearShifter.unknown:
       return True
+    # SOFT warnings withdraw commands through the normal runtime/state machine.
+    # They do not revoke the driver's separate armed lateral intent.
   return False
 
 
@@ -401,11 +395,11 @@ class AolCardIntent:
     rearm_safe = (CS.canValid and not CS.canTimeout and not self._calibration_fault and
                   not CS.steerFaultPermanent and not CS.steerFaultTemporary and
                   CS.gearShifter not in (GearShifter.park, GearShifter.neutral, GearShifter.reverse, GearShifter.unknown))
-    if self._calibration_inhibit and main_derived:
+    if self._calibration_inhibit and self._calibration_fault and main_derived:
       self._calibration_rearm_required = True
     elif self._calibration_rearm_required and fresh_edge and rearm_safe and self.settings.enabled:
       self._calibration_rearm_required = False
-    if self._calibration_inhibit or self._calibration_rearm_required:
+    if self._calibration_fault or self._calibration_rearm_required:
       self._record_latch_clear('calibration_inhibit')
       self.allowed_latch = False
     self._calibration_main = main
@@ -439,4 +433,4 @@ class AolCardIntent:
   def output(self, CS) -> tuple[bool, bool, bool]:
     driving = CS.gearShifter not in (GearShifter.neutral, GearShifter.park, GearShifter.reverse, GearShifter.unknown)
     return (self.allowed_latch and self.settings.enabled and CS.canValid and driving,
-            self.pause_lateral, self.pause_longitudinal)
+            self.pause_lateral or self._calibration_inhibit, self.pause_longitudinal or self._calibration_inhibit)

openpilot/starpilot/aol/tests/gm_bolt_recovery.py

@@ -299,14 +299,14 @@ def exercise_bolt_cancel_recovery(self, identity, removed, *, critical=None, com
           events.add(getattr(EventName, communications))
         if regen or brake:
           events.add(EventName.pedalPressed)
-        if fault and critical in ('controlsMismatch', 'calibrationInvalid'):
+        if fault and critical in ('controlsMismatch', 'steerUnavailable'):
           events.add(getattr(EventName, critical))
         ctx['events'] = [int(event) for event in events.names]
         if communication_loss:
           self.assertTrue(events.contains(ET.NO_ENTRY), msg=ctx)
           self.assertTrue(events.contains(ET.SOFT_DISABLE), msg=ctx)
           self.assertEqual(events.contains(ET.IMMEDIATE_DISABLE),
-                           bool(fault and critical in ('controlsMismatch', 'permanentEPS')), msg=ctx)
+                           bool(fault and critical in ('controlsMismatch', 'steerUnavailable', 'permanentEPS')), msg=ctx)
         if regen or brake:
           self.assertIn(EventName.pedalPressed, events.names, msg=ctx)
           self.assertTrue(events.contains(ET.USER_DISABLE), msg=ctx)

openpilot/starpilot/aol/tests/test_chrysler.py

@@ -275,14 +275,17 @@ class TestChryslerAol(unittest.TestCase):
     intent.observe_calibration(cs, events=ready.to_msg())
     self.assertFalse(intent.startup_calibration_pending)
     self.assertTrue(intent.allowed_latch)
-    self.assertTrue(intent_disarming_fault(cp, cold.to_msg(), cs, startup_calibration_pending=False))
-    # A later calibration revocation is a real withdrawal, not another cold boot.
-    intent.update(cs, now_ns=1_010_000_000, fault_active=True)
+    self.assertFalse(intent_disarming_fault(cp, cold.to_msg(), cs, startup_calibration_pending=False))
+    # A later SOFT calibration warning pauses commands without erasing intent.
+    intent.update(cs, now_ns=1_010_000_000, fault_active=False)
+    self.assertTrue(intent.allowed_latch)
+    # A true immediate fault still requires a new driver gesture.
+    intent.update(cs, now_ns=1_020_000_000, fault_active=True)
     self.assertFalse(intent.allowed_latch)
-    intent.update(cs, now_ns=1_020_000_000, fault_active=False)
+    intent.update(cs, now_ns=1_030_000_000, fault_active=False)
     self.assertFalse(intent.allowed_latch)
 
-  def test_cold_readiness_never_exempts_other_faults_or_immediate_calibration(self):
+  def test_soft_readiness_and_true_immediate_calibration_remain_distinct(self):
     from openpilot.cereal import log
     from openpilot.selfdrive.selfdrived.events import Events
     from openpilot.starpilot.aol.vehicle import intent_disarming_fault
@@ -292,7 +295,8 @@ class TestChryslerAol(unittest.TestCase):
                  log.OnroadEvent.EventName.doorOpen, log.OnroadEvent.EventName.canError):
       events = Events()
       events.add(name)
-      self.assertTrue(intent_disarming_fault(cp, events.to_msg(), cs, startup_calibration_pending=True), name)
+      self.assertEqual(intent_disarming_fault(cp, events.to_msg(), cs, startup_calibration_pending=True),
+                       any(event.immediateDisable for event in events.to_msg()), name)
     event = log.OnroadEvent.new_message(name=log.OnroadEvent.EventName.calibrationIncomplete, immediateDisable=True)
     self.assertTrue(intent_disarming_fault(cp, [event], cs, startup_calibration_pending=True))
     cs.steerFaultPermanent = True
@@ -304,39 +308,40 @@ class TestChryslerAol(unittest.TestCase):
     from openpilot.cereal import log
     from openpilot.selfdrive.selfdrived.events import Events
     from openpilot.starpilot.car.chrysler.aol import cold_calibration_intent_current
-    cp = cp_for(CAR.JEEP_GRAND_CHEROKEE)
-    cs = state()
-    events = Events()
-    events.add(log.OnroadEvent.EventName.calibrationIncomplete)
-    axis = SimpleNamespace(intentProducerSessionId='card-1', intentIntegrityEpoch=0,
-      desiredLongitudinal=False, longitudinalActive=False, lateralActive=False,
-      desiredLateral=False, observedMonoTime=2_000_000_000,
-      faultReason='critical', faultEventMonoTime=1_900_000_000)
-    native = SimpleNamespace(pandaInventory=(object(),), requestedLongitudinal=False,
-      longitudinalAllowed=False, requestedLateral=False, lateralAllowed=False)
-    class Peers:
-      seen = valid = alive = {'onroadEvents': True}
-      logMonoTime = {'onroadEvents': 1_900_000_000}
-      def __getitem__(self, _):
-        return events.to_msg()
-    feedback = SimpleNamespace(failed=False, ever_lateral_active=False, signature=None, cold_main_wait_ns=0)
-    def observe():
-      return cold_calibration_intent_current(feedback, Peers(), cp, cs, axis, native, 2_010_000_000,
-        pending=True, producer='card-1')
-    self.assertTrue(observe())
-    for obj, field, value in ((feedback, 'failed', True), (feedback, 'ever_lateral_active', True),
-        (axis, 'intentProducerSessionId', 'restarted-card'), (axis, 'desiredLongitudinal', True),
-        (native, 'requestedLongitudinal', True), (native, 'lateralAllowed', True),
-        (cs, 'doorOpen', True), (cs, 'steeringDisengage', True)):
-      previous = getattr(obj, field)
-      setattr(obj, field, value)
-      self.assertFalse(observe(), field)
-      setattr(obj, field, previous)
-    for name in (log.OnroadEvent.EventName.canError, log.OnroadEvent.EventName.calibrationInvalid,
-                 log.OnroadEvent.EventName.doorOpen):
-      events.add(name)
-      self.assertFalse(observe(), name)
-      events.clear()
+    for reason in ('critical', 'availabilityPause'):
+      cp = cp_for(CAR.JEEP_GRAND_CHEROKEE)
+      cs = state()
+      events = Events()
       events.add(log.OnroadEvent.EventName.calibrationIncomplete)
-    Peers.logMonoTime['onroadEvents'] = 1
-    self.assertFalse(observe())
+      axis = SimpleNamespace(intentProducerSessionId='card-1', intentIntegrityEpoch=0,
+        desiredLongitudinal=False, longitudinalActive=False, lateralActive=False,
+        desiredLateral=False, observedMonoTime=2_000_000_000,
+        faultReason=reason, faultEventMonoTime=1_900_000_000)
+      native = SimpleNamespace(pandaInventory=(object(),), requestedLongitudinal=False,
+        longitudinalAllowed=False, requestedLateral=False, lateralAllowed=False)
+      class Peers:
+        seen = valid = alive = {'onroadEvents': True}
+        logMonoTime = {'onroadEvents': 1_900_000_000}
+        def __getitem__(self, _, events=events):
+          return events.to_msg()
+      feedback = SimpleNamespace(failed=False, ever_lateral_active=False, signature=None, cold_main_wait_ns=0)
+      def observe(feedback=feedback, cp=cp, cs=cs, axis=axis, native=native, Peers=Peers):
+        return cold_calibration_intent_current(feedback, Peers(), cp, cs, axis, native, 2_010_000_000,
+          pending=True, producer='card-1')
+      self.assertTrue(observe())
+      for obj, field, value in ((feedback, 'failed', True), (feedback, 'ever_lateral_active', True),
+          (axis, 'intentProducerSessionId', 'restarted-card'), (axis, 'desiredLongitudinal', True),
+          (native, 'requestedLongitudinal', True), (native, 'lateralAllowed', True),
+          (cs, 'doorOpen', True), (cs, 'steeringDisengage', True)):
+        previous = getattr(obj, field)
+        setattr(obj, field, value)
+        self.assertFalse(observe(), field)
+        setattr(obj, field, previous)
+      for name in (log.OnroadEvent.EventName.canError, log.OnroadEvent.EventName.calibrationInvalid,
+                   log.OnroadEvent.EventName.doorOpen):
+        events.add(name)
+        self.assertFalse(observe(), name)
+        events.clear()
+        events.add(log.OnroadEvent.EventName.calibrationIncomplete)
+      Peers.logMonoTime['onroadEvents'] = 1
+      self.assertFalse(observe())

openpilot/starpilot/aol/tests/test_gm.py

@@ -490,7 +490,7 @@ class TestGmAol(unittest.TestCase):
       events.add(log.OnroadEvent.EventName.selfdrivedLagging)
       self.assertTrue(events.contains(ET.NO_ENTRY))
       self.assertTrue(events.contains(ET.SOFT_DISABLE))
-      self.assertTrue(disarming_fault(events.to_msg(), cs))  # Other vehicle owners retain their existing law.
+      self.assertFalse(disarming_fault(events.to_msg(), cs))  # SOFT withdraws output while retaining driver intent.
       for standard in (False, True):
         with patch.object(selected, 'sm', {'onroadEvents': events.to_msg()}):
           fault = selected.aol_disarming_fault(cs, 1_000_000_000, 1_000_000_001)
@@ -537,7 +537,7 @@ class TestGmAol(unittest.TestCase):
     with OpenpilotPrefix(), patch.dict(os.environ, {'SIMULATION': '1', 'AOL_REPLAY_RUNTIME': '0'}):
       settings = Params()
       settings.put_bool('AlwaysOnLateral', True, block=True)
-      for extra in (log.OnroadEvent.EventName.controlsMismatch, log.OnroadEvent.EventName.processNotRunning, None):
+      for extra in (log.OnroadEvent.EventName.controlsMismatch, log.OnroadEvent.EventName.steerUnavailable, None):
         with self.subTest(extra=extra):
           selected = self.card(pedal_params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True), settings)
           cs = structs.CarState(canValid=True, gearShifter='drive', vEgo=20.)
@@ -893,7 +893,7 @@ class TestGmAol(unittest.TestCase):
         self.assertEqual(any(requested), supported)
 
 
-  def test_calibration_recovery_requires_new_main_edge_through_actual_callers(self):
+  def test_calibration_pause_retains_intent_and_recovers_without_new_main_edge(self):
     from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
     from openpilot.starpilot.aol.tests.test_runtime import sample_car_state
     from openpilot.selfdrive.selfdrived.state import StateMachine
@@ -919,10 +919,10 @@ class TestGmAol(unittest.TestCase):
         sd.sm = messaging.SubMaster(['aolIntentWire', 'aolSafetyWire', 'modelV2',
                                      'extrinsicsCalibration', 'driverMonitoringState', 'pandaStates'])
         cs = structs.CarState(canValid=True, vEgo=20., gearShifter='drive')
-        for tick, (phase, expected) in enumerate((('unbuckled', True), ('unbuckled', True), ('healthy', True), ('invalid', False), ('recovered', False),
-                                                  ('held', False), ('unmapped_lkas', False), ('unmapped_main', False),
-                                                  ('release_lkas', False), ('main_off', False), ('fault_edge', False),
-                                                  ('recovered_again', False), ('main_off', False), ('new_main', True),
+        for tick, (phase, expected) in enumerate((('unbuckled', True), ('unbuckled', True), ('healthy', True), ('invalid', False), ('recovered', True),
+                                                  ('held', True), ('unmapped_lkas', True), ('unmapped_main', True),
+                                                  ('release_lkas', True), ('main_off', False), ('fault_edge', False),
+                                                  ('recovered_again', True), ('main_off', False), ('new_main', True),
                                                   ('temporary', False), ('healthy', True), ('reverse', False),
                                                   ('healthy', True), ('model_stale', False), ('healthy', True))):
           now = 1_000_000_000 + tick * 10_000_000
@@ -947,8 +947,7 @@ class TestGmAol(unittest.TestCase):
           selected.aol_card_intent.update(cs)
           selected.observe_aol_calibration(cs, now, False)
           self.assertEqual(selected.aol_card_intent.allowed_latch,
-                           phase not in ('invalid', 'recovered', 'held', 'unmapped_lkas', 'unmapped_main', 'release_lkas',
-                                         'main_off', 'fault_edge', 'recovered_again'))
+                           phase != 'main_off')
           intent = messaging.new_message('aolIntentWire', 0, valid=True, logMonoTime=now)
           allowed, pause_lat, pause_long = selected.aol_card_intent.output(cs)
           intent.aolIntentWire = encode_intent(IntentState('card', tick + 1, now, now, now + 30_000_000,
@@ -965,7 +964,7 @@ class TestGmAol(unittest.TestCase):
           sd.sm.update_msgs(now / 1e9, [intent.as_reader(), receipt.as_reader(), model.as_reader(),
                                        calibration.as_reader(), monitoring.as_reader()])
           sd.aol_car_state_log_ns, sd.enabled, sd.active = now, False, False
-          sd.events.clear()
+          sd.events = events
           if cs.seatbeltUnlatched:
             sd.events.add(EventName.seatbeltNotLatched)
             self.assertTrue(sd.events.contains(ET.NO_ENTRY))
@@ -1140,7 +1139,7 @@ class TestGmAol(unittest.TestCase):
           self.exercise_bolt_cancel_recovery(identity, removed)
 
   def test_bolt_cancel_fault_rearm(self):
-    for critical in ('controlsMismatch', 'calibrationInvalid', 'permanentEPS'):
+    for critical in ('controlsMismatch', 'steerUnavailable', 'permanentEPS'):
       with self.subTest(critical=critical):
         self.exercise_bolt_cancel_recovery(CAR.CHEVROLET_BOLT_CC_2018_2021, False, critical=critical)
 
@@ -1150,7 +1149,7 @@ class TestGmAol(unittest.TestCase):
         self.exercise_bolt_cancel_recovery(CAR.CHEVROLET_BOLT_CC_2018_2021, False, communications=event)
 
   def test_bolt_communications_preserves_critical_rearm(self):
-    for critical in ('controlsMismatch', 'calibrationInvalid', 'permanentEPS'):
+    for critical in ('controlsMismatch', 'steerUnavailable', 'permanentEPS'):
       with self.subTest(critical=critical):
         self.exercise_bolt_cancel_recovery(CAR.CHEVROLET_BOLT_CC_2018_2021, False,
                                           critical=critical, communications='commIssueAvgFreq')
@@ -1384,7 +1383,7 @@ class TestGmAol(unittest.TestCase):
     self.exercise_volt_ascm_acc_intent()
 
   def test_volt_ascm_mixed_critical_acc_fault_requires_main_cycle(self):
-    for critical in ('controlsMismatch', 'calibrationInvalid', 'permanentEPS'):
+    for critical in ('controlsMismatch', 'steerUnavailable', 'permanentEPS'):
       with self.subTest(critical=critical):
         self.exercise_volt_ascm_acc_intent(critical=critical)
 
@@ -1475,7 +1474,7 @@ class TestGmAol(unittest.TestCase):
           safety.safety_tick()
           safety.set_aol_test_heartbeat(True)
           events = car_events.update(cs, previous, structs.CarControl())
-          if fault and critical in ('controlsMismatch', 'calibrationInvalid'):
+          if fault and critical in ('controlsMismatch', 'steerUnavailable'):
             events.add(getattr(EventName, critical))
           self.assertEqual(EventName.accFaulted in events.names, fault)
           if fault:

openpilot/starpilot/aol/tests/test_gm_held_native.py

@@ -654,7 +654,7 @@ def test_actual_sd_belt_classification_wire_reaches_next_card_cycle(gesture, fau
       now = NOW + tick * 10_000_000
       current['now'], current['controls'] = now, False
       cs = receive_card(now, gesture if tick < 3 else None, tick < 3, (tick + 1) % 4)
-      if tick == 2 and fault is not None:
+      if tick == 2 and fault == 'immediate':
         assert card.aol_transport_feedback.failed and not card.aol_card_intent.allowed_latch
         return
       assert not card.aol_transport_feedback.failed and card.aol_card_intent.allowed_latch
@@ -679,8 +679,11 @@ def test_actual_sd_belt_classification_wire_reaches_next_card_cycle(gesture, fau
       assert native.observedMonoTime == NOW and native.validUntilMonoTime == NOW + 200_000_000
       assert native.requestedLateral and native.lateralAllowed and native.requestedLongitudinal and native.longitudinalAllowed
       assert not axis.longitudinalActive and not command.longActive
-      if fault is not None:
+      if fault == 'immediate':
         assert str(axis.faultReason) == 'critical'
+      elif fault == 'other_soft' and tick == 1:
+        assert str(axis.faultReason) == 'availabilityPause'
+        assert not axis.lateralActive and not command.latActive
       else:
         assert str(axis.faultReason) == 'none' and axis.lateralActive
         assert not sd.enabled  # Pedal/Cancel still disables ordinary LONG; rebuckling cannot resume it.

openpilot/starpilot/aol/tests/test_gm_ordinary_warning.py

@@ -0,0 +1,161 @@
+"""Parsed SDGM sources and serialized peers; native responses are controlled fixtures."""
+
+from dataclasses import replace
+
+import pytest
+from opendbc.car.gm.tests.test_volt_sdgm_control import sdgm_params
+from openpilot.cereal import messaging
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.selfdrived.events import Events, EventName
+from openpilot.starpilot.aol.tests.test_gm_held_native import NOW, Peers, peers, _sdgm_pedal_state
+from openpilot.starpilot.aol.transport_pause import TransportPauseFeedback
+
+
+@pytest.mark.parametrize('c9', (False, True))
+@pytest.mark.parametrize('warning', ('calibrationInvalid', 'lowMemory', 'processNotRunning', 'radarTempUnavailable'))
+@pytest.mark.parametrize('native_state', ('held_full', 'withdrawn_long', 'all_off'))
+def test_sdgm_warning_withdrawal_keeps_only_intent(c9, warning, native_state):
+  from openpilot.starpilot.aol.tests.test_gm import TestGmAol as GmFixture
+
+  with OpenpilotPrefix():
+    saved = Params()
+    saved.put_bool('AlwaysOnLateral', True, block=True)
+    card = GmFixture.card(sdgm_params(alpha=True, brake_c9=c9), saved)
+    try:
+      cp = card.CP
+      feedback = TransportPauseFeedback(cp)
+      cs = _sdgm_pedal_state(card, NOW, 0)
+      card.aol_card_intent.update(cs, fault_active=False, now_ns=NOW)
+      sm, native = peers(cp, card.slc_producer_session, NOW)
+      assert not feedback.observe(sm, cp, cs, NOW, latched=True, intent_producer_session=card.slc_producer_session)
+      assert feedback.lateral_baseline_ns == NOW
+      for tick in range(1, 6):
+        now = NOW + tick * 10_000_000
+        cs = _sdgm_pedal_state(card, now, tick)
+        controls = native_state == 'held_full'
+        current_native = replace(
+          native,
+          longitudinalAllowed=controls,
+          lateralAllowed=native_state != 'all_off',
+          requestedLateral=native_state != 'all_off',
+          requestedLongitudinal=native_state != 'all_off',
+          pandaInventory=tuple(replace(slot, controlsAllowed=controls) for slot in native.pandaInventory),
+        )
+        sm, _ = peers(cp, card.slc_producer_session, now, native=current_native, controls=controls, sequence=tick + 1, long_active=False)
+        axis = sm.messages['aolAxisState'].as_builder()
+        axis.aolAxisState.faultReason = 'availabilityPause'
+        axis.aolAxisState.faultEventMonoTime = now
+        axis.aolAxisState.desiredLateral = False
+        axis.aolAxisState.lateralActive = False
+        axis.aolAxisState.nativeAcknowledged = False
+        events = Events()
+        events.add(getattr(EventName, warning))
+        event = messaging.new_message('onroadEvents', len(events), valid=True, logMonoTime=now)
+        event.onroadEvents = events.to_msg()
+        sm = Peers([sm.messages['aolSafetyWire'], sm.messages['pandaStates'], axis.as_reader(), event.as_reader()])
+        if not c9 and warning == 'lowMemory' and native_state == 'all_off' and tick == 1:
+          _check_warning_guards(feedback, sm, cp, cs, now, card.slc_producer_session)
+        assert not feedback.observe(sm, cp, cs, now, latched=True, intent_producer_session=card.slc_producer_session), (native_state, feedback.__dict__)
+        assert feedback.ordinary_warning_qualified and not feedback.failed
+        card.aol_card_intent.update(cs, fault_active=False, now_ns=now)
+        assert card.aol_card_intent.allowed_latch
+        # Retention does not write back or manufacture native authority.
+        assert current_native == replace(
+          native,
+          longitudinalAllowed=controls,
+          lateralAllowed=native_state != 'all_off',
+          requestedLateral=native_state != 'all_off',
+          requestedLongitudinal=native_state != 'all_off',
+          pandaInventory=tuple(replace(slot, controlsAllowed=controls) for slot in native.pandaInventory),
+        )
+    finally:
+      card.vehicle_startup.close()
+
+
+def _check_warning_guards(feedback, sm, cp, cs, now, producer):
+  from copy import copy
+  from openpilot.starpilot.aol.wire import decode_safety, encode_safety
+
+  for fault in (
+    'unarmed',
+    'producer',
+    'epoch',
+    'session',
+    'source_stale',
+    'source_future',
+    'can_invalid',
+    'can_timeout',
+    'permanent_eps',
+    'acc_fault',
+    'door',
+    'steering_off',
+    'main_off',
+    'gear_unknown',
+    'native_expired',
+    'native_future',
+    'inventory',
+    'native_rx',
+    'heartbeat',
+    'panda_rx',
+    'panda_heartbeat',
+    'immediate',
+    'event_expired',
+    'event_future',
+  ):
+    tested = copy(feedback)
+    messages = {name: event.as_builder() for name, event in sm.messages.items()}
+    state = cs.as_reader().as_builder()
+    axis = messages['aolAxisState'].aolAxisState
+    native = decode_safety(messages['aolSafetyWire'].aolSafetyWire)
+    latched = fault != 'unarmed'
+    if fault == 'producer':
+      axis.intentProducerSessionId = 'different'
+    if fault == 'epoch':
+      axis.intentIntegrityEpoch += 1
+    if fault == 'session':
+      axis.sessionId = 'different'
+    if fault == 'source_stale':
+      axis.sourceCarStateMonoTime = now - 30_000_001
+    if fault == 'source_future':
+      axis.sourceCarStateMonoTime = now + 1
+    if fault == 'can_invalid':
+      state.canValid = False
+    if fault == 'can_timeout':
+      state.canTimeout = True
+    if fault == 'permanent_eps':
+      state.steerFaultPermanent = True
+    if fault == 'acc_fault':
+      state.accFaulted = True
+    if fault == 'door':
+      state.doorOpen = True
+    if fault == 'steering_off':
+      state.steeringDisengage = True
+    if fault == 'main_off':
+      state.cruiseState.available = False
+    if fault == 'gear_unknown':
+      state.gearShifter = 'unknown'
+    if fault == 'native_expired':
+      native = replace(native, validUntilMonoTime=now - 1)
+    if fault == 'native_future':
+      native = replace(native, observedMonoTime=now + 1, validUntilMonoTime=now + 200_000_001)
+      messages['aolSafetyWire'].logMonoTime = now + 1
+    if fault == 'inventory':
+      native = replace(native, pandaInventory=tuple(replace(slot, hardwareSerial='other') for slot in native.pandaInventory))
+    if fault == 'native_rx':
+      native = replace(native, pandaInventory=tuple(replace(slot, safetyRxChecksInvalid=True) for slot in native.pandaInventory))
+    if fault == 'heartbeat':
+      native = replace(native, pandaInventory=tuple(replace(slot, heartbeatLost=True) for slot in native.pandaInventory))
+    if fault == 'panda_rx':
+      messages['pandaStates'].pandaStates[0].safetyRxChecksInvalid = True
+    if fault == 'panda_heartbeat':
+      messages['pandaStates'].pandaStates[0].heartbeatLost = True
+    if fault == 'immediate':
+      messages['onroadEvents'].onroadEvents[0].immediateDisable = True
+    if fault == 'event_expired':
+      messages['onroadEvents'].logMonoTime = now - 1_500_000_001
+    if fault == 'event_future':
+      messages['onroadEvents'].logMonoTime = now + 1
+    messages['aolSafetyWire'].aolSafetyWire = encode_safety(native)
+    shadow = Peers([event.as_reader() for event in messages.values()])
+    assert not tested._ordinary_warning_intent_current(shadow, cp, state, axis, now, now, latched=latched, producer=producer), fault

openpilot/starpilot/aol/tests/test_independent_lateral_soft_takeover.py

@@ -19,7 +19,7 @@ def test_independent_aol_low_memory_withdrawal_selects_takeover_alert(identity):
     assert axis.lateralActive and command.latActive and not host.enabled
     axis, host, command = step(20, critical_event=log.OnroadEvent.EventName.lowMemory)
     assert not axis.lateralActive and not command.latActive and not host.enabled
-    assert sd.aol_fault_reason == 'critical'
+    assert sd.aol_fault_reason == 'availabilityPause'
     state = sd.pm.messages['selfdriveState'].selfdriveState
     assert state.alertType == 'lowMemory/softDisable', state.to_dict()
     assert state.alertStatus == log.SelfdriveState.AlertStatus.critical

openpilot/starpilot/aol/tests/test_runtime.py

@@ -282,7 +282,7 @@ class CardIntentTests(unittest.TestCase):
     self.assertTrue(direct.lateral_active)
     self.assertFalse(direct.desired_longitudinal)
     events.add(log.OnroadEvent.EventName.doorOpen)
-    self.assertTrue(disarming_fault(events.to_msg(), state))
+    self.assertFalse(disarming_fault(events.to_msg(), state))
     self.assertTrue(aol_no_entry(events.names, state, paddle_only_cancel=False))
 
   def test_temporary_eps_and_known_gears_preserve_latch_with_zero_steering(self):
@@ -328,7 +328,7 @@ class CardIntentTests(unittest.TestCase):
           dm_lockout=False, pause_brake_mps=0)
         self.assertTrue(decision.lateral_active)
 
-  def test_calibration_rearm_follows_live_mapping_without_arming_on_change(self):
+  def test_calibration_pause_retains_intent_and_mapping_change_cannot_rearm(self):
     from dataclasses import replace
     from openpilot.starpilot.aol.intent import AOL_TOGGLE
     owner = AolCardIntent(AolSettings(True, 0., 0, 0, (0, 0, 0), (0, 0, 0)))
@@ -340,17 +340,26 @@ class CardIntentTests(unittest.TestCase):
     invalid = Events()
     invalid.add(log.OnroadEvent.EventName.calibrationInvalid)
     owner.observe_calibration(state, events=invalid.to_msg())
-    self.assertFalse(owner.allowed_latch)
+    self.assertTrue(owner.allowed_latch)
+    self.assertEqual(owner.output(state)[1:], (True, True))
     owner.settings = replace(owner.settings, lkas_action=AOL_TOGGLE)
     owner.update(state)
-    owner.observe_calibration(state, events=invalid.to_msg())
-    self.assertFalse(owner.allowed_latch)
-    owner.update(state)
     owner.observe_calibration(state, events=[])
-    self.assertFalse(owner.allowed_latch)  # The mapping change is not an arming gesture.
+    self.assertTrue(owner.allowed_latch)
+    self.assertEqual(owner.output(state)[1:], (False, False))
+    state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=True)]
+    owner.update(state)
+    self.assertFalse(owner.allowed_latch)
+    state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=False)]
+    owner.update(state)
+    state.buttonEvents = []
+    owner.settings = replace(owner.settings, main_action=AOL_TOGGLE)
+    owner.update(state)
+    owner.settings = replace(owner.settings, main_action=0)
+    owner.update(state)
+    self.assertFalse(owner.allowed_latch)  # Changing another mapping is not an arming gesture.
     state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=True)]
     owner.update(state)
-    owner.observe_calibration(state, events=[])
     self.assertTrue(owner.allowed_latch)
 
   def test_low_speed_brake_suppresses_output_without_destroying_explicit_latch(self):
@@ -382,7 +391,8 @@ class CardIntentTests(unittest.TestCase):
                  log.OnroadEvent.EventName.overheat):
       events = Events()
       events.add(name)
-      self.assertTrue(disarming_fault(events.to_msg(), state))
+      self.assertEqual(disarming_fault(events.to_msg(), state),
+                       any(event.immediateDisable for event in events.to_msg()))
     events = Events()
     events.add(log.OnroadEvent.EventName.wrongGear)
     self.assertFalse(disarming_fault(events.to_msg(), state))  # Events may lag the return to Drive.
@@ -2124,32 +2134,38 @@ class UiProcessIntentTests(unittest.TestCase):
     owner.update(state)
     self.assertFalse(owner.allowed_latch)  # Physical CANCEL remains intentional withdrawal.
 
-  def test_context_denies_unknown_mixed_stale_missing_and_critical_faults(self):
+  def test_soft_process_context_guards_remain_distinct_from_immediate_faults(self):
     card = self.make_card()
     state = car_state()
     card.sm.events.add(log.OnroadEvent.EventName.processNotRunning)
     event_ns = 1_020_000_000
-    self.assertTrue(card.aol_disarming_fault(state, event_ns, event_ns + 1))
+    self.assertFalse(card.aol_disarming_fault(state, event_ns, event_ns + 1))
+    self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(event_ns, event_ns + 1))
     for failures in ({'controlsd'}, {'unknown_service'}, {'ui', 'modeld'}, set()):
       with self.subTest(failures=failures):
         card.aol_process_fault_context.history.clear()
         self.observe(card, 1_000_000_000, failures)
-        self.assertTrue(card.aol_disarming_fault(state, event_ns, event_ns + 1))
+        self.assertFalse(card.aol_disarming_fault(state, event_ns, event_ns + 1))
+        self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(event_ns, event_ns + 1))
     self.observe(card, 1_100_000_000, {'ui'})
-    self.assertTrue(card.aol_disarming_fault(state, 1_120_000_000, 2_600_000_001))
+    self.assertFalse(card.aol_disarming_fault(state, 1_120_000_000, 2_600_000_001))
+    self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(1_120_000_000, 2_600_000_001))
     self.observe(card, 1_150_000_000, {'ui', 'modeld'})
-    self.assertTrue(card.aol_disarming_fault(state, 1_120_000_000, 1_160_000_000))
+    self.assertFalse(card.aol_disarming_fault(state, 1_120_000_000, 1_160_000_000))
+    self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(1_120_000_000, 1_160_000_000))
     card.aol_process_fault_context.history.clear()
     self.observe(card, 1_100_000_000, {'ui'})
     card.sm.processes[0].shouldBeRunning = False
     card.aol_process_fault_context.observe(card.sm, 1_120_000_000)
-    self.assertTrue(card.aol_disarming_fault(state, 1_120_000_000, 1_120_000_001))
+    self.assertFalse(card.aol_disarming_fault(state, 1_120_000_000, 1_120_000_001))
+    self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(1_120_000_000, 1_120_000_001))
     self.observe(card, 1_100_000_000, {'ui'})
     for source_flag in ('seen', 'valid', 'alive'):
       with self.subTest(source_flag=source_flag):
         getattr(card.sm, source_flag)['managerState'] = False
         card.aol_process_fault_context.observe(card.sm, 1_120_000_000)
-        self.assertTrue(card.aol_disarming_fault(state, 1_120_000_000, 1_120_000_001))
+        self.assertFalse(card.aol_disarming_fault(state, 1_120_000_000, 1_120_000_001))
+        self.assertFalse(card.aol_process_fault_context.temporary_ui_failure(1_120_000_000, 1_120_000_001))
         getattr(card.sm, source_flag)['managerState'] = True
         self.observe(card, 1_100_000_000, {'ui'})
     for name in (log.OnroadEvent.EventName.controlsMismatch, log.OnroadEvent.EventName.steerUnavailable):

openpilot/starpilot/aol/transport_pause.py

@@ -299,6 +299,8 @@ class TransportPauseFeedback:
     self.availability_producer = ''
     self.availability_epoch = None
     self.source_retention_started_ns = 0
+    self.ordinary_warning_retained = False
+    self.ordinary_warning_qualified = False
     self.healthy_feedback_ns = 0
     self.healthy_native_ns = 0
     self.healthy_inventory = None
@@ -358,6 +360,69 @@ class TransportPauseFeedback:
       return None
     return native
 
+  def _ordinary_warning_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer):
+    """Retain armed intent through a fresh SOFT withdrawal; grant no output."""
+    from openpilot.starpilot.aol.vehicle import policy_for
+    from openpilot.starpilot.nostalgia import physical_cancel
+    from opendbc.car import structs
+
+    if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
+        not self.session or not self.lateral_baseline_ns or not self.lateral_inventory or
+        not policy_for(cp).intent_supported or
+        policy_for(cp).cruise_main_required and not cs.cruiseState.available or
+        str(axis.sessionId) != self.session or
+        not producer or producer != self.availability_producer or axis.intentProducerSessionId != producer or
+        axis.intentIntegrityEpoch != self.availability_epoch or
+        not cs.canValid or cs.canTimeout or cs.steerFaultPermanent or cs.accFaulted or cs.doorOpen or
+        cs.steeringDisengage or
+        cs.gearShifter not in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport,
+                              structs.CarState.GearShifter.low, structs.CarState.GearShifter.brake) or
+        not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= stamp <= now_ns or
+        now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
+        str(axis.faultReason) not in ('none', 'availabilityPause') or
+        not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
+      return False
+    event_ns = int(sm.logMonoTime['onroadEvents'])
+    events = sm['onroadEvents']
+    if (not 0 < event_ns <= now_ns or now_ns - event_ns > 1_500_000_000 or
+        any(event.immediateDisable for event in events)):
+      return False
+    warning = any(event.softDisable for event in events)
+    recovering = bool(getattr(self, 'ordinary_warning_retained', False) and not warning)
+    if not warning and not recovering:
+      return False
+    stock_user = stock_lateral_user_event(cp)
+    # Physical withdrawals still go through the family button/pedal owner.
+    # This helper cannot rearm a latch that that owner explicitly clears.
+    if any(event.userDisable and event.name != stock_user and not (
+        event.name == log.OnroadEvent.EventName.pedalPressed and (cs.brakePressed or cs.regenBraking) or
+        event.name == log.OnroadEvent.EventName.buttonCancel and (physical_cancel(cs, cp) or
+          cp.brand == 'gm' and 0 < self.gm_withdrawal_ns == event_ns <= now_ns and
+          now_ns - event_ns <= SAFETY_MAX_AGE_NS)) for event in events):
+      return False
+    prior = self.signature
+    active_join = bool(warning and prior is not None and str(axis.faultReason) == 'none' and
+      axis.lateralActive and axis.desiredLateral and axis.nativeAcknowledged and
+      prior[9] and prior[11] and prior[13] and prior[14:16] == (producer, self.availability_epoch) and
+      0 < prior[8] <= now_ns <= prior[4] and now_ns - prior[8] <= AXIS_MAX_AGE_NS and
+      0 < axis.faultEventMonoTime < event_ns and now_ns - event_ns <= AXIS_MAX_AGE_NS)
+    withdrawn = bool(not axis.lateralActive and not axis.longitudinalActive and
+      not axis.desiredLongitudinal and
+      (warning and str(axis.faultReason) == 'availabilityPause' and axis.faultEventMonoTime == event_ns <= stamp or
+       recovering and str(axis.faultReason) == 'none'))
+    if not active_join and not withdrawn:
+      return False
+    native = _configured_transport_native(sm, cp, self.session, now_ns,
+      allow_restricted_lateral=withdrawn or bool(cs.steerFaultTemporary),
+      require_longitudinal=not withdrawn)
+    if (native is None or not native.pandaInventory or
+        (native.pandaSerial, tuple(slot.hardwareSerial for slot in native.pandaInventory)) != self.lateral_inventory or
+        native.observedMonoTime < self.lateral_native_ns or
+        active_join and not (native.requestedLateral and native.lateralAllowed)):
+      return False
+    self.ordinary_warning_retained = True
+    return True
+
   def _temporary_host_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer):
     """A fresh inactive proof may retain intent, never native/output authority."""
     from openpilot.starpilot.nostalgia import physical_cancel
@@ -1083,6 +1148,7 @@ class TransportPauseFeedback:
 
   def observe(self, sm, cp, cs, now_ns, *, latched=True, cancel_source_ns=0, brake_source_ns=0,
               gm_withdrawal_source_ns=0, intent_producer_session='', main_arm_ns=0, startup_calibration_pending=False, intent_paused=False):
+    self.ordinary_warning_qualified = False
     self.temporary_host_qualified = False
     self.qualified = False
     self.awaiting = False
@@ -1091,6 +1157,7 @@ class TransportPauseFeedback:
     try:
       axis = sm['aolAxisState']
       if not latched or cp_identity(cp) != self.cp or self.session and str(axis.sessionId) != self.session:
+        self.ordinary_warning_retained = False
         self.temporary_eps_event_ns = 0
         self.ccnc_main_off_source_ns = 0
         self.cold_park_waiting = False
@@ -1197,6 +1264,11 @@ class TransportPauseFeedback:
           axis.desiredLateral and not axis.lateralActive):
         self.failed = True
         return True
+      if current and continuity and self._ordinary_warning_intent_current(sm, cp, cs, axis, stamp, now_ns,
+          latched=latched, producer=intent_producer_session):
+        self.ordinary_warning_qualified = any(event.softDisable for event in sm['onroadEvents'])
+        self.sequence, self.signature = int(axis.sequence), signature
+        return False
       if current and continuity and self._temporary_host_intent_current(sm, cp, cs, axis, stamp, now_ns,
           latched=latched, producer=intent_producer_session):
         self.sequence, self.signature = int(axis.sequence), signature
@@ -1469,6 +1541,7 @@ class TransportPauseFeedback:
         self.lateral_only_baseline = bool(lateral_baseline and self.availability_producer and
           not axis.desiredLongitudinal and not axis.longitudinalActive and not native.requestedLongitudinal)
         if lateral_baseline and native is not None:
+          self.ordinary_warning_retained = False
           self.ever_lateral_active = True
           self.lag_withdrawal_ns = 0
           self.cold_main_wait_ns = 0

openpilot/starpilot/car/chrysler/aol.py

@@ -86,7 +86,7 @@ def cold_calibration_intent_current(feedback, sm, cp, cs, axis, native, now_ns,
     return False
   cold = any(e.name == cal and e.softDisable and not e.immediateDisable for e in events)
   if cold:
-    if (not pending or str(axis.faultReason) != 'critical' or not 0 < axis.faultEventMonoTime <= event_ns or
+    if (not pending or str(axis.faultReason) not in ('critical', 'availabilityPause') or not 0 < axis.faultEventMonoTime <= event_ns or
         axis.desiredLateral or native.requestedLateral or native.lateralAllowed):
       return False
     feedback.cold_main_wait_ns = int(axis.observedMonoTime)
@@ -94,7 +94,7 @@ def cold_calibration_intent_current(feedback, sm, cp, cs, axis, native, now_ns,
   # The old OFF response may precede the ready event. Its original transport
   # and native leases are validated by the caller; this bounded join arms none.
   return bool(0 < feedback.cold_main_wait_ns <= now_ns <= feedback.cold_main_wait_ns + SAFETY_MAX_AGE_NS and
-              str(axis.faultReason) in ('critical', 'none') and
+              str(axis.faultReason) in ('critical', 'availabilityPause', 'none') and
               (not native.requestedLateral and not native.lateralAllowed or
                str(axis.faultReason) == 'none' and axis.desiredLateral and
                native.requestedLateral and native.lateralAllowed))

openpilot/starpilot/longitudinal/tests/test_gm_volt_cc_policy.py

@@ -263,7 +263,7 @@ class TestVoltCc(unittest.TestCase):
       with self.assertRaises(Registered):
         Car(CI=ci, RI=RadarInterface(ci.CP))
     self.assertEqual(len(captured), 1)
-    self.assertEqual(set(captured[0]), {'pandaStates', 'carControl', 'onroadEvents', 'deviceState'})
+    self.assertEqual(set(captured[0]), {'pandaStates', 'carControl', 'onroadEvents', 'deviceState', 'autoSteeringLearning'})
 
   def test_actual_card_transport_epoch_and_individual_source_loss(self):
     controls, card, ci, packer, sent, base, offset = fixture()

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "42fc1f2e5fa03b715453bd816ac12bd7d82e88747c87f2bac2b229f4f1d134ba", "schema_version": 1}
+{"protected_inventory_sha256": "fca1c3fd9221979c32c2dd4081b050561aca2b12c58936388a543cb94e915ef4", "schema_version": 1}

tools/ci/run_vehicle_tests.py

@@ -41,6 +41,7 @@ AOL_LIFECYCLE_TARGETS = (
   "openpilot/starpilot/aol/tests/test_gm_steering_receive_order.py",
   "openpilot/starpilot/aol/tests/test_gm_main_response.py",
   "openpilot/starpilot/aol/tests/test_gm_held_native.py",
+  "openpilot/starpilot/aol/tests/test_gm_ordinary_warning.py",
   "openpilot/starpilot/aol/tests/test_gm_startup_main.py",
   "openpilot/starpilot/aol/tests/test_gm_pedal_order.py",
   "openpilot/starpilot/aol/tests/test_gm_cancel_remap.py",