Fix Bolt pedal startup, resume, and AOL override recovery

opendbc_repo/opendbc/safety/modes/gm.h

@@ -236,7 +236,9 @@ static bool gm_cruise_status_owner_ready(void) {
     ready = gm_cruise_status_main && gm_cruise_status_drive && gm_cruise_status_eps &&
             gm_pedal_sensor_current();
   }
-  return ready && !brake_pressed && !gas_pressed && !regen_braking;
+  // Bolt's 0x3D1 replacement is dashboard status only. Accelerator override
+  // still withdraws pedal/paddle actuation in their independent TX gates.
+  return ready && !brake_pressed && (!gas_pressed || gm_cruise_status_bolt) && !regen_braking;
 }
 
 static void gm_rx_hook(const CANPacket_t *msg) {

opendbc_repo/opendbc/safety/tests/test_gm_bolt_pedal.py

@@ -1008,6 +1008,29 @@ class TestGmBoltPedalSafety(unittest.TestCase):
             self.assertEqual(self.safety.safety_fwd_hook(0, 0x3D1), 2)
           self.assertFalse(self.safety.safety_tx_hook(self.packet((0x3D1, status[1], 2))))
 
+  def test_accelerator_override_keeps_only_dashboard_replacement(self):
+    from opendbc.car.gm import gmcan
+    for word in (0xBD, 0x9D, 0x19D, 0xE700, 0xE701, 0xE702):
+      with self.subTest(word=word):
+        self.safety.set_safety_hooks(CarParams.SafetyModel.gm, word)
+        self.safety.init_tests()
+        self.safety.reset_recorded_can()
+        self.status_sources(1_000_000)
+        raw = self.stock('ECMCruiseControl', {})
+        self.safety.safety_rx_hook(raw)
+        self.assertTrue(self.safety.safety_rx_hook(self.sensor(1, gas=30.)))
+        self.assertTrue(self.safety.get_gas_pressed_prev())
+        status = gmcan.create_ecm_cruise_control_command(self.packer, 0, True, 80.)
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(status)))
+        self.assertEqual(self.safety.safety_fwd_hook(0, 0x3D1), -1)
+        self.safety.safety_rx_hook(raw)
+        self.assertEqual(self.recorded(), [(0x3D1, 0, status[1])])
+        self.safety.set_controls_allowed(True)
+        self.assertFalse(self.safety.safety_tx_hook(self.packet(create_pedal_command(self.packer, .2, 0))))
+        self.safety.safety_tx_hook(self.packet(create_bolt_regen_paddle(self.packer, True)))
+        self.safety.safety_rx_hook(self.stock("EBCMRegenPaddle", {}))
+        self.assertEqual(self.recorded(), [(0x3D1, 0, status[1])])
+
   def test_status_physical_jitter_does_not_halve_output_cadence(self):
     from opendbc.car.gm import gmcan
     for word in (0xBD, 0x9D, 0x19D, 0xE700, 0xE701, 0xE702):

openpilot/selfdrive/car/cruise.py

@@ -85,7 +85,8 @@ class VCruiseHelper:
     if not self.CP.pcmCruise:
       for button in CS.buttonEvents:
         self.pending_resume_ns = 0
-        if (not enabled and CS.cruiseState.available and not button.pressed and
+        resume_edge = not button.pressed or self.CP.brand == 'gm' and CS.buttonEnable
+        if (not enabled and CS.cruiseState.available and resume_edge and
             button.type in (ButtonType.accelCruise, ButtonType.resumeCruise) and self.v_cruise_initialized):
           self.pending_resume_ns = now_ns
 

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -1208,6 +1208,10 @@ class SelfdriveD:
     previous = self.aol_axis_decision
     was_enabled = self.enabled
     self.aol_axis_decision = AxisDecision()
+    if healthy_expiry and not was_enabled and self.cold_gm_main_readiness(CS, intent, now_ns):
+      self.aol_fault_reason = 'availabilityPause'
+      self.aol_publication_retained_intent = intent
+      return
     pause = getattr(self, 'aol_availability_pause', None)
     if (healthy_expiry and pause is not None and intent is not None and
         getattr(self, 'aol_fault_reason', 'none') in ('none', 'availabilityPause') and
@@ -1359,6 +1363,8 @@ class SelfdriveD:
     if source_clock_scoped:
       self._expire_axis_source(CS, now_ns)
       self._expire_axis_continuity(CS, now_ns)
+    if self.aol_axis_decision.lateral_active or self.aol_axis_decision.longitudinal_active:
+      self.aol_ever_active = True
     reason = getattr(self, 'aol_fault_reason', 'none')
     event_now_ns = now_ns
     ce_send = None
@@ -1600,6 +1606,18 @@ class SelfdriveD:
         sample.vEgo < (self.aol_settings.pause_brake_mps if self.aol_settings is not None else 0.0)
         for sample, _ in getattr(self, 'car_state_samples', ())))
 
+  def cold_gm_main_readiness(self, CS, intent, now_ns):
+    from openpilot.starpilot.car.gm.aol import cold_main_readiness
+    from openpilot.starpilot.aol.transport_pause import _healthy_transport_native
+    if (self.CP.brand != 'gm' or self.enabled or self.active or getattr(self, 'aol_ever_active', False) or self.aol_authority_lost or
+        not self.events.contains(ET.SOFT_DISABLE) or self.events.contains(ET.IMMEDIATE_DISABLE) or
+        intent is None or not intent.settingsQualified or not intent.allowedLatch or not intent.lateralArmed or
+        not 0 < intent.carStateLogMonoTime <= intent.observedMonoTime <= now_ns <= intent.validUntilMonoTime or
+        now_ns - intent.carStateLogMonoTime > SAFETY_MAX_AGE_NS):
+      return False
+    native = _healthy_transport_native(self.sm, self.CP, CS, self.aol_session_id, now_ns)
+    return cold_main_readiness(self.CP, CS, native, self.events.to_msg())
+
   def ordinary_native_long_withdrawal(self, CS, native, previous, requested, intent, *, source_current, now_ns):
     """Keep the existing ordinary Panda mismatch owner for a LONG-only drop.
 
@@ -1870,7 +1888,8 @@ class SelfdriveD:
           source_ns=self.aol_car_state_log_ns, now_ns=now_ns, axis_session_id=self.aol_session_id,
           pending_lateral_request=not previous.lateral_active and not self.aol_authority_lost))
         for event in self.events.to_msg())
-      self.aol_fault_reason = ('availabilityPause' if ordinary_availability else
+      cold_main_wait = self.cold_gm_main_readiness(CS, intent, now_ns)
+      self.aol_fault_reason = ('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')
@@ -1909,6 +1928,8 @@ class SelfdriveD:
         # Preserve the exact current SOFT event and normal countdown. A live
         # standard-axis proof describes healthy current authority, not critical.
         self.aol_fault_reason = 'none'
+      if cold_main_wait:
+        self.aol_axis_decision = AxisDecision()
       if self.aol_ordinary_source_pause:
         self.aol_axis_decision = self.ordinary_source_withdrawal()
       elif availability:

openpilot/starpilot/aol/tests/test_gm_held_native.py

@@ -106,6 +106,23 @@ def baseline():
   return card, feedback, native
 
 
+@pytest.mark.parametrize('health_allowed', [False, True])
+def test_native_long_withdrawal_before_parser_keeps_existing_card_intent(health_allowed):
+  with OpenpilotPrefix():
+    card, feedback, prior = baseline()
+    now = NOW + 10_000_000
+    cs = parse(card, now, counter=1)
+    native = replace(prior, observedMonoTime=now, validUntilMonoTime=now + 200_000_000,
+      longitudinalAllowed=False, pandaInventory=(replace(prior.pandaInventory[0], controlsAllowed=health_allowed),))
+    sm, _ = peers(card.CP, card.slc_producer_session, now, controls=health_allowed, native=native, sequence=2)
+    # The host's last full-axis result precedes the native paddle/brake edge.
+    reset = feedback.observe(sm, card.CP, cs, now, latched=True,
+                             intent_producer_session=card.slc_producer_session)
+    card.aol_card_intent.update(cs, fault_active=reset, now_ns=now)
+    assert not reset and not feedback.failed and card.aol_card_intent.allowed_latch
+    assert not native.longitudinalAllowed
+
+
 @pytest.mark.parametrize('gesture', ['brake', 'paddle', 'cancel'])
 def test_real_physical_withdrawal_retains_latch_with_held_long_wire(gesture):
   with OpenpilotPrefix():

openpilot/starpilot/aol/tests/test_gm_route18e_recovery.py

@@ -0,0 +1,169 @@
+"""Route 18e regressions through actual parser, Card, SelfdriveD and Controls."""
+from contextlib import ExitStack
+from unittest.mock import patch
+import time
+
+import pytest
+from opendbc.can import CANPacker
+from opendbc.car import Bus
+from opendbc.car.structs import car
+from opendbc.car.gm.bolt_cc import button_bytes
+from opendbc.car.gm.tests.test_bolt_pedal import params
+from opendbc.car.gm.values import CAR, CruiseButtons, DBC
+from openpilot.cereal import messaging
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.car.cruise import VCruiseHelper
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests import test_gm as gm_fixtures
+from openpilot.starpilot.aol.tests.test_gm_held_native import NOW, parse as physical_packets
+from openpilot.starpilot.aol.tests.test_seatbelt_cold_start import deliver_card as deliver
+from openpilot.selfdrive.selfdrived.events import EventName
+
+
+def deliver_card(card, sd, controls, cs, tick):
+  # Both clocks are explicit fixture inputs, including on macOS hosts.
+  with patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True), \
+       patch.object(time, 'clock_gettime_ns', return_value=NOW + tick * 10_000_000):
+    deliver(card, sd, controls, cs, tick)
+
+
+def parse(card, now, counter=0, *, packets_only=False):
+  rows = physical_packets(card, now, counter=counter, packets_only=True)
+  packer = CANPacker(DBC[card.CP.carFingerprint][Bus.pt])
+  rows.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCruiseState': 0}))
+  if packets_only:
+    return rows
+  if now == NOW:
+    card.CI.update([(now - 1_000_000, rows)])
+  cs = card.CI.update([(now, rows)])
+  assert cs.canValid and not cs.canTimeout
+  return cs
+
+
+@pytest.mark.parametrize('event', [EventName.commIssue, EventName.commIssueAvgFreq, EventName.locationdTemporaryError,
+                                  EventName.controlsMismatch, EventName.processNotRunning])
+def test_cold_main_waits_for_host_readiness_but_keeps_real_faults(event):
+  with OpenpilotPrefix(), ExitStack() as cleanup:
+    saved = Params()
+    saved.put_bool('AlwaysOnLateral', True, block=True)
+    card = gm_fixtures.TestGmAol.card(params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True), saved)
+    cleanup.callback(card.vehicle_startup.close)
+    case = fixture.IoniqAvailabilityTests(methodName='runTest')
+    cleanup.callback(case.doCleanups)
+    sd, controls, drive = case._authority_callers(False, cp=card.CP.as_reader().as_builder(), genuine_events=True)
+    card.slc_producer_session = 'authority-card'
+    cs = parse(card, NOW)
+    card.aol_card_intent.update(cs, fault_active=False, now_ns=NOW)
+    original_events = sd.update_events
+    axis, state, command = drive(0, observed_state=cs, native_request_lat=False, native_request_long=False,
+                                allowed=False, long_allowed=False, critical_event=event)
+    temporary = event in (EventName.commIssue, EventName.commIssueAvgFreq, EventName.locationdTemporaryError)
+    assert str(axis.faultReason) == ('availabilityPause' if temporary else 'critical')
+    assert not command.latActive and not command.longActive and not state.enabled
+    deliver_card(card, sd, controls, cs, 0)
+    assert card.aol_card_intent.allowed_latch == temporary
+    if temporary:
+      # The source's original 30ms output deadline still expires during startup.
+      sd._expire_axis_source(cs, NOW + 31_000_000)
+      assert sd.aol_car_state_log_ns == 0 and str(sd.aol_fault_reason) == 'availabilityPause'
+      assert EventName.controlsMismatch not in sd.events.names
+      sd.update_events = original_events
+      axis, _, command = drive(4, native_request_long=False, native_request_lat=False, allowed=False)
+      assert axis.desiredLateral and not axis.lateralActive and not command.latActive and not command.longActive
+      deliver_card(card, sd, controls, parse(card, NOW + 40_000_000, counter=1), 4)
+      assert card.aol_card_intent.allowed_latch
+      axis, _, command = drive(5, native_request_long=False)
+      assert axis.lateralActive and command.latActive and not command.longActive
+      deliver_card(card, sd, controls, parse(card, NOW + 50_000_000, counter=2), 5)
+      assert card.aol_card_intent.allowed_latch
+
+
+@pytest.mark.parametrize('candidate', [CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021,
+                                      CAR.CHEVROLET_BOLT_CC_2022_2023, CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL])
+def test_gm_resume_press_survives_neutral_samples_before_control_ack(candidate):
+  with OpenpilotPrefix(), ExitStack() as cleanup:
+    card = gm_fixtures.TestGmAol.card(params(candidate, pedal=True, camera=True), Params())
+    cleanup.callback(card.vehicle_startup.close)
+    card.v_cruise_helper = VCruiseHelper(card.CP)
+    card.v_cruise_helper.v_cruise_kph = 92.8
+    card.CC_prev = car.CarControl(enabled=False)
+    card.CS_prev = parse(card, NOW)
+    for tick in range(3):
+      now = NOW + (tick + 1) * 10_000_000
+      packets = parse(card, now, counter=tick % 4, packets_only=True)
+      packets = [packet for packet in packets if packet[0] != 0x1E1]
+      packets.append((0x1E1, button_bytes(CruiseButtons.RES_ACCEL, tick % 4), 0))
+      cs = card.CI.update([(now, packets)])
+      assert cs.buttonEnable == (tick == 0)
+      control = messaging.new_message('carControl', valid=True, logMonoTime=now)
+      control.carControl.enabled = tick == 2
+      card.sm.update_msgs(now / 1e9, [control.as_reader()])
+      with patch('openpilot.selfdrive.car.cruise.monotonic_ns', return_value=now):
+        card.update_cruise_speed(cs)
+      card.CS_prev = cs
+    assert card.v_cruise_helper.v_cruise_kph == 92.8
+
+
+@pytest.mark.parametrize('bad', ['native_denied', 'expired_native', 'rx_fault', 'producer', 'epoch',
+                                'source_zero', 'expired_wait', 'immediate', 'process', 'main_off', 'door'])
+def test_cold_intent_retention_never_bypasses_provenance_or_faults(bad):
+  from dataclasses import replace
+  from openpilot.starpilot.aol.tests.test_gm_held_native import baseline, peers, Peers
+  from openpilot.starpilot.aol.transport_pause import TransportPauseFeedback
+  from openpilot.starpilot.aol.wire import encode_safety
+  from openpilot.selfdrive.selfdrived.events import Events
+  with OpenpilotPrefix(), ExitStack() as cleanup:
+    card, _, prior = baseline()
+    cleanup.callback(card.vehicle_startup.close)
+    feedback = TransportPauseFeedback(card.CP)
+
+    def sample(now, *, recovery=False, mutate=False):
+      cs = parse(card, now, counter=1 if recovery else 0)
+      native = replace(prior, observedMonoTime=now, validUntilMonoTime=now + 200_000_000,
+        requestedLateral=False, lateralAllowed=False, requestedLongitudinal=False, longitudinalAllowed=False,
+        pandaInventory=(replace(prior.pandaInventory[0], controlsAllowed=False),))
+      incoming, _ = peers(card.CP, card.slc_producer_session, now, controls=False, native=native,
+                           sequence=2 if recovery else 1, long_active=False)
+      axis = incoming.messages['aolAxisState'].as_builder()
+      a = axis.aolAxisState
+      a.lateralActive = a.longitudinalActive = a.desiredLongitudinal = False
+      a.desiredLateral = recovery
+      a.nativeAcknowledged = recovery
+      a.faultReason = 'none' if recovery else 'availabilityPause'
+      a.faultEventMonoTime = now
+      events = Events()
+      if not recovery:
+        events.add(EventName.commIssue)
+      if mutate:
+        if bad == 'native_denied':
+          native = replace(native, requestedLateral=True)
+        elif bad == 'expired_native':
+          native = replace(native, observedMonoTime=now - 200_000_001, validUntilMonoTime=now - 1)
+        elif bad == 'rx_fault':
+          native = replace(native, pandaInventory=(replace(native.pandaInventory[0], safetyRxChecksInvalid=True),))
+        elif bad == 'producer':
+          a.intentProducerSessionId = 'foreign'
+        elif bad == 'epoch':
+          a.intentIntegrityEpoch += 1
+        elif bad == 'source_zero':
+          a.sourceCarStateMonoTime = 0
+        elif bad == 'immediate':
+          events.add(EventName.controlsMismatch)
+        elif bad == 'process':
+          events.add(EventName.processNotRunning)
+        elif bad == 'main_off':
+          cs.cruiseState.available = False
+        elif bad == 'door':
+          cs.doorOpen = True
+      event = messaging.new_message('onroadEvents', len(events), valid=True, logMonoTime=now)
+      event.onroadEvents = events.to_msg()
+      wire = messaging.new_message('aolSafetyWire', 0, valid=True, logMonoTime=native.observedMonoTime)
+      wire.aolSafetyWire = encode_safety(native)
+      sm = Peers([axis.as_reader(), event.as_reader(), wire.as_reader(), incoming.messages['pandaStates']])
+      return feedback.observe(sm, card.CP, cs, now, latched=True, intent_producer_session=card.slc_producer_session)
+
+    assert not sample(NOW)
+    assert feedback.cold_main_wait_ns == NOW
+    assert sample(NOW + (210_000_000 if bad == 'expired_wait' else 10_000_000), recovery=True, mutate=True)
+    assert feedback.failed

openpilot/starpilot/aol/tests/test_volt_lag_pedal_intent.py

@@ -9,25 +9,26 @@ from opendbc.can import CANPacker
 from opendbc.car import Bus
 from opendbc.car.gm.tests.test_volt_sdgm_control import sdgm_params
 from opendbc.car.gm.bolt_cc import button_bytes
-from opendbc.car.gm.values import DBC, CruiseButtons
+from opendbc.car.gm.values import DBC, CruiseButtons, CAR, is_bolt_pedal_profile
+from opendbc.car.gm.tests.test_bolt_pedal import params as bolt_params
 from openpilot.cereal import messaging
 from openpilot.common.params import Params
 from openpilot.common.prefix import OpenpilotPrefix
 from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
 from openpilot.starpilot.aol.tests import test_gm as gm_fixture
 from openpilot.starpilot.aol.tests.test_gm_startup_main import parsed_packets
-from openpilot.starpilot.aol.tests.test_seatbelt_cold_start import deliver_card
+from openpilot.starpilot.aol.tests.test_gm_route18e_recovery import deliver_card
 from openpilot.selfdrive.selfdrived.events import Events, EventName
 from openpilot.starpilot.aol.wire import decode_safety, encode_safety
 
 BASE = 10_000_000_000
 
 
-def callers(cleanup):
+def callers(cleanup, cp=None):
   params = Params()
   cleanup.callback(params._finalizer)
   params.put_bool('AlwaysOnLateral', True, block=True)
-  card = gm_fixture.TestGmAol.card(sdgm_params(), params)
+  card = gm_fixture.TestGmAol.card(sdgm_params() if cp is None else cp, params)
   cleanup.callback(card.vehicle_startup.close)
   case = fixture.IoniqAvailabilityTests(methodName='runTest')
   cleanup.callback(case.doCleanups)
@@ -37,6 +38,17 @@ def callers(cleanup):
 
   def state(tick, pedal='none', main=True):
     now = BASE + tick * 10_000_000
+    if is_bolt_pedal_profile(card.CP):
+      from openpilot.starpilot.aol.tests.test_gm_held_native import parse
+      packets = parse(card, now, gesture={'regen': 'paddle'}.get(pedal, pedal), counter=tick % 4, packets_only=True)
+      packets = [packet for packet in packets if packet[0] != 0xC9]
+      packets.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCruiseState': 0}))
+      packets.append(packer.make_can_msg('ECMEngineStatus', 0, {'CruiseMainOn': int(main), 'BrakePressed': int(pedal == 'brake')}))
+      if tick == 0:
+        card.CI.update([(now - 1_000_000, packets)])
+      cs = card.CI.update([(now, packets)])
+      assert cs.canValid and not cs.canTimeout
+      return cs
     packets = parsed_packets(card.CP, packer, now, tick, 20., main)
     addresses = {packer.make_can_msg(name, 0, {})[0] for name in ('ECMAcceleratorPos', 'EBCMRegenPaddle')}
     packets = [p for p in packets if p[0] not in addresses]
@@ -68,11 +80,15 @@ def callers(cleanup):
   return card, sd, controls, state, drive
 
 
+@pytest.mark.parametrize('profile', [None, *[(candidate, removed) for candidate in
+  (CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021, CAR.CHEVROLET_BOLT_CC_2022_2023,
+   CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL) for removed in (False, True)]])
 @pytest.mark.parametrize('pedal', ['brake', 'regen'])
 @pytest.mark.parametrize('order', ['together', 'axis_first', 'event_first'])
-def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order):
+def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order, profile):
   with OpenpilotPrefix(), ExitStack() as cleanup:
-    card, sd, controls, parse, drive = callers(cleanup)
+    cp = None if profile is None else bolt_params(profile[0], pedal=True, camera=not profile[1], removed=profile[1])
+    card, sd, controls, parse, drive = callers(cleanup, cp)
     feedback = card.aol_transport_feedback
     baseline = (feedback.lateral_baseline_ns, feedback.lateral_source_ns, feedback.lateral_native_ns,
                 feedback.healthy_feedback_ns, feedback.healthy_native_ns)
@@ -99,9 +115,11 @@ def test_parsed_pedal_lag_keeps_intent_without_renewing_authority(pedal, order):
     assert baseline == (feedback.lateral_baseline_ns, feedback.lateral_source_ns, feedback.lateral_native_ns,
                         feedback.healthy_feedback_ns, feedback.healthy_native_ns)
     cs = parse(3)
-    with mock.patch.object(sd.rk.avg_dt, 'get_average', return_value=.02):
-      axis, _, command = drive(3, cs, allowed=False, native_request_lat=False)
+    axis, _, command = drive(3, cs, allowed=False, native_request_lat=False)
     assert not axis.lateralActive and not command.latActive and not command.longActive
+    assert axis.desiredLateral and str(axis.faultReason) == 'none'
+    deliver_card(card, sd, controls, cs, 3)
+    assert card.aol_card_intent.allowed_latch and not feedback.failed
     cs = parse(4)
     axis, _, command = drive(4, cs)
     assert axis.lateralActive and command.latActive and not command.longActive

openpilot/starpilot/aol/transport_pause.py

@@ -257,6 +257,9 @@ class TransportPauseFeedback:
     self.awaiting = False
     self.failed = False
     self.lateral_baseline_ns = 0
+    self.ever_lateral_active = False
+    self.lag_withdrawal_ns = 0
+    self.cold_main_wait_ns = 0
     self.lateral_only_baseline = False
     self.lateral_source_ns = 0
     self.lateral_inventory = None
@@ -284,14 +287,16 @@ class TransportPauseFeedback:
     method neither grants outputs nor exempts any native or control fault.
     """
     from openpilot.starpilot.nostalgia import physical_cancel
+    from openpilot.starpilot.car.gm.aol import native_accepts_cp
+    gm = native_accepts_cp(cp, int(cp.safetyConfigs[0].safetyModel.raw), int(cp.safetyConfigs[0].safetyParam))
     if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
         not self.session or str(axis.sessionId) != self.session or
         not producer or producer != self.availability_producer or
         str(getattr(axis, 'intentProducerSessionId', '')) != producer or
         int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
         not self.lateral_inventory or not self.lateral_baseline_ns or
-        str(axis.faultReason) != 'none' or str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
-        not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive or
+        str(axis.faultReason) != 'none' or not gm and str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
+        not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive and not gm or
         not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
         now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
         not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
@@ -308,8 +313,8 @@ class TransportPauseFeedback:
       return None
     owners = [index for index, config in enumerate(cp.safetyConfigs)
               if str(config.safetyModel) not in ('silent', 'noOutput')]
-    return native if (len(owners) == 1 and not native.pandaInventory[owners[0]].controlsAllowed and
-                      not sm['pandaStates'][owners[0]].controlsAllowed) else None
+    return native if (len(owners) == 1 and (gm or not native.pandaInventory[owners[0]].controlsAllowed and
+                      not sm['pandaStates'][owners[0]].controlsAllowed)) else None
 
   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."""
@@ -388,7 +393,11 @@ class TransportPauseFeedback:
                     faults[0].softDisable and not faults[0].immediateDisable)
     # Healthy cycles never reach policy/native/provenance work. A critical
     # proof can precede its event, but cannot yet establish a lag exemption.
-    if not isolated and (reason != 'critical' or faults):
+    recovering = bool(cp.brand == 'gm' and not faults and reason == 'none' and
+      0 < self.lag_withdrawal_ns <= now_ns <= self.lag_withdrawal_ns + SAFETY_MAX_AGE_NS and
+      axis.desiredLateral and not axis.lateralActive and axis.nativeAcknowledged and
+      not axis.desiredLongitudinal and not axis.longitudinalActive)
+    if not recovering and not isolated and (reason != 'critical' or faults):
       return False
     previous = self.signature
     joining = bool(self.pending_event and previous is not None and previous[5] == 'critical' and
@@ -420,7 +429,7 @@ class TransportPauseFeedback:
       now_ns - axis.faultEventMonoTime <= AXIS_MAX_AGE_NS and
       (joining or known_lag or not faults and previous is not None and previous[5] == 'none' and
        0 <= now_ns - self.lateral_baseline_ns <= AXIS_MAX_AGE_NS))
-    if not (matched or held_original or pending):
+    if not (matched or held_original or pending or recovering):
       return False
     from openpilot.starpilot.aol.vehicle import policy_for
     if not policy_for(cp).intent_supported:
@@ -433,13 +442,13 @@ class TransportPauseFeedback:
       return False
     users = [event for event in events if event.userDisable]
     if users:
-      from opendbc.car.gm.values import is_volt_sdgm_accepted_envelope
+      from opendbc.car.gm.values import is_volt_sdgm_accepted_envelope, is_bolt_pedal_profile
       from openpilot.starpilot.nostalgia import physical_cancel
-      # On the accepted Volt SDGM owner, an already LAT-only physical brake
+      # On accepted Volt SDGM and Bolt pedal owners, a LAT-only physical brake
       # or regen sample also emits ordinary pedalPressed/USER_DISABLE. It
       # must not turn an otherwise source-qualified temporary lag withdrawal
       # into a permanent driver-intent reset. No axis or lease is retained.
-      volt_pedal = bool(self.lateral_only_baseline and is_volt_sdgm_accepted_envelope(cp) and
+      gm_pedal = bool(self.lateral_only_baseline and (is_volt_sdgm_accepted_envelope(cp) or is_bolt_pedal_profile(cp)) and
         (cs.brakePressed or cs.regenBraking) and cs.cruiseState.available and
         not cs.doorOpen and not cs.seatbeltUnlatched and not cs.steeringDisengage and not physical_cancel(cs, cp) and
         not axis.desiredLongitudinal and not axis.longitudinalActive and
@@ -447,7 +456,7 @@ class TransportPauseFeedback:
         all(event.name == log.OnroadEvent.EventName.pedalPressed and event.noEntry and
             not event.softDisable and not event.immediateDisable for event in users))
       stock_user = stock_lateral_user_event(cp)
-      if not volt_pedal and (not getattr(self, 'lateral_only_baseline', False) or stock_user is None or
+      if not gm_pedal and (not getattr(self, 'lateral_only_baseline', False) or stock_user is None or
           any(event.name != stock_user or event.softDisable or event.immediateDisable for event in users) or
           not stock_lateral_off(cp, cs, native)):
         return False
@@ -457,6 +466,8 @@ class TransportPauseFeedback:
       self.awaiting = True
     elif joining:
       self.pending_event = 0
+    if matched and cp.brand == 'gm':
+      self.lag_withdrawal_ns = event_ns
     return True
 
   def _temporary_steering_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer, signature):
@@ -962,6 +973,10 @@ class TransportPauseFeedback:
                      cp_identity(cp) == self.cp)
       continuity = bool(not self.session or axis.sessionId == self.session and
                         (axis.sequence > self.sequence or axis.sequence == self.sequence and signature == self.signature))
+      if (self.cold_main_wait_ns and current and continuity and self.signature is not None and
+          signature[14:16] != self.signature[14:16]):
+        self.failed = True
+        return True
       if current and continuity and 0 < main_arm_ns <= now_ns <= main_arm_ns + AXIS_MAX_AGE_NS:
         from openpilot.starpilot.car.gm.aol import prior_main_off_intent_current
         if prior_main_off_intent_current(self, sm, cp, cs, axis, now_ns,
@@ -970,17 +985,47 @@ class TransportPauseFeedback:
           # the new request nor renews any command, native or healthy lease.
           self.sequence, self.signature = int(axis.sequence), signature
           return False
+      if (cp.brand == 'gm' and current and continuity and latched and not self.failed and
+          not self.ever_lateral_active and not self.pending_event):
+        from openpilot.starpilot.car.gm.aol import cold_main_readiness
+        native = _healthy_transport_native(sm, cp, cs, str(axis.sessionId), now_ns)
+        event_ns = int(sm.logMonoTime['onroadEvents'])
+        recovering = bool(reason == 'none' and axis.desiredLateral and not axis.desiredLongitudinal and
+          not axis.lateralActive and not axis.longitudinalActive and axis.nativeAcknowledged and
+          0 < self.cold_main_wait_ns <= now_ns <= self.cold_main_wait_ns + SAFETY_MAX_AGE_NS and
+          0 < axis.sourceCarStateMonoTime <= stamp and now_ns - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS)
+        if ((reason == 'availabilityPause' and not any(signature[9:13]) or recovering) and
+            intent_producer_session and signature[14] == intent_producer_session and
+            (self.signature is None or signature[14:16] == self.signature[14:16]) and
+            axis.faultEventMonoTime == event_ns and sm.seen['onroadEvents'] and
+            sm.valid['onroadEvents'] and sm.alive['onroadEvents'] and 0 < event_ns <= stamp and
+            now_ns - event_ns <= AXIS_MAX_AGE_NS and
+            cold_main_readiness(cp, cs, native, sm['onroadEvents'], recovering=recovering)):
+          if not recovering:
+            self.cold_main_wait_ns = event_ns
+          self.session = str(axis.sessionId)
+          self.sequence, self.signature = int(axis.sequence), signature
+          self.temporary_host_qualified = True
+          return False
+      if (self.cold_main_wait_ns and current and continuity and reason == 'none' and
+          axis.desiredLateral and not axis.lateralActive):
+        self.failed = True
+        return True
       if current and continuity and self._angle_stop_intent_current(sm, cp, cs, axis, stamp, now_ns,
           latched=latched, producer=intent_producer_session):
         self.sequence, self.signature = int(axis.sequence), signature
         return False
-      if current and continuity and (reason == 'critical' or
+      if current and continuity and (reason == 'critical' or self.lag_withdrawal_ns and not axis.lateralActive or
           getattr(sm, 'updated', {}).get('onroadEvents', False)) and self._temporary_lag_intent_current(sm, cp, cs, axis, stamp, now_ns,
           latched=latched, producer=intent_producer_session, signature=signature):
         # Received identity only; every original permission/output deadline and
         # healthy baseline timestamp remains unchanged.
         self.sequence, self.signature = int(axis.sequence), signature
         return False
+      if (cp.brand == 'gm' and self.lag_withdrawal_ns and current and continuity and reason == 'none' and
+          axis.desiredLateral and not axis.lateralActive):
+        self.failed = True
+        return True
       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
@@ -1154,6 +1199,9 @@ 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.ever_lateral_active = True
+          self.lag_withdrawal_ns = 0
+          self.cold_main_wait_ns = 0
           self.temporary_eps_event_ns = 0
           self.lateral_native_ns = int(native.observedMonoTime)
           self.lateral_inventory = (native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory))

openpilot/starpilot/car/gm/aol.py

@@ -122,6 +122,24 @@ def create_intent(cp, settings):
   return GmAolCardIntent(settings, cp) if qualified_gm(cp) else None
 
 
+def cold_main_readiness(cp, cs, native, events, *, recovering=False):
+  """A MAIN request may wait for host startup without acquiring either axis."""
+  from openpilot.cereal import log
+  names = (log.OnroadEvent.EventName.commIssue, log.OnroadEvent.EventName.commIssueAvgFreq,
+           log.OnroadEvent.EventName.locationdTemporaryError)
+  faults = [event for event in events if event.softDisable or event.immediateDisable]
+  host_wait = not faults if recovering else bool(faults and
+    all(event.softDisable and not event.immediateDisable and event.name in names for event in faults))
+  return bool(qualified_gm(cp) and aol_enabled(cp) and host_wait and
+    not any(event.userDisable for event in events) and
+    cs.canValid and not cs.canTimeout and cs.cruiseState.available and
+    not cs.steerFaultTemporary and not cs.steerFaultPermanent and not cs.accFaulted and
+    not cs.doorOpen and not cs.seatbeltUnlatched and not cs.steeringDisengage and
+    native is not None and native.pandaInventory and
+    not native.requestedLateral and not native.lateralAllowed and
+    not native.requestedLongitudinal and not native.longitudinalAllowed)
+
+
 def prior_main_off_intent_current(feedback, sm, cp, cs, axis, now_ns, *, latched, main_arm_ns, producer):
   """An older idle response cannot reject a newly armed cruise MAIN request."""
   from opendbc.car.structs import car