Preserve Tesla screen intent and Palisade Cancel behavior

opendbc_repo/opendbc/car/hyundai/carstate.py

@@ -348,9 +348,10 @@ class CarState(CarStateBase):
       self.lkas11 = copy.copy(cp_cam.vl["LKAS11"])
     self.clu11 = copy.copy(cp.vl["CLU11"])
     from opendbc.car.hyundai.classic_long_aol import independent_buttons_qualified
+    from opendbc.car.hyundai.blended_stock_aol import qualified_alpha
     self.ev6_aol_samples = ()
     self.ev6_aol_source_healthy = False
-    if independent_buttons_qualified(self.CP):
+    if independent_buttons_qualified(self.CP) or qualified_alpha(self.CP, marked_only=True):
       from opendbc.can.parser import MAX_BAD_COUNTER
       source = cp.message_states.get(cp.dbc.name_to_msg['CLU11'].address)
       self.ev6_aol_sample_stamp_ns = int(source.timestamps[-1]) if source is not None and source.timestamps else 0

opendbc_repo/opendbc/car/tesla/carstate.py

@@ -9,7 +9,8 @@ from opendbc.car.tesla.values import CAR, DBC, CANBUS, GEAR_MAP, STEER_THRESHOLD
 
 class TeslaScreenCANParser(CANParser):
   def __init__(self):
-    super().__init__('tesla_model3_vehicle', [('UI_status2', 0)], CANBUS.vehicle)
+    # Screen gestures are optional; their silence is not a control-bus fault.
+    super().__init__('tesla_model3_vehicle', [('UI_status2', float('nan'))], CANBUS.vehicle)
 
   def update(self, strings, sendcan=False):
     if strings and not isinstance(strings[0], list | tuple):

opendbc_repo/opendbc/car/tesla/tests/test_screen_button.py

@@ -80,3 +80,36 @@ def test_literal_optional_screen_frames_baseline_edge_and_exact_dlc():
                                                               (structs.CarState.ButtonEvent.Type.lkas, True)]
   parser.update([(1_030_000_000, [(frame[0], frame[1][:-1], 1), (frame[0], frame[1], 0)])])
   assert cs.update_screen_button(parser) == []
+
+
+@pytest.mark.parametrize('screen_seen', [False, True])
+def test_optional_screen_silence_does_not_report_vehicle_bus_loss(screen_seen):
+  parser = TeslaScreenCANParser()
+  if screen_seen:
+    parser.update([(1_000_000_000, [(0x3DF, bytes(8), CANBUS.vehicle)])])
+  for stamp in (1_500_000_001, 12_000_000_000):
+    parser.update([(stamp, [(0x111, bytes(8), CANBUS.vehicle)])])
+    assert not parser.bus_timeout
+    assert parser.can_valid
+    assert parser.vl_all['UI_status2']['UI_activeTouchPoints'] == []
+
+
+@pytest.mark.parametrize('missing_bus', [None, CANBUS.party, CANBUS.autopilot_party])
+def test_screen_silence_keeps_required_vehicle_bus_timeouts(missing_bus):
+  cp = screen_params()
+  apply_screen_button(cp, True, False)
+  ci = CarInterface(cp)
+  packer = CANPacker('tesla_model3_party')
+  messages = {
+    CANBUS.party: ['DI_speed', 'DI_systemStatus', 'ESP_B', 'EPAS3S_sysStatus', 'ESP_status', 'UI_warning', 'DI_state'],
+    CANBUS.autopilot_party: ['SCCM_steeringAngleSensor', 'DAS_status', 'DAS_control', 'DAS_steeringControl', 'DAS_settings'],
+  }
+  for index in range(150):
+    frames = [packer.make_can_msg(name, bus, {}) for bus, names in messages.items()
+              if index < 50 or bus != missing_bus for name in names]
+    if index == 0:
+      frames.append((0x3DF, bytes(8), CANBUS.vehicle))
+    result = ci.update([(1_000_000_000 + index * 10_000_000, frames)])
+  assert result.canTimeout is (missing_bus is not None)
+  if missing_bus is None:
+    assert result.canValid

opendbc_repo/opendbc/safety/modes/hyundai_blended_stock_aol.h

@@ -158,10 +158,29 @@ static void blended_aol_rx(const CANPacket_t *msg) {
     bool pressed = false;
     if ((msg->bus == pt) && (msg->addr == 0x4F1U) && (GET_LEN(msg) == 4U)) {
       if ((msg->data[0] & 7U) == 4U) {
-        blended_aol_clear();
+        // Mixed LONG Pause/Resume still clears ordinary/LONG authority in the
+        // cruise hook. Retain only an already claimed, current lateral token.
+        // CANCEL cannot arm, renew a request/gesture lease, or clear an inhibit.
+        const bool source_current = (blended_aol_token_source < 3U) &&
+          (safety_get_ts_elapsed(now, blended_aol_source_ts[blended_aol_token_source]) <= 300000U);
+        const bool retain_lateral = blended_aol_long && !GET_BIT(msg, 3U) &&
+          blended_aol_token && blended_aol_claimed && blended_aol_session && heartbeat_engaged &&
+          blended_aol_main_seen && blended_aol_main &&
+          (safety_get_ts_elapsed(now, blended_aol_main_ts) <= 100000U) && source_current &&
+          blended_aol_request_seen && ((aol_host_axis_mask & 1U) != 0U) &&
+          (safety_get_ts_elapsed(now, aol_host_request_ts) <= AOL_HOST_REQUEST_TIMEOUT_US) &&
+          aol_rx_healthy() && !relay_malfunction && !safety_rx_checks_invalid;
+        if (!retain_lateral) {
+          blended_aol_clear();
+        }
+        // A simultaneous MAIN is not an arming gesture. The advancing physical
+        // source remains observable without changing its token or host lease.
+        blended_aol_pressed[0] = GET_BIT(msg, 3U);
+        blended_aol_source_ts[0] = now;
+      } else {
+        source = 0U;
+        pressed = GET_BIT(msg, 3U);
       }
-      source = 0U;
-      pressed = GET_BIT(msg, 3U);
     } else if ((msg->bus == pt) && (msg->addr == 0x391U) && (GET_LEN(msg) == 8U)) {
       source = 1U;
       pressed = GET_BIT(msg, 4U);

opendbc_repo/opendbc/safety/modes/tesla_screen_aol.h

@@ -13,6 +13,7 @@ static bool tesla_screen_belt_latched = false;
 static bool tesla_screen_eps_ready = false;
 static bool tesla_screen_session = false;
 static bool tesla_screen_autopark = false;
+static bool tesla_screen_acc_fault = false;
 static int tesla_screen_override = 0;
 
 static void tesla_screen_clear(void) {
@@ -32,6 +33,7 @@ static void tesla_screen_reset(void) {
   tesla_screen_eps_ready = false;
   tesla_screen_session = false;
   tesla_screen_autopark = false;
+  tesla_screen_acc_fault = false;
 }
 
 static uint8_t tesla_screen_request(void) {
@@ -54,7 +56,7 @@ static uint8_t tesla_screen_request(void) {
 
 static bool tesla_screen_physical_ready(void) {
   return tesla_screen_drive && tesla_screen_occupant_ready && tesla_screen_eps_ready &&
-    !steering_disengage && !tesla_screen_autopark;
+    !steering_disengage && !tesla_screen_autopark && !tesla_screen_acc_fault;
 }
 
 static uint8_t tesla_screen_permission(void) {
@@ -108,14 +110,17 @@ static void tesla_screen_rx(const CANPacket_t *msg, bool autopark) {
     }
     if (msg_matches(msg, 0x286U, 0U, 8U)) {
       const unsigned int state = (msg->data[1] >> 4U) & 7U;
+      tesla_screen_acc_fault = state == 5U;
       const bool cruise = ((state == 2U) || (state == 3U) || (state == 4U) || (state == 6U) || (state == 7U)) && !tesla_screen_autopark;
       const bool main_available = ((state == 1U) || cruise) && !tesla_screen_autopark;
       if (cruise && !tesla_screen_cruise) {
         tesla_screen_override = 0;
       }
-      if (tesla_screen_main && !main_available && !brake_pressed) {
+      // Stock availability cannot revoke an explicit screen selection.
+      if (tesla_screen_main && !main_available && !brake_pressed && (tesla_screen_override != 1)) {
         tesla_screen_override = -1;
       }
+      if (tesla_screen_acc_fault) { tesla_screen_clear(); }
       tesla_screen_main = main_available;
       tesla_screen_cruise = cruise;
     }

opendbc_repo/opendbc/safety/tests/test_hyundai_blended_alpha.py

@@ -313,8 +313,8 @@ class TestHyundaiBlendedAlphaAol(unittest.TestCase):
         self.assertEqual(self.request(1), 0)
         self.assertFalse(self.safety.get_controls_allowed())
 
-  def test_main_off_cancel_and_heartbeat_loss_revoke_physical_token(self):
-    for cause in ('main_off', 'cancel', 'heartbeat'):
+  def test_main_off_and_heartbeat_loss_revoke_physical_token(self):
+    for cause in ('main_off', 'heartbeat'):
       self.now = 1_000_000
       self.counter = 0
       self.reset()
@@ -325,13 +325,98 @@ class TestHyundaiBlendedAlphaAol(unittest.TestCase):
         continue
       if cause == 'main_off':
         self.feed(available=False)
-      elif cause == 'cancel':
-        self.feed(button=4)
       else:
         self.safety.set_aol_test_heartbeat(False)
       self.assertEqual(self.request(1), 0)
       self.assertFalse(self.safety.get_controls_allowed())
 
+  def test_cancel_retains_only_existing_claimed_lateral_and_stops_long(self):
+    for word in (0x2004, 0x2014):
+      for held in (7, 23):  # Both reported short and longer Pause/Resume holds.
+        self.counter, self.now = 0, 1_000_000
+        self.reset(word)
+        self.warm()
+        self.feed(lda=True)
+        self.assertEqual(self.request(1), 0 if self.release else 1)
+        if self.release:
+          continue
+        self.feed(button=2)
+        self.feed()
+        self.assertTrue(self.safety.get_controls_allowed())
+        self.assertEqual(self.request(3), 3)
+        for _ in range(held):
+          self.feed(button=4)
+          self.assertFalse(self.safety.get_controls_allowed())
+          self.assertEqual(self.request(1), 1)
+        self.feed()
+        self.assertFalse(self.safety.get_controls_allowed())
+        self.assertEqual(self.request(1), 1)
+        if self.bus:
+          self.primary_counter = 0
+          steering = TestHyundaiBlendedAlpha.primary(self, 1, True)
+        else:
+          from opendbc.car.hyundai.hyundaican import hyundai_checksum
+
+          def steering_checksum(message):
+            address, data, bus = message
+            return address, bytes([hyundai_checksum(data[1:8])]) + data[1:8], bus
+          steering = self.packer.make_can_msg_safety('LKAS11', 0,
+            {'CF_Lkas_MsgCount': 0, 'CR_Lkas_StrToqReq': 1, 'CF_Lkas_ActToi': 1}, fix_checksum=steering_checksum)
+        self.assertTrue(self.safety.safety_tx_hook(steering), 'Only retained LAT can authorize steering')
+        accel = self.packer.make_can_msg_safety('SCC11', self.bus, {'aReqRaw': .1})
+        self.assertFalse(self.safety.safety_tx_hook(accel))
+        self.assertEqual(self.request(0), 0)  # Configured cancellation remains authoritative.
+
+  def test_cancel_cannot_arm_or_restore_expired_or_withdrawn_authority(self):
+    for word in (0x2004, 0x2014):
+      for cause in ('cold', 'unclaimed', 'withdrawn', 'stale_request', 'heartbeat', 'main_off', 'relay'):
+        with self.subTest(word=word, cause=cause):
+          self.counter, self.now = 0, 1_000_000
+          self.reset(word)
+          self.warm()
+          if cause != 'cold':
+            self.feed(lda=True)
+          if cause not in ('cold', 'unclaimed'):
+            self.assertEqual(self.request(1), 0 if self.release else 1)
+          if self.release:
+            continue
+          if cause == 'withdrawn':
+            self.request(0)
+          elif cause == 'stale_request':
+            for _ in range(32):
+              self.feed()  # Fresh main and token source, without a host renewal.
+          elif cause == 'heartbeat':
+            self.safety.set_aol_test_heartbeat(False)
+          elif cause == 'relay':
+            self.safety.set_relay_malfunction(True)
+          self.feed(button=4, available=cause != 'main_off')
+          self.assertEqual(self.request(1), 0, cause)
+
+  def test_cancel_cannot_restore_expired_source_simultaneous_main_or_bad_rx(self):
+    for word in (0x2004, 0x2014):
+      for cause in ('source', 'simultaneous_main', 'bad_rx'):
+        with self.subTest(word=word, cause=cause):
+          self.counter, self.now = 0, 1_000_000
+          self.reset(word)
+          self.warm()
+          self.feed(lda=True)
+          self.assertEqual(self.request(1), 0 if self.release else 1)
+          if self.release:
+            continue
+          if cause == 'source':
+            for _ in range(32):
+              self.feed(omit='BCM_PO_11')
+              self.safety.aol_set_host_request(1)  # Fresh request, expired physical LDA source.
+          elif cause == 'bad_rx':
+            invalid = self.packer.make_can_msg_safety('TCS13', self.bus, {'AliveCounterTCS': self.counter % 8}, fix_checksum=checksum)
+            invalid[0].data[6] ^= 1
+            self.assertFalse(self.safety.safety_rx_hook(invalid))
+          packet = self.packer.make_can_msg_safety('CLU11', self.bus,
+            {'CF_Clu_CruiseSwState': 4, 'CF_Clu_CruiseSwMain': int(cause == 'simultaneous_main'),
+             'CF_Clu_AliveCnt1': self.counter % 16})
+          self.assertTrue(self.safety.safety_rx_hook(packet))
+          self.assertEqual(self.request(1), 0, cause)
+
   def test_cluster_lda_source_needs_its_own_neutral_edge(self):
     self.reset()
     self.warm()

opendbc_repo/opendbc/safety/tests/test_tesla_screen_button.py

@@ -33,7 +33,8 @@ class TestTeslaScreenButton(unittest.TestCase):
       ('DAS_steeringControl', 2, {'DAS_steeringControlType': stock_type}),
       ('DI_speed', 0, {'DI_vehicleSpeed': 0}),
       ('ESP_B', 0, {'ESP_vehicleSpeed': 0, 'ESP_wheelSpeedsQF': 1, 'ESP_vehicleStandstillSts': 1}),
-      ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': eps, 'EPAS3S_handsOnLevel': hands}),
+      ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': eps, 'EPAS3S_handsOnLevel': hands,
+                             'EPAS3S_torsionBarTorque': 0}),
       ('DI_systemStatus', 0, {'DI_gear': gear, 'DI_accelPedalPos': 0}),
       ('ESP_status', 0, {'ESP_driverBrakeApply': 2 if brake else 1}),
       ('UI_warning', 0, {'anyDoorOpen': int(door), 'buckleStatus': int(belt)}),
@@ -106,6 +107,64 @@ class TestTeslaScreenButton(unittest.TestCase):
             self.assertEqual(self.permission(), 0)
           self.arm()
 
+  def test_availability_loss_preserves_only_explicit_screen_arm(self):
+    tx_packer = CANPackerSafety('tesla_model3_party')
+    for word in (512, 768, 1536, 1792):
+      with self.subTest(word=word):
+        self.start(word)
+        self.arm()
+        self.touch(0)
+        for state in (1, 0, 1, 0, 1):
+          self.feed(state=state)
+          self.assertEqual(self.permission(3), 1)
+          positive = tx_packer.make_can_msg_safety('DAS_control', 0, {'DAS_accelMin': 1, 'DAS_accelMax': 1})
+          self.assertFalse(self.lib.safety_tx_hook(positive))
+        self.touch(3)
+        self.assertEqual(self.permission(), 0)
+        for state in (0, 1, 0, 1):
+          self.feed(state=state)
+          self.assertEqual(self.permission(), 0)
+
+        self.start(word)
+        self.feed(state=1)
+        self.assertEqual(self.permission(), 1)
+        self.feed(state=0)
+        self.assertEqual(self.permission(), 0)
+        self.feed(state=1)
+        self.assertEqual(self.permission(), 0)
+
+  def test_cruise_fault_clears_screen_arm_and_blocks_gestures(self):
+    for word in (512, 768, 1536, 1792):
+      with self.subTest(word=word):
+        self.start(word)
+        self.arm()
+        self.feed(state=1)
+        self.feed(state=5)
+        self.assertEqual(self.permission(), 0)
+        self.touch(0)
+        self.touch(3)
+        self.assertEqual(self.permission(), 0)
+        for state in (0, 1):
+          self.feed(state=state)
+          self.assertEqual(self.permission(), 0)
+        self.arm()
+
+  def test_availability_loss_cannot_hide_physical_fault_or_brake(self):
+    for word in (512, 768, 1536, 1792):
+      faults = [{'door': True}, {'eps': 3}, {'hands': 3}, {'gear': 0}, {'autopark': 3}]
+      if word & 1024:
+        faults.append({'brake': True})
+      for fault in faults:
+        with self.subTest(word=word, fault=fault):
+          self.start(word)
+          self.arm()
+          self.feed(state=1)
+          self.feed(state=0, **fault)
+          self.assertEqual(self.permission(), 0)
+          for state in (0, 1, 0, 1):
+            self.feed(state=state)
+            self.assertEqual(self.permission(), 0)
+
   def test_brake_disengagement_cannot_be_undone_by_availability(self):
     self.start(1536)
     self.arm()

openpilot/selfdrive/car/card.py

@@ -60,6 +60,7 @@ from opendbc.car.ford.aol import (
   temporary_restriction as ford_temporary_restriction,
 )
 from opendbc.car.hyundai.canfd_angle_aol import native_observation, temporary_restriction, qualified as qualified_angle_aol
+from opendbc.car.hyundai.blended_stock_aol import qualified_alpha as qualified_blended_alpha_aol
 from opendbc.car.hyundai.canfd_owner import config_index as hyundai_canfd_config_index
 from opendbc.car.hyundai.captured_adrv_startup import CapturedADRVStartup
 from openpilot.starpilot.car.nissan.aol import (
@@ -615,9 +616,10 @@ class Car:
         self.aol_card_intent, cancel_source_ns=cancel_source_ns, now_ns=now_ns,
         producer=getattr(self, 'slc_producer_session', ''))
       cancel_feedback = self.aol_transport_feedback
-      if (hkg_cancel_retained and self.CP.carFingerprint in (HYUNDAI_CAR.KIA_EV9, HYUNDAI_CAR.HYUNDAI_IONIQ_5_PE) and
+      if (hkg_cancel_retained and (self.CP.carFingerprint in (HYUNDAI_CAR.KIA_EV9, HYUNDAI_CAR.HYUNDAI_IONIQ_5_PE) or
+          qualified_blended_alpha_aol(self.CP, marked_only=True)) and
           cancel_feedback is not None and not cancel_feedback.longitudinal_cancel_ns):
-        # The pure guard proved this exact CCNC raw CANCEL/FULL join. Bind
+        # The pure guard proved this exact raw CANCEL/FULL join. Bind
         # its first local receipt for the existing bounded withdrawal handshake;
         # held samples never renew it or any source/native/output deadline.
         if cancel_source_ns > cancel_feedback.last_cancel_source_ns:

openpilot/starpilot/aol/intent.py

@@ -141,10 +141,11 @@ def disarming_fault(events, CS, *, temporary_ui_process_failure: bool = False,
 
 class AolCardIntent:
   def __init__(self, settings: AolSettings, *, explicit_latch: bool = False, lkas_button_supported: bool = True,
-               independent_cruise_buttons: bool = False):
+               independent_cruise_buttons: bool = False, preserve_unassigned_cancel: bool = False):
     self.settings = settings
     self.explicit_latch = explicit_latch
     self.independent_cruise_buttons = independent_cruise_buttons
+    self.preserve_unassigned_cancel = preserve_unassigned_cancel
     self.lkas_button_supported = lkas_button_supported
     self.allowed_latch = False
     # Observational only: records the exact branch that changed True to False.
@@ -341,12 +342,12 @@ class AolCardIntent:
                        (not independent_buttons or int(ButtonType.mainCruise) not in consumed_buttons)) or
                       (self._lkas_held and lkas_toggle and
                        (not independent_buttons or int(ButtonType.lkas) not in consumed_buttons)))
-      if not previously_held and gesture_held and (not independent_buttons or not cancel_pressed):
+      if not previously_held and gesture_held and (not (independent_buttons or self.preserve_unassigned_cancel) or not cancel_pressed):
         self._last_latch_edge_ns = now_ns
       if cancel_pressed:
         # An unassigned cruise CANCEL cannot toggle, arm or renew independent
         # lateral. Keep the existing withdrawal for configured CANCEL actions.
-        if not self.independent_cruise_buttons or any(self.settings.cancel_actions):
+        if not (self.independent_cruise_buttons or self.preserve_unassigned_cancel) or any(self.settings.cancel_actions):
           self._record_latch_clear('cancel_press')
           self.allowed_latch = False
       elif not self._gesture_neutral_seen:

openpilot/starpilot/aol/tests/test_palisade_cancel.py

@@ -0,0 +1,214 @@
+"""Mixed Palisade Pause/Resume retains only existing, current lateral authority."""
+from dataclasses import replace
+from types import SimpleNamespace
+from typing import cast
+import pytest
+from opendbc.car import structs
+from openpilot.cereal import log
+from openpilot.selfdrive.selfdrived.events import Events
+from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
+from openpilot.starpilot.aol.intent import AolSettings, AOL_TOGGLE
+from openpilot.starpilot.aol.runtime import AxisDecision
+from openpilot.starpilot.aol.wire import IntentState, SafetyState, PandaSlot
+from openpilot.starpilot.aol.tests.test_classic_brake_withdrawal import profile
+from openpilot.starpilot.car.hyundai.aol import create_intent
+
+
+class Messages(dict):
+  ready = True
+  def all_checks(self, services=None):
+    return self.ready
+
+
+@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii'))
+@pytest.mark.parametrize('case', ['retain', 'prior_off', 'settings_off', 'configured_cancel', 'intent_off', 'stale_source',
+                                 'native_denied', 'native_expired', 'wrong_session', 'wrong_inventory', 'hard_event',
+                                 'model', 'dm', 'reverse', 'steer_fault'])
+def test_actual_selfdrived_cancel_keeps_only_current_owned_lateral(identity, case):
+  now = 4_000_000_000
+  cp = profile(identity)
+  config = cp.safetyConfigs[0]
+  model, word = int(config.safetyModel.raw), int(config.safetyParam)
+  intent = IntentState('card', 1, now, now, now + 200_000_000, True, False, False, True, True)
+  slot = PandaSlot(0, 'panda', model, word, cp.alternativeExperience)
+  inventory = (slot,)
+  native = SafetyState(1, True, now, now + 200_000_000, model, word, True, False, True, False,
+                       'panda', 'axis', now, inventory)
+  cs = structs.CarState(canValid=True, gearShifter='drive', vEgo=10.)
+  cs.cruiseState.available = True
+  cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+  sm = Messages(driverMonitoringState=SimpleNamespace(alertLevel=0, lockout=False, alwaysOnLockout=False),
+                extrinsicsCalibration=SimpleNamespace(calStatus=log.ExtrinsicsCalibration.Status.calibrated))
+  sd = SimpleNamespace(CP=cp, active=False, enabled=False, initialized=True, sm=sm,
+                       aol_settings=AolSettings(True, 0., AOL_TOGGLE, 0, (0,0,0), (0,0,0)),
+                       aol_axis_decision=AxisDecision(lateral_active=True, native_acknowledged=True),
+                       aol_car_state_log_ns=now, aol_session_id='axis', aol_dm_lateral_inhibit=False,
+                       nostalgia_paddle_cancel=False, events=Events())
+  sd.events.add(log.OnroadEvent.EventName.buttonCancel)
+  if case == 'prior_off':
+    sd.aol_axis_decision = AxisDecision()
+  elif case == 'settings_off':
+    sd.aol_settings = replace(sd.aol_settings, enabled=False)
+  elif case == 'configured_cancel':
+    sd.aol_settings = replace(sd.aol_settings, cancel_actions=(3,0,0))
+  elif case == 'intent_off':
+    intent = replace(intent, allowedLatch=False)
+  elif case == 'stale_source':
+    sd.aol_car_state_log_ns = now - 30_000_001
+    intent = replace(intent, carStateLogMonoTime=sd.aol_car_state_log_ns)
+  elif case == 'native_denied':
+    native = replace(native, lateralAllowed=False)
+  elif case == 'native_expired':
+    native = replace(native, validUntilMonoTime=now-1)
+  elif case == 'wrong_session':
+    native = replace(native, axisSessionId='other')
+  elif case == 'wrong_inventory':
+    native = replace(native, pandaInventory=(*inventory[:-1], replace(slot, safetyParam=word ^ 0x80)))
+  elif case == 'hard_event':
+    sd.events.add(log.OnroadEvent.EventName.controlsMismatch)
+  elif case == 'model':
+    sm.ready = False
+  elif case == 'dm':
+    sm['driverMonitoringState'].lockout = True
+  elif case == 'reverse':
+    cs.gearShifter = 'reverse'
+  elif case == 'steer_fault':
+    cs.steerFaultPermanent = True
+  result = SelfdriveD._decide_aol_axes(cast(SelfdriveD, sd), cs, intent, native, now_ns=now)
+  assert result.lateral_active == (case == 'retain')
+  assert not result.longitudinal_active
+  assert log.OnroadEvent.EventName.buttonCancel in sd.events.names
+
+
+@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii'))
+@pytest.mark.parametrize('duration', (61_047_724, 221_004_624))
+@pytest.mark.parametrize('cancel_actions', ((0, 0, 0), (3, 0, 0), (0, 3, 0), (0, 0, 3)))
+def test_host_cancel_hold_keeps_only_unassigned_preexisting_latch(identity, duration, cancel_actions):
+  cp = profile(identity)
+  settings = AolSettings(True, 0., AOL_TOGGLE, 0, (0, 0, 0), cancel_actions)
+  owner = create_intent(cp, settings)
+  cs = structs.CarState(canValid=True, gearShifter='drive')
+  cs.cruiseState.available = True
+  owner.update(cs, now_ns=1_000_000_000)
+  cs.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+  owner.update(cs, now_ns=1_010_000_000)
+  assert owner.allowed_latch
+  edge, arm = owner._last_latch_edge_ns, owner.physical_arm_ns
+  cs.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=False)]
+  owner.update(cs, now_ns=1_020_000_000)
+  for stamp, events in ((1_030_000_000, [('cancel', True)]),
+                        (1_040_000_000, []), (1_030_000_000 + duration, [('cancel', False)])):
+    cs.buttonEvents = [structs.CarState.ButtonEvent(type=kind, pressed=pressed) for kind, pressed in events]
+    owner.update(cs, now_ns=stamp)
+    assert owner.allowed_latch == (not any(cancel_actions))
+    assert (owner._last_latch_edge_ns, owner.physical_arm_ns) == (edge, arm)
+  assert not owner.independent_cruise_buttons  # MAIN/LKAS semantics are not changed.
+
+
+@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii'))
+@pytest.mark.parametrize('simultaneous', (None, 'lkas', 'mainCruise'))
+def test_cold_cancel_never_arms_or_renews_gesture(identity, simultaneous):
+  owner = create_intent(profile(identity), AolSettings(True, 0., AOL_TOGGLE, 0, (0,0,0), (0,0,0)))
+  cs = structs.CarState(canValid=True, gearShifter='drive')
+  cs.cruiseState.available = True
+  owner.update(cs, now_ns=1_000_000_000)
+  buttons = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+  if simultaneous:
+    buttons.append(structs.CarState.ButtonEvent(type=simultaneous, pressed=True))
+  cs.buttonEvents = buttons
+  owner.update(cs, now_ns=1_010_000_000)
+  assert not owner.allowed_latch
+  assert owner._last_latch_edge_ns == owner.physical_arm_ns == 0
+
+
+@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii'))
+def test_card_retains_original_cancel_receipt_until_lateral_only_ack(identity):
+  """Real Card caller, synthetic transport; parser receipt is tested separately."""
+  from unittest.mock import Mock, patch
+  from opendbc.car.hyundai.blended_longitudinal import BlendedStartup, candidate_from_stock
+  from opendbc.car.hyundai.tests.test_palisade_2023 import params as mixed_params
+  from openpilot.common.params import Params
+  from openpilot.common.prefix import OpenpilotPrefix
+  from openpilot.starpilot.aol.tests.test_ioniq_availability import IoniqAvailabilityTests, car_state
+
+  case = IoniqAvailabilityTests(methodName='runTest')
+  with OpenpilotPrefix():
+    try:
+      stock = mixed_params('hdaii' if identity == 'palisade_hdaii' else 'hdai')
+      cp = candidate_from_stock(stock, alpha_requested=True, native_qualified=True)
+      startup = BlendedStartup(stock, cp.as_reader().as_builder(), (Mock(), Mock()))
+      startup.prepare(admission=lambda: True)
+      case.addCleanup(startup.close)
+      Params().put('LKASButtonControl', AOL_TOGGLE, block=True)
+      sd, _, drive, card, receive = case._ioniq_availability_callers(True, cp=cp, startup_owner=startup)
+      sd.aol_settings = card.aol_settings
+      source = 10_020_000_000
+      cs = car_state()
+      cs.cruiseState.available = True
+      cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+      # Native withdrawal arrives before the next SelfdriveD publication.
+      drive(2, native_request_lat=True, native_request_long=True, long_allowed=False,
+            panda_controls_allowed=False)
+      old_axis = sd.pm.messages['aolAxisState'].as_builder()
+      old_axis.logMonoTime = old_axis.aolAxisState.observedMonoTime = source - 1_000_000
+      old_axis.aolAxisState.sourceCarStateMonoTime = source - 1_000_000
+      old_axis.aolAxisState.validUntilMonoTime = source + 29_000_000
+      old_axis.aolAxisState.mode = 'combined'
+      old_axis.aolAxisState.desiredLongitudinal = old_axis.aolAxisState.longitudinalActive = True
+      with patch.object(card, 'aol_cancel_source', return_value=source):
+        assert receive(source, axis=old_axis.as_reader(), state=cs)
+      assert card.aol_transport_feedback.longitudinal_cancel_ns == source
+      # No second physical press, no newer axis yet, now the cancellation event.
+      event = log.Event.new_message(logMonoTime=source)
+      event.init('onroadEvents', 1)
+      event.onroadEvents[0].name = 'buttonCancel'
+      event.onroadEvents[0].userDisable = True
+      event.onroadEvents[0].noEntry = True
+      cs.buttonEvents = []
+      with patch.object(card, 'aol_cancel_source', return_value=0):
+        assert receive(source + 10_000_000, event=event.as_reader(), state=cs)
+      assert card.aol_transport_feedback.longitudinal_cancel_ns == source
+      assert not card.aol_transport_feedback.failed
+      # SelfdriveD receives the same physical press; its actual LAT-only output
+      # and the native LAT-only request close the bounded withdrawal handshake.
+      cancel = car_state()
+      cancel.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+      axis, state, command = drive(4, observed_state=cancel, native_request_lat=True,
+        native_request_long=False, long_allowed=False, panda_controls_allowed=False)
+      assert not state.enabled and axis.lateralActive and command.latActive
+      assert not axis.longitudinalActive and not command.longActive
+      with patch.object(card, 'aol_cancel_source', return_value=0):
+        assert receive(source + 20_000_000, axis=sd.pm.messages['aolAxisState'],
+                       event=sd.pm.messages['onroadEvents'], state=cs)
+      assert card.aol_card_intent.allowed_latch
+      assert card.aol_transport_feedback.longitudinal_cancel_ns == 0
+    finally:
+      case.doCleanups()
+
+
+@pytest.mark.parametrize('identity', ('palisade_hdai', 'palisade_hdaii'))
+@pytest.mark.parametrize('fault', ('none', 'wrong_bus', 'stale', 'unmarked', 'stock'))
+def test_cancel_source_is_bound_to_real_mixed_parser_receipt(identity, fault):
+  from opendbc.can import CANPacker
+  from opendbc.car import Bus
+  from opendbc.car.hyundai.interface import CarInterface
+  from openpilot.selfdrive.car.card import Car
+  cp = profile(identity)
+  if fault == 'unmarked':
+    cp.alternativeExperience = 0
+  elif fault == 'stock':
+    cp.openpilotLongitudinalControl, cp.pcmCruise = False, True
+    cp.safetyConfigs[0].safetyParam &= ~4
+  ci = CarInterface(cp)
+  parser = ci.can_parsers[Bus.pt]
+  packer = CANPacker(parser.dbc.name)
+  for tick in range(15):
+    now = 2_000_000_000 + tick * 20_000_000
+    packet = packer.make_can_msg('CLU11', 2 if fault == 'wrong_bus' else parser.bus,
+      {'CF_Clu_CruiseSwState': 4 if tick == 14 else 0, 'CF_Clu_AliveCnt1': tick})
+    cs = ci.update([(now, [packet])])
+  if fault == 'wrong_bus':
+    cs.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+  received = now - 30_000_001 if fault == 'stale' else now
+  caller = SimpleNamespace(CI=ci, _aol_can_observation=(cs, received, now, now))
+  assert Car.aol_cancel_source(cast(Car, caller), cs, now) == (now if fault == 'none' else 0)

openpilot/starpilot/aol/transport_pause.py

@@ -805,11 +805,12 @@ class TransportPauseFeedback:
   def _longitudinal_cancel_current(self, sm, cp, cs, axis, now_ns, *, latched, cancel_source_ns, withdrawal_required=True) -> SafetyState | None:
     from opendbc.car.hyundai.canfd_stock_aol import qualified_long as qualified
     from openpilot.starpilot.car.hyundai.aol import qualified_ccnc_long_aol as ev9_qualified
+    from opendbc.car.hyundai.blended_stock_aol import qualified_alpha as mixed_qualified
     from opendbc.car.structs import car
 
     if (not latched or self.failed or self.paused or self.pending_event or
         not self.session or str(axis.sessionId) != self.session or
-        not (qualified(cp, marked_only=True) or ev9_qualified(cp, marked_only=True)) or
+        not (qualified(cp, marked_only=True) or ev9_qualified(cp, marked_only=True) or mixed_qualified(cp, marked_only=True)) or
         not self.lateral_baseline_ns or not 0 <= now_ns - self.lateral_baseline_ns <= AXIS_MAX_AGE_NS or
         str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or
         not axis.desiredLateral or not axis.lateralActive or cs.accFaulted or

openpilot/starpilot/car/hyundai/aol.py

@@ -388,7 +388,8 @@ def physical_cancel_lateral_allowed(CP, CS, intent, native, *, settings, previou
         not any(e.type == car.CarState.ButtonEvent.Type.cancel for e in CS.buttonEvents)):
       return False
     ev6_long = CP.carFingerprint == HyundaiCAR.KIA_EV6 and qualified_canfd_long(CP, marked_only=True)
-    return bool((ev6_long or independent_cruise_buttons_supported(CP, marked_only=True)) and
+    return bool((ev6_long or qualified_blended_alpha(CP, marked_only=True) or
+                 independent_cruise_buttons_supported(CP, marked_only=True)) and
       previous is not None and previous.lateral_active and previous.native_acknowledged and
       intent is not None and intent.settingsQualified and intent.allowedLatch and intent.lateralArmed and
       not intent.pauseLateral and intent.producerSessionId and intent.carStateLogMonoTime == source_ns and
@@ -427,9 +428,10 @@ def retained_shared_long_cancel_intent(CP, CS, sm, feedback, owner, *, cancel_so
       from openpilot.starpilot.car.hyundai.ev6_intent import Ev6CardIntent
       ev6_long = bool(isinstance(owner, Ev6CardIntent) and owner._buttons_current and
                       owner.physical.authorized and owner._last_latch_edge_ns > 0)
-    if (not (ev6_long or qualified_sonata_independent(CP) or qualified_ccnc_long_aol(CP, marked_only=True) or
+    mixed_long = qualified_blended_alpha(CP, marked_only=True)
+    if (not (ev6_long or mixed_long or qualified_sonata_independent(CP) or qualified_ccnc_long_aol(CP, marked_only=True) or
              CP.carFingerprint in INDEPENDENT_LONG_BUTTON_CARS and qualified_canfd_long(CP, marked_only=True)) or
-        owner is None or not owner.explicit_latch or not (owner.independent_cruise_buttons or ev6_long) or
+        owner is None or not owner.explicit_latch or not (owner.independent_cruise_buttons or ev6_long or mixed_long) or
         not owner.allowed_latch or owner.pause_lateral or owner.pause_longitudinal or owner._fault_inhibit):
       return False
     settings = owner.settings

openpilot/starpilot/car/hyundai/classic_long_intent.py

@@ -1,4 +1,5 @@
 from opendbc.car.hyundai.classic_long_aol import independent_buttons_qualified
+from opendbc.car.hyundai.blended_stock_aol import qualified_alpha
 from openpilot.starpilot.aol.intent import AolCardIntent
 
 
@@ -7,7 +8,8 @@ class ClassicLongCardIntent(AolCardIntent):
 
   def __init__(self, cp, settings):
     super().__init__(settings, explicit_latch=True, independent_cruise_buttons=(
-      independent_buttons_qualified(cp, marked_only=False)))
+      independent_buttons_qualified(cp, marked_only=False)),
+      preserve_unassigned_cancel=qualified_alpha(cp))
     self.requires_fault_observation = True
     self.observe_stock_engagement = True
 

openpilot/starpilot/car/tesla/screen_intent.py

@@ -21,7 +21,9 @@ class TeslaScreenCardIntent(AolCardIntent):
     super().update(filtered.as_reader(), **kwargs)
     engaged = bool(CS.cruiseState.enabled)
     available = bool(CS.cruiseState.available)
-    if (engaged and not self.previous_enabled) or (self.previous_available and not available and not CS.brakePressed):
+    # Stock cruise availability does not revoke an explicit screen selection.
+    if (engaged and not self.previous_enabled) or (self.previous_available and not available and not CS.brakePressed and
+                                                  self.screen_override is not True):
       self.screen_override = None if engaged and not self.previous_enabled else False
       self.brake_blocked = False
       self.pause_lateral = False

openpilot/starpilot/car/tesla/tests/test_screen_intent.py

@@ -181,3 +181,52 @@ def test_availability_off_still_allows_a_new_ordinary_cruise_engagement():
   assert not owner.allowed_latch
   owner.update(state(enabled=True).as_reader(), now_ns=50)
   assert owner.allowed_latch and owner.screen_override is None
+
+
+@pytest.mark.parametrize('brake_disengage', [False, True])
+def test_stock_cruise_availability_cannot_erase_existing_screen_arm(brake_disengage):
+  owner = intent(brake_disengage)
+  owner.update(state(available=False, press=False).as_reader(), now_ns=10)
+  owner.update(state(available=False, press=True).as_reader(), now_ns=20)
+  owner.update(state(available=True, press=False).as_reader(), now_ns=30)
+  assert owner.screen_override is True
+  for index, available in enumerate((False, True, False, True)):
+    cs = state(available=available)
+    cs.gasPressed = bool(index % 2)
+    owner.update(cs.as_reader(), now_ns=40 + index)
+    assert owner.allowed_latch and owner.screen_override is True
+  owner.update(state(press=True).as_reader(), now_ns=50)
+  assert not owner.allowed_latch and owner.pause_lateral
+  owner.update(state(available=False, press=False).as_reader(), now_ns=60)
+  owner.update(state(available=True).as_reader(), now_ns=70)
+  assert not owner.allowed_latch
+
+
+def test_cruise_baseline_availability_loss_does_not_create_screen_arm():
+  owner = intent()
+  owner.update(state().as_reader(), now_ns=10)
+  assert owner.allowed_latch and owner.screen_override is None
+  owner.update(state(available=False).as_reader(), now_ns=20)
+  assert not owner.allowed_latch and owner.screen_override is False
+  owner.update(state().as_reader(), now_ns=30)
+  assert not owner.allowed_latch
+
+
+@pytest.mark.parametrize('fault', ['canTimeout', 'doorOpen', 'steerFaultPermanent', 'steeringDisengage', 'accFaulted', 'cancel', 'brake'])
+def test_availability_loss_cannot_hide_fault_cancel_or_configured_brake(fault):
+  owner = intent(brake=fault == 'brake')
+  owner.update(state(available=False, press=False).as_reader(), now_ns=10)
+  owner.update(state(available=False, press=True).as_reader(), now_ns=20)
+  owner.update(state(press=False).as_reader(), now_ns=30)
+  cs = state(available=False)
+  if fault == 'cancel':
+    cs.buttonEvents = [structs.CarState.ButtonEvent(type=ButtonType.cancel, pressed=True)]
+  elif fault == 'brake':
+    cs.brakePressed = True
+  else:
+    setattr(cs, fault, True)
+  owner.update(cs.as_reader(), now_ns=40)
+  assert not owner.allowed_latch
+  for index, available in enumerate((False, True, False, True)):
+    owner.update(state(available=available).as_reader(), now_ns=50 + index)
+    assert not owner.allowed_latch

openpilot/starpilot/car/tesla/tests/test_screen_pedal_continuity.py

@@ -0,0 +1,100 @@
+"""Independent screen arming through real parser, intent, native and axis callers."""
+from types import SimpleNamespace
+
+import pytest
+
+from opendbc.can import CANPacker
+from opendbc.car.tesla.interface import CarInterface
+from opendbc.car.tesla.screen_button import apply_screen_button
+from opendbc.car.tesla.tests.test_screen_button import screen_params
+from opendbc.safety.tests.libsafety import libsafety_py
+from openpilot.cereal import log
+from openpilot.selfdrive.selfdrived.events import Events
+from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
+from openpilot.starpilot.aol.intent import AolSettings
+from openpilot.starpilot.aol.runtime import AxisDecision
+from openpilot.starpilot.aol.wire import IntentState, PandaSlot, SafetyState
+from openpilot.starpilot.car.tesla.screen_intent import TeslaScreenCardIntent
+
+
+class HealthyPeers(dict):
+  def all_checks(self, services):
+    return all(service in self for service in services)
+
+
+@pytest.mark.parametrize('cooperative', [False, True])
+@pytest.mark.parametrize('brake_disengage', [False, True])
+def test_explicit_screen_axis_survives_sparse_touch_and_stock_availability(cooperative, brake_disengage):
+  cp = screen_params()
+  apply_screen_button(cp, True, brake_disengage)
+  cp.safetyConfigs[0].safetyParam |= 256 * cooperative
+  cp.alternativeExperience = 32
+  ci = CarInterface(cp)
+  owner = TeslaScreenCardIntent(cp, AolSettings(True, 0., 0, 0, (0, 0, 0), (0, 0, 0)))
+  packer = CANPacker('tesla_model3_party')
+  peers = HealthyPeers(modelV2=SimpleNamespace(),
+                       extrinsicsCalibration=SimpleNamespace(calStatus=log.ExtrinsicsCalibration.Status.calibrated),
+                       driverMonitoringState=SimpleNamespace(lockout=False, alwaysOnLockout=False, alertLevel=0))
+  sd = SimpleNamespace(CP=cp, active=False, enabled=False, initialized=True, sm=peers, aol_settings=owner.settings,
+                       aol_axis_decision=AxisDecision(), aol_car_state_log_ns=0, aol_session_id='screen-test',
+                       aol_dm_lateral_inhibit=False, nostalgia_paddle_cancel=False, aol_cruise_main_required=False,
+                       events=Events())
+  native = libsafety_py.libsafety
+  native.init_tests()
+  native.set_alternative_experience(32)
+  native.set_safety_hooks(10, cp.safetyConfigs[0].safetyParam)
+  native.set_aol_test_heartbeat(True)
+  try:
+    for tick in range(250):
+      now = 1_000_000_000 + tick * 10_000_000
+      native.set_timer(now // 1000)
+      stock_state = 1 if 40 <= tick < 80 or 140 <= tick < 180 else 0
+      messages = (
+        ('DAS_control', 2, {}), ('DAS_steeringControl', 2, {}),
+        ('SCCM_steeringAngleSensor', 2, {}), ('DAS_status', 2, {}), ('DAS_settings', 2, {}),
+        ('DI_speed', 0, {'DI_vehicleSpeed': 0}),
+        ('ESP_B', 0, {'ESP_vehicleSpeed': 0, 'ESP_wheelSpeedsQF': 1, 'ESP_vehicleStandstillSts': 1}),
+        ('EPAS3S_sysStatus', 0, {'EPAS3S_internalSAS': 0, 'EPAS3S_eacStatus': 1,
+                               'EPAS3S_handsOnLevel': 3 if tick >= 240 else 0, 'EPAS3S_torsionBarTorque': 0}),
+        ('DI_systemStatus', 0, {'DI_gear': 4, 'DI_accelPedalPos': 20 if tick % 30 < 15 else 0}),
+        ('ESP_status', 0, {'ESP_driverBrakeApply': 2 if not brake_disengage and 100 <= tick < 120 else 1}),
+        ('UI_warning', 0, {'buckleStatus': 1}),
+        ('DI_state', 0, {'DI_cruiseState': stock_state}),
+      )
+      frames = [packer.make_can_msg(name, bus, values) for name, bus, values in messages]
+      if tick in (20, 21):
+        touch = bytearray(8)
+        touch[3] = 3 if tick == 21 else 0
+        frames.append((0x3DF, bytes(touch), 1))
+      for address, data, bus in frames:
+        assert native.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data))
+      native.safety_tick()
+      cs = ci.update([(now, frames)]).as_reader()
+      if tick > 1:
+        assert cs.canValid and not cs.canTimeout
+      owner.update(cs, now_ns=now)
+      allowed, pause, long_pause = owner.output(cs)
+      intention = IntentState('screen-card', tick + 1, now, now, now + 30_000_000, allowed, pause, long_pause, True, allowed)
+      permission = native.aol_get_permission_mask()
+      request = native.aol_get_request_mask()
+      safety = SafetyState(1, True, now, now + 200_000_000, 10, cp.safetyConfigs[0].safetyParam,
+                           bool(permission & 1), bool(permission & 2), bool(request & 1), bool(request & 2),
+                           'test-panda', 'screen-test', now,
+                           (PandaSlot(0, 'test-panda', 10, cp.safetyConfigs[0].safetyParam, 32, bool(native.get_controls_allowed())),))
+      sd.aol_car_state_log_ns = now
+      sd.events = Events()
+      sd.events.add(log.OnroadEvent.EventName.pcmDisable)
+      if not cs.cruiseState.available:
+        sd.events.add(log.OnroadEvent.EventName.wrongCarMode)
+      decision = SelfdriveD._decide_aol_axes(sd, cs, intention, safety, now_ns=now)
+      sd.aol_axis_decision = decision
+      native.aol_set_host_request(int(decision.desired_lateral))
+      assert not decision.longitudinal_active
+      if 22 <= tick < 240:
+        assert decision.lateral_active
+      elif tick >= 240:
+        assert not decision.lateral_active
+  finally:
+    native.init_tests()
+    native.set_alternative_experience(0)
+    native.set_safety_hooks(0, 0)