Restore HKG stock angle arming and steering recovery

opendbc_repo/opendbc/car/hyundai/angle_tx_recovery.py

@@ -0,0 +1,43 @@
+"""Correlate recent local steering rejection echoes; never authorize output."""
+from collections import deque
+import time
+
+
+class AngleTxRecovery:
+  def __init__(self, address, bus, clock=None):
+    self.clock = time.monotonic_ns if clock is None else clock
+    self.address = address
+    self.bus = bus
+    self.sent = deque(maxlen=4)
+    self.last_packet_ns = 0
+    self.pending_frame = -1
+    self.pending_sent_ns = -1
+
+  def record(self, frames, frame):
+    now_ns = self.clock()
+    for address, data, bus in frames:
+      if address == self.address and bus == self.bus and len(data) == 32 and (data[9] >> 4 & 3) == 2:
+        self.sent.append((frame, now_ns, bytes(data)))
+
+  def observe(self, packets, frame):
+    now_ns = self.clock()
+    for stamp, frames in packets:
+      if stamp <= self.last_packet_ns:
+        continue
+      self.last_packet_ns = stamp
+      for address, data, bus in frames:
+        if address != self.address or bus != self.bus + 0xC0 or len(data) != 32:
+          continue
+        # Consume an exact recent command once. The original local send clock
+        # also bounds delayed echoes when controller updates have been paused.
+        match = next(((sent_frame, sent_ns, sent_data) for sent_frame, sent_ns, sent_data in self.sent
+                      if 0 < frame - sent_frame <= 3 and 0 <= now_ns - sent_ns <= 30_000_000 and sent_data == bytes(data)), None)
+        if match is not None:
+          self.sent.remove(match)
+          self.pending_frame = frame
+          self.pending_sent_ns = match[1]
+
+  def take_rejection(self, frame):
+    pending = self.pending_frame == frame and 0 <= self.clock() - self.pending_sent_ns <= 30_000_000
+    self.pending_frame = -1
+    return pending

opendbc_repo/opendbc/car/hyundai/carcontroller.py

@@ -34,7 +34,8 @@ from opendbc.car.hyundai.g90_lead import G90LeadState, eligible as g90_lead_elig
 from opendbc.car.hyundai.gv70_camera_lead import eligible as gv70_lead_eligible
 from opendbc.car.hyundai.canfd_lead import select_lead, ABSENT
 from opendbc.car.hyundai.torque_ev_scc import eligible as torque_ev_scc_eligible, options as torque_ev_scc_options, clock_boot_ns, fallback_lead
-from opendbc.car.hyundai.ccnc_ev_stock import replacement_requested
+from opendbc.car.hyundai.ccnc_ev_stock import replacement_requested, qualified as qualified_ccnc_ev_stock
+from opendbc.car.hyundai.angle_tx_recovery import AngleTxRecovery
 from opendbc.car.hyundai.values import is_blended, HyundaiFlags, HyundaiSafetyFlags, Buttons, CarControllerParams, CAR
 from opendbc.car.interfaces import CarControllerBase
 from opendbc.car.vehicle_model import VehicleModel
@@ -110,6 +111,9 @@ class CarController(CarControllerBase):
     self.angle_vm = VehicleModel(CP) if CP.flags & HyundaiFlags.CANFD_ANGLE_STEERING else None
     self.apply_angle_last = 0.0
     self.angle_gain_last = 0.0
+    self.angle_tx_recovery = (AngleTxRecovery(0x110, self.CAN.ACAN)
+      if CP.carFingerprint in (CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9) and
+      (qualified_angle_aol(CP, marked_only=True) or qualified_ccnc_ev_stock(CP)) else None)
 
     self.accel_last = 0
     self.apply_torque_last = 0
@@ -279,6 +283,8 @@ class CarController(CarControllerBase):
       new_actuators.steeringAngleDeg = self.apply_angle_last
     new_actuators.accel = accel
 
+    if self.angle_tx_recovery is not None:
+      self.angle_tx_recovery.record(can_sends, self.frame)
     self.frame += 1
     if _projection is not None:
       _projection.finish(new_actuators.torque)
@@ -469,6 +475,9 @@ class CarController(CarControllerBase):
         angle_active = angle_aol_active
       if ccnc_ev_stock:
         angle_active = ccnc_ev_replacement
+      if self.angle_tx_recovery is not None and self.angle_tx_recovery.take_rejection(self.frame):
+        self.apply_angle_last = CS.out.steeringAngleDeg
+        self.angle_gain_last = 0.0
       desired_angle = float(np.clip(CC.actuators.steeringAngleDeg, -self.params.ANGLE_LIMITS.STEER_ANGLE_MAX,
                                     self.params.ANGLE_LIMITS.STEER_ANGLE_MAX))
       safety_speed = max(CS.out.vEgoRaw - 1.0, 1.0)

opendbc_repo/opendbc/car/hyundai/interface.py

@@ -378,6 +378,10 @@ class CarInterface(CarInterfaceBase):
     return ret
 
   def update(self, can_packets):
+    recovery = getattr(getattr(self, 'CC', None), 'angle_tx_recovery', None)
+    if recovery is not None:
+      can_packets = list(can_packets)
+      recovery.observe(can_packets, self.CC.frame)
     if (self.CS.pe_ready is not None or self.CS.gv70_camera_lead is not None or self.CS.ev9_camera_lead is not None or
         self.CS.ioniq6_camera_lead is not None or self.CS.torque_ev_camera_lead is not None or self.CS.forte_lkas_sources is not None):
       # This isolated optional parser must never join required controls health.

opendbc_repo/opendbc/safety/tests/test_hyundai_angle_tx_recovery.py

@@ -0,0 +1,148 @@
+"""Real parser/controller/native rejection recovery and echo provenance guards."""
+import unittest
+
+import pytest
+
+from opendbc.can import CANPacker
+from opendbc.car import structs
+from opendbc.car.hyundai.angle_tx_recovery import AngleTxRecovery
+from opendbc.car.hyundai.interface import CarInterface
+from opendbc.car.hyundai.tests.test_stock_angle_aol import source_profile
+from opendbc.car.hyundai.values import CAR
+from opendbc.safety.tests.libsafety import libsafety_py
+from openpilot.starpilot.aol.tests.test_canfd_location_native_recovery import physical_frames
+
+
+class TestAngleTxEcho(unittest.TestCase):
+  @staticmethod
+  def frame():
+    body = bytearray(32)
+    body[9] = 0x20
+    return 0x110, bytes(body), 0
+
+  def test_only_current_exact_local_rejected_command_requests_one_reset(self):
+    address, data, bus = self.frame()
+    for delta in (1, 2, 3):
+      recovery = AngleTxRecovery(address, bus)
+      recovery.record([(address, data, bus)], 10)
+      recovery.observe([(100, [(address, data, bus + 0xC0)])], 10 + delta)
+      self.assertTrue(recovery.take_rejection(10 + delta))
+      self.assertFalse(recovery.take_rejection(10 + delta))
+      recovery.observe([(101, [(address, data, bus + 0xC0)])], 11 + delta)
+      self.assertFalse(recovery.take_rejection(11 + delta))
+
+  def test_stale_wrong_returned_or_unmatched_echo_never_resets(self):
+    address, data, bus = self.frame()
+    for case in ('returned', 'wrong_bus', 'wrong_address', 'wrong_data', 'short', 'old_frame',
+                 'same_frame', 'unknown', 'inactive', 'old_packet', 'late_consume'):
+      recovery = AngleTxRecovery(address, bus)
+      emitted = bytearray(data)
+      if case == 'inactive':
+        emitted[9] = 0x10
+      if case != 'unknown':
+        recovery.record([(address, bytes(emitted), bus)], 10)
+      echo = (address + (case == 'wrong_address'), bytes([data[0] ^ 1]) + data[1:] if case == 'wrong_data' else
+              data[:-1] if case == 'short' else data, bus + (0x80 if case == 'returned' else 0xC1 if case == 'wrong_bus' else 0xC0))
+      frame = 14 if case == 'old_frame' else 10 if case == 'same_frame' else 11
+      if case == 'old_packet':
+        recovery.observe([(100, [])], frame)
+      recovery.observe([(100, [echo])], frame)
+      self.assertFalse(recovery.take_rejection(frame + (case == 'late_consume')), case)
+
+  def test_feedback_uses_original_local_send_deadline_even_when_controller_is_paused(self):
+    for case in ('expired_send', 'expired_consume', 'reversed'):
+      clock = [100_000_000]
+      recovery = AngleTxRecovery(0x110, 0, clock=lambda values=clock: values[0])
+      address, data, bus = self.frame()
+      recovery.record([(address, data, bus)], 10)
+      clock[0] += 30_000_001 if case == 'expired_send' else -1 if case == 'reversed' else 10_000_000
+      recovery.observe([(100, [(address, data, bus + 0xC0)])], 11)
+      if case == 'expired_consume':
+        clock[0] = 130_000_001
+      self.assertFalse(recovery.take_rejection(11), case)
+
+
+@pytest.mark.parametrize('identity', [CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9])
+@pytest.mark.parametrize('old,measured,speed', [(-19.5, -14.7, 5.336), (-4.7, -6.9, 9.212), (-6.5, -3.1, 20.822)])
+def test_actual_stock_angle_controller_recovers_after_local_native_rejection(identity, old, measured, speed):
+  cp = source_profile(identity)
+  cp.alternativeExperience = 32
+  ci = CarInterface(cp)
+  native = libsafety_py.libsafety
+  native.set_alternative_experience(32)
+  assert native.set_safety_hooks(structs.CarParams.SafetyModel.hyundaiCanfd, cp.safetyConfigs[0].safetyParam) == 0
+  native.init_tests()
+  native.set_alternative_experience(32)
+  packer = CANPacker('hyundai_canfd_generated')
+  echo = []
+
+  def receive(tick, lkas=0, button=0):
+    now_ns = 1_000_000_000 + tick * 10_000_000
+    native.set_timer(now_ns // 1000)
+    frames = physical_frames(ci, tick, lkas=lkas, button=button)
+    replaced = []
+    for address, data, bus in frames:
+      name = packer.dbc.addr_to_msg[address].name
+      values = None
+      if name == 'MDPS':
+        values = {'MDPS_PaStrAnglVal': measured, 'MDPS_EstStrAnglVal': measured,
+                  'MDPS_StrTqSnsrVal': 0, 'MDPS_OutTqVal': 0, 'COUNTER': tick % 256}
+      elif name == 'STEERING_SENSORS':
+        values = {'STEERING_ANGLE': measured, 'COUNTER': tick % 256}
+      elif name == 'WHEEL_SPEEDS':
+        values = {**dict.fromkeys(('WHL_SpdFLVal', 'WHL_SpdFRVal', 'WHL_SpdRLVal', 'WHL_SpdRRVal'), speed * 3.6),
+                  'COUNTER': tick % 256}
+      frame = packer.make_can_msg(name, bus, values) if values is not None else (address, data, bus)
+      assert native.safety_rx_hook(libsafety_py.make_CANPacket(frame[0], frame[2], frame[1])), (hex(address), bus)
+      replaced.append(frame)
+    if tick % 10 == 0:
+      native.aol_set_host_request(1)
+      native.set_aol_test_heartbeat(True)
+      native.safety_tick()
+    state = ci.update([(now_ns, replaced + echo)])
+    return now_ns, state
+
+  for tick in range(1, 51):
+    now_ns, state = receive(tick, lkas=int(tick in (40, 41)))
+  assert state.canValid and not state.canTimeout
+  assert native.aol_get_permission_mask() == 1
+  assert abs(state.steeringAngleDeg - measured) < .11
+  ci.CC.apply_angle_last = old
+  command = structs.CarControl(latActive=True)
+  command.actuators.steeringAngleDeg = old - 1 if old < measured else old + 1
+  _, frames = ci.apply(command.as_reader(), now_ns)
+  steering = [frame for frame in frames if frame[0] == 0x110]
+  assert len(steering) == 1
+  first = steering[0]
+  assert not native.safety_tx_hook(libsafety_py.make_CANPacket(first[0], first[2], first[1]))
+  echo[:] = [(first[0], first[1], first[2] + 0xC0)]
+  accepted = False
+  for tick in range(51, 55):
+    now_ns, state = receive(tick)
+    assert state.canValid
+    _, frames = ci.apply(command.as_reader(), now_ns)
+    steering = [frame for frame in frames if frame[0] == 0x110]
+    assert len(steering) == 1
+    frame = steering[0]
+    accepted = bool(native.safety_tx_hook(libsafety_py.make_CANPacket(frame[0], frame[2], frame[1])))
+    if accepted:
+      break
+    echo[:] = [(frame[0], frame[1], frame[2] + 0xC0)]
+  assert accepted, 'Host command must recover from measured angle while every unchanged native limiter still runs'
+  assert native.aol_get_permission_mask() == 1
+
+  # The unchanged native physical CANCEL keeps already-authorized independent
+  # lateral. Configured LKAS OFF belongs to the host, whose zero request
+  # must still veto an active controller command.
+  cancel_tick = tick + 2 - tick % 2
+  now_ns, state = receive(cancel_tick, button=4)
+  assert native.aol_get_permission_mask() == 1
+  _, frames = ci.apply(command.as_reader(), now_ns)
+  frame = next(frame for frame in frames if frame[0] == 0x110)
+  assert native.safety_tx_hook(libsafety_py.make_CANPacket(frame[0], frame[2], frame[1]))
+  now_ns, state = receive(cancel_tick + 2, lkas=1)
+  native.aol_set_host_request(0)
+  assert native.aol_get_permission_mask() == 0
+  _, frames = ci.apply(command.as_reader(), now_ns)
+  frame = next(frame for frame in frames if frame[0] == 0x110)
+  assert not native.safety_tx_hook(libsafety_py.make_CANPacket(frame[0], frame[2], frame[1]))

openpilot/selfdrive/car/card.py

@@ -66,7 +66,7 @@ from openpilot.starpilot.car.nissan.aol import (
   qualified as qualified_nissan_aol, native_observation as nissan_native_observation,
   temporary_restriction as nissan_temporary_restriction,
 )
-from openpilot.starpilot.car.hyundai.aol import aol_safety_config as hyundai_aol_safety_config
+from openpilot.starpilot.car.hyundai.aol import aol_safety_config as hyundai_aol_safety_config, pending_stock_angle_arm
 from openpilot.starpilot.car.hyundai.aol import retained_shared_long_cancel_intent as hkg_cancel_intent_retained
 from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, current_axis, current_native
 from openpilot.starpilot.aol.vehicle import create_intent as create_aol_intent, native_latch_rejected, policy_for as aol_policy_for
@@ -607,7 +607,11 @@ class Car:
           cancel_feedback.last_cancel_source_ns = cancel_source_ns
       rivian_brake_retained = rivian_brake_intent_retained(self.CP, CS, self.sm, self.aol_transport_feedback,
         self.aol_card_intent, now_ns=now_ns, producer=self.slc_producer_session)
-      transport_reset = not (hkg_cancel_retained or rivian_brake_retained) and self.aol_transport_feedback.observe(self.sm, self.CP, CS, now_ns,
+      angle_arm_pending = pending_stock_angle_arm(self.CP, CS, self.sm, self.aol_card_intent,
+        now_ns=now_ns, producer=self.slc_producer_session,
+        acknowledged_arm_ns=getattr(self, '_angle_aol_acknowledged_arm_ns', 0))
+      transport_reset = not (hkg_cancel_retained or rivian_brake_retained or angle_arm_pending) and self.aol_transport_feedback.observe(
+        self.sm, self.CP, CS, now_ns,
         latched=self.aol_card_intent.allowed_latch, cancel_source_ns=cancel_source_ns,
         brake_source_ns=brake_source_ns,
         gm_withdrawal_source_ns=gm_withdrawal_source_ns,
@@ -632,7 +636,7 @@ class Car:
           not feedback.retention_started_ns and not feedback.retention_qualified and not feedback.retention_pending and
           not any(e.immediateDisable or e.softDisable or e.userDisable for e in self.sm['onroadEvents']))
         fault_active = transport_reset or self.aol_disarming_fault(CS, event_ns, now_ns,
-          transport_observed_healthy=reuse_transport)
+          transport_observed_healthy=reuse_transport or angle_arm_pending)
       angle_aol, ford_aol, honda_aol, mazda_aol, preap_aol, volkswagen_aol = self.aol_permission_owners()
       nissan_aol = qualified_nissan_aol(self.CP, marked_only=True)
       rivian_aol = qualified_rivian_aol(self.CP, marked_only=True)
@@ -680,7 +684,11 @@ class Car:
         self._angle_aol_pending_since_ns = pending_since
         if lost:
           fault_active = True
-      rejected = native_reset or native_latch_rejected(self.CP, native, state=CS if angle_panda_ready else None)
+      if (angle_aol and native is not None and native.requestedLateral and native.lateralAllowed and
+          0 < self.aol_card_intent.physical_arm_ns <= native.observedMonoTime):
+        self._angle_aol_acknowledged_arm_ns = self.aol_card_intent.physical_arm_ns
+      rejected = native_reset or (not angle_arm_pending and
+        native_latch_rejected(self.CP, native, state=CS if angle_panda_ready else None))
       rejection_ns = now_ns if native_reset else int(native.observedMonoTime) if rejected else 0
       observe_physical = getattr(self.aol_card_intent, 'observe_physical_samples', None)
       if observe_physical is not None:

openpilot/starpilot/aol/intent.py

@@ -148,6 +148,7 @@ class AolCardIntent:
     self._lkas_held = False
     self._fault_inhibit = False
     self._last_latch_edge_ns = 0
+    self.physical_arm_ns = 0
     self._native_rejection_ns = 0
     self._engagement_source_ns = 0
     self._engagement_consumed = True
@@ -344,6 +345,8 @@ class AolCardIntent:
           self._record_latch_clear('physical_toggle')
         self.allowed_latch = not lateral_engaged
         self.pause_lateral = lateral_engaged
+        if not lateral_engaged:
+          self.physical_arm_ns = now_ns
 
     for event in CS.buttonEvents:
       button = int(event.type.raw)

openpilot/starpilot/aol/tests/test_stock_angle_physical_recovery.py

@@ -0,0 +1,193 @@
+"""Actual host callers; native observations are explicit synthetic test inputs.
+
+Qualification still requires the physical RX/C++ heartbeat/native TX join.
+"""
+from contextlib import ExitStack
+
+import pytest
+
+from opendbc.car import structs
+from opendbc.car.hyundai.canfd_angle_aol import MODELS
+from opendbc.car.hyundai.values import CAR
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.intent import read_settings
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+
+
+@pytest.mark.parametrize('identity', sorted(MODELS))
+def test_stock_angle_physical_recovery_retains_independent_intent(identity):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    saved = Params()
+    saved.put('LKASButtonControl', 9, block=True)
+    saved.put('MainCruiseButtonControl', 0, block=True)
+    for key in ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl'):
+      saved.put(key, 0, block=True)
+    sd, _, step, card, receive = case._ioniq_availability_callers(
+      full_op=False, initially_armed=False, stock_angle=identity)
+    sd.aol_settings = read_settings(saved)
+    assert card.aol_card_intent.settings == sd.aol_settings
+    original_settings = {key: saved.get(key) for key in
+                         ('AlwaysOnLateral', 'LKASButtonControl', 'MainCruiseButtonControl', 'DisengageOnAccelerator')}
+    state = fixture.car_state()
+    state.vEgoRaw = state.vEgo
+    state.cruiseState.enabled = True
+    # A stock SET may request lateral before the native physical latch exists.
+    for tick in range(2, 32):
+      axis, _, command = step(tick, observed_state=state, allowed=False,
+                              native_request_lat=True, panda_controls_allowed=True)
+      assert sd.enabled and not command.latActive
+      receive(10_000_000_000 + tick * 10_000_000, state=state,
+              axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+    # Physical LKAS arrives before the next native permission publication.
+    state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+    for tick in (32, 33):
+      axis, _, command = step(tick, observed_state=state, allowed=False,
+                              native_request_lat=True, panda_controls_allowed=True)
+      receive(10_000_000_000 + tick * 10_000_000, state=state,
+              axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+      assert not command.latActive
+      assert card.aol_card_intent.allowed_latch
+      state.buttonEvents = []
+    for tick in (34, 35):
+      axis, _, command = step(tick, observed_state=state, allowed=True,
+                              native_request_lat=True, panda_controls_allowed=True)
+      receive(10_000_000_000 + tick * 10_000_000, state=state,
+              axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+      assert card.aol_card_intent.allowed_latch
+      assert axis.lateralActive and command.latActive
+    state.cruiseState.enabled = False
+    state.buttonEvents = [structs.CarState.ButtonEvent(type='cancel', pressed=True)]
+    axis, _, command = step(36, observed_state=state, allowed=True,
+                            native_request_lat=True, panda_controls_allowed=False)
+    receive(10_360_000_000, state=state, axis=sd.pm.messages['aolAxisState'],
+            event=sd.pm.messages['onroadEvents'])
+    assert not sd.enabled and not command.longActive
+    assert card.aol_card_intent.allowed_latch and axis.lateralActive and command.latActive
+    assert {key: saved.get(key) for key in original_settings} == original_settings
+
+
+@pytest.mark.parametrize('identity', [CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9])
+@pytest.mark.parametrize('veto', ['deadline', 'held', 'can', 'eps', 'native_invalid', 'panda_rx', 'relay', 'explicit_off', 'ack_then_denied'])
+def test_stock_angle_pending_arm_never_renews_or_masks_restrictions(identity, veto):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    saved = Params()
+    saved.put('LKASButtonControl', 9, block=True)
+    saved.put('MainCruiseButtonControl', 0, block=True)
+    sd, _, step, card, receive = case._ioniq_availability_callers(
+      full_op=False, initially_armed=False, stock_angle=identity)
+    sd.aol_settings = read_settings(saved)
+    state = fixture.car_state()
+    state.vEgoRaw = state.vEgo
+    state.cruiseState.enabled = True
+
+    def cycle(tick, **kwargs):
+      axis, _, command = step(tick, observed_state=state, allowed=False,
+                              native_request_lat=True, panda_controls_allowed=True, **kwargs)
+      receive(10_000_000_000 + tick * 10_000_000, state=state,
+              axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+      return command
+
+    for tick in range(2, 32):
+      cycle(tick)
+    state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+    cycle(32)
+    assert card.aol_card_intent.allowed_latch
+    arm = card.aol_card_intent.physical_arm_ns
+    state.buttonEvents = []
+    if veto in ('deadline', 'held'):
+      for tick in range(33, 53):
+        if veto == 'held':
+          state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+        command = cycle(tick)
+        assert not command.latActive
+        assert card.aol_card_intent.physical_arm_ns == arm
+        assert card.aol_card_intent.allowed_latch == (tick < 52)
+      state.buttonEvents = []
+      axis, _, _ = step(53, observed_state=state, allowed=True, native_request_lat=True, panda_controls_allowed=True)
+      receive(10_530_000_000, state=state, axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+      assert not card.aol_card_intent.allowed_latch
+      return
+    if veto == 'explicit_off':
+      state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=False)]
+      cycle(33)
+      state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)]
+      cycle(34)
+      assert not card.aol_card_intent.allowed_latch
+      return
+    if veto == 'ack_then_denied':
+      axis, _, command = step(33, observed_state=state, allowed=True, native_request_lat=True, panda_controls_allowed=True)
+      receive(10_330_000_000, state=state, axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+      assert command.latActive and card.aol_card_intent.allowed_latch
+      cycle(34)
+      assert not card.aol_card_intent.allowed_latch
+      return
+    if veto == 'can':
+      state.canValid = False
+    elif veto == 'eps':
+      state.steerFaultPermanent = True
+    command = cycle(33, native_valid=veto != 'native_invalid', panda_fault=veto == 'panda_rx', relay_fault=veto == 'relay')
+    assert not command.latActive
+    assert not card.aol_card_intent.allowed_latch
+
+
+@pytest.mark.parametrize('veto', ['no_physical', 'already_acknowledged', 'paused', 'disabled', 'producer', 'source_zero',
+                                'source_future', 'source_age', 'axis_expired', 'axis_lease', 'event_future', 'native_expired', 'profile'])
+def test_pending_stock_angle_arm_requires_original_current_proofs(veto):
+  from dataclasses import replace
+  from openpilot.starpilot.car.hyundai.aol import pending_stock_angle_arm
+
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    saved = Params()
+    saved.put('LKASButtonControl', 9, block=True)
+    saved.put('MainCruiseButtonControl', 0, block=True)
+    sd, _, step, card, receive = case._ioniq_availability_callers(
+      full_op=False, initially_armed=False, stock_angle=CAR.HYUNDAI_IONIQ_5_PE)
+    sd.aol_settings = read_settings(saved)
+    state = fixture.car_state()
+    state.vEgoRaw = state.vEgo
+    state.cruiseState.enabled = True
+    for tick in range(2, 33):
+      state.buttonEvents = [structs.CarState.ButtonEvent(type='lkas', pressed=True)] if tick == 32 else []
+      axis, _, _ = step(tick, observed_state=state, allowed=False, native_request_lat=True, panda_controls_allowed=True)
+      receive(10_000_000_000 + tick * 10_000_000, state=state,
+              axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+    owner = card.aol_card_intent
+    now = 10_320_000_001
+    producer = card.slc_producer_session
+    cp = card.CP.as_reader().as_builder()
+    acknowledged = 0
+    assert pending_stock_angle_arm(cp, state, card.sm, owner, now_ns=now, producer=producer)
+    if veto == 'no_physical':
+      owner.physical_arm_ns = 0
+    elif veto == 'already_acknowledged':
+      acknowledged = owner.physical_arm_ns
+    elif veto == 'paused':
+      owner.pause_lateral = True
+    elif veto == 'disabled':
+      owner.settings = replace(owner.settings, enabled=False)
+    elif veto == 'producer':
+      producer = 'different-producer'
+    elif veto in ('source_zero', 'source_future', 'source_age', 'axis_lease'):
+      axis = card.sm['aolAxisState'].as_builder()
+      if veto == 'axis_lease':
+        axis.validUntilMonoTime = axis.observedMonoTime + 30_000_001
+      else:
+        axis.sourceCarStateMonoTime = 0 if veto == 'source_zero' else now - 30_000_001 if veto == 'source_age' else now + 1
+      card.sm.data['aolAxisState'] = axis.as_reader()
+    elif veto == 'axis_expired':
+      card.sm.logMonoTime['aolAxisState'] = now - 30_000_001
+    elif veto == 'event_future':
+      card.sm.logMonoTime['onroadEvents'] = now + 1
+    elif veto == 'native_expired':
+      card.sm.logMonoTime['aolSafetyWire'] = now - 200_000_001
+    elif veto == 'profile':
+      cp.safetyConfigs[0].safetyParam ^= 4
+    assert not pending_stock_angle_arm(cp, state, card.sm, owner, now_ns=now, producer=producer,
+                                      acknowledged_arm_ns=acknowledged)

openpilot/starpilot/car/hyundai/aol.py

@@ -458,3 +458,38 @@ def expected_stopped_lateral_denial(CP, CS, native, *, sm, axis_session_id, now_
                                    allow_restricted_lateral=True) == native)
   except (AttributeError, KeyError, IndexError, TypeError, ValueError, OverflowError):
     return False
+
+
+def pending_stock_angle_arm(CP, CS, sm, owner, *, now_ns, producer, acknowledged_arm_ns=0):
+  """Keep a fresh physical arm pending its original native acknowledgement."""
+  from opendbc.car.hyundai.canfd_angle_aol import ANGLE_ACK_TIMEOUT_NS
+  from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, current_axis
+  from openpilot.starpilot.aol.transport_pause import _configured_transport_native
+
+  try:
+    arm = owner.physical_arm_ns
+    if (not qualified_angle_aol(CP, marked_only=True) or CP.openpilotLongitudinalControl or not CP.pcmCruise or
+        not owner.allowed_latch or not owner.settings.enabled or owner.pause_lateral or
+        arm != owner._last_latch_edge_ns or arm == acknowledged_arm_ns or not 0 < arm <= now_ns < arm + ANGLE_ACK_TIMEOUT_NS or
+        not CS.canValid or CS.canTimeout or CS.steerFaultTemporary or CS.steerFaultPermanent or CS.accFaulted or
+        CS.gearShifter != car.CarState.GearShifter.drive or temporary_restriction(CP, CS)):
+      return False
+    axis = current_axis(sm, now_ns=now_ns)
+    if (axis is None or str(axis.faultReason) != 'none' or axis.faultSessionId != axis.sessionId or
+        not axis.desiredLateral or axis.lateralActive or axis.desiredLongitudinal or axis.longitudinalActive or
+        axis.observedMonoTime != sm.logMonoTime['aolAxisState'] or
+        not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
+        now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
+        axis.validUntilMonoTime - axis.observedMonoTime > AXIS_MAX_AGE_NS or
+        axis.intentProducerSessionId != producer or not producer):
+      return False
+    event_ns = int(sm.logMonoTime['onroadEvents'])
+    if (not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
+        not 0 < event_ns <= now_ns or now_ns - event_ns > 1_500_000_000 or
+        any(event.immediateDisable or event.softDisable or event.userDisable for event in sm['onroadEvents'])):
+      return False
+    native = _configured_transport_native(sm, CP, axis.sessionId, now_ns, allow_restricted_lateral=True)
+    return bool(native is not None and native.pandaInventory and not native.lateralAllowed and
+                not native.requestedLongitudinal and not native.longitudinalAllowed)
+  except (AttributeError, KeyError, IndexError, TypeError, ValueError, OverflowError):
+    return False

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "9ec33c166d80f04fccdaee9a917679f69742ad6af0480f84bcb3cb451673121e", "schema_version": 1}
+{"protected_inventory_sha256": "1e2e9fd8080aee9393642923cca4d93d651f937ec0725f6cb08e313ad1e42fa8", "schema_version": 1}

tools/ci/run_vehicle_tests.py

@@ -24,6 +24,8 @@ from tools.ci import lifecycle_shards, vehicle_pytest
 # Representative production callers plus their source/native refusal cases.
 # The emitted selected_test_ids record exact parametrized configurations.
 AOL_LIFECYCLE_TARGETS = (
+  'opendbc_repo/opendbc/safety/tests/test_hyundai_angle_tx_recovery.py',
+  'openpilot/starpilot/aol/tests/test_stock_angle_physical_recovery.py',
   'openpilot/starpilot/tests/test_ford_configuration_parity.py',
   'openpilot/starpilot/tests/test_ford_configuration_admission.py',
   'openpilot/starpilot/tests/test_ford_release_long_preferences.py',

tools/ci/tests/test_vehicle_ci.py

@@ -17,6 +17,12 @@ from tools.ci.record_firmware import compile_commands
 
 class TestVehicleCollection(unittest.TestCase):
 
+  def test_stock_angle_recovery_callers_are_required_by_vehicle_ci(self):
+    for path in ('opendbc_repo/opendbc/safety/tests/test_hyundai_angle_tx_recovery.py',
+                 'openpilot/starpilot/aol/tests/test_stock_angle_physical_recovery.py'):
+      self.assertIn(path, targets_for('aol-lifecycle'))
+    self.assertIn('opendbc_repo/opendbc/safety/tests/test_hyundai_angle_tx_recovery.py', targets_for('safety-debug'))
+
   def test_hkg_mode_parity_callers_are_required(self):
     path = 'opendbc_repo/opendbc/safety/tests/test_hkg_mode_parity.py'
     self.assertIn(path, targets_for('aol-lifecycle'))