fix(slc): prevent confirmation prompts when longitudinal control is disengaged

openpilot/selfdrive/controls/plannerd.py

@@ -440,6 +440,8 @@ def starpilot_main():
   if conditional_host is not None:
     subscribed_services.append('starpilotRadarState')
   optional_slc = ['slcDashboardObservation', 'slcVisionObservation', 'mapdOut', 'starpilotCarState'] if slc_available else []
+  if slc_available and feature_enabled(params, CP, 'aol', os.environ):
+    optional_slc.append('aolAxisState')
   subscribed_services.extend(optional_slc)
   subscribed_services.append('starpilotNavigation')
   optional_conditional = ['starpilotNavigation'] + (['starpilotRadarState'] if conditional_host is not None else []) + optional_slc

openpilot/starpilot/speed_limits/README.md

@@ -54,8 +54,12 @@ The source and arrow share a centered row; all sign contents use the same geomet
 on both screens. Small arrows are baked at double resolution and text is aligned
 to final pixels for clear native rendering.
 Steering-wheel + accepts and − declines. Tapping the sign opens the same
-Accept / Reject choices on either native screen. A pending manual choice expires
-after ten monotonic seconds, including time without active longitudinal control.
+Accept / Reject choices on either native screen. Manual confirmations appear only
+while system longitudinal control is engaged, including a temporary accelerator
+override. Disengaged and AOL lateral-only modes keep displaying limits without
+prompting. Losing longitudinal engagement cancels the pending choice without
+declining it; re-engaging starts a fresh choice if needed. A pending manual choice
+expires after ten monotonic seconds while engaged.
 Decline and timeout leave set speed unchanged. An action at the exact deadline
 can still complete its presented decision; a later action cannot accept it.
 
@@ -83,8 +87,10 @@ This is approximate timing, not a promise to reach the target speed at the sign.
 The first offer latches one boundary encounter. Slowing or stopping does not
 retract it or restart the existing ten-second confirmation deadline. Source loss
 removes authority and hides the pending prompt; fresh recovery of the same
-boundary uses the original deadline and a current presentation. Decline, expiry,
-manual override and completed application survive the boundary becoming current.
+boundary uses the original deadline and a current presentation. Losing longitudinal
+engagement cancels an unanswered choice; re-engaging starts a fresh deadline.
+Decline, expiry, manual override and completed application survive the boundary
+becoming current.
 An unchanged approach limit cannot be offered again solely because the preview
 is lost or disabled. Current-source evidence, command freshness, expected SET,
 driver-button checks and Toyota's native capacity/release checks still apply.

openpilot/starpilot/speed_limits/acceptance.py

@@ -424,6 +424,9 @@ def step(state: State, observation: Observation, authority: Authority, policy: P
     return finish("accepted", target=previous_target())
   confirmation = policy.confirm_higher if previous_speed is None or candidate.speed_mps > previous_speed else (
     policy.confirm_lower if candidate.speed_mps < previous_speed else policy.confirm_lower or policy.confirm_higher)
+  if confirmation and not authority.can_control:
+    state = replace(state, pending=None, timer_running=False)
+    return finish("inactive")
   if not new_limit and old.pending is not None and old.pending.automatic == (not confirmation):
     pending = replace(old.pending, candidate=candidate)
     if old.timer_running:
@@ -452,9 +455,6 @@ def step(state: State, observation: Observation, authority: Authority, policy: P
     elif old.presentation is None or old.presentation.candidate != candidate:
       receipt = replace(receipt, status="different_presentation")
     else:
-      # The driver is always asked, even with only lateral or stock ACC active; the
-      # choice is recorded as accepted history and only commands a target when
-      # this system owns longitudinal control.
       receipt = replace(receipt, consumed=True, status=action.kind.value)
       if action.kind is ActionKind.ACCEPT:
         state, history = _accept(state, candidate, now_ns)

openpilot/starpilot/speed_limits/advance.py

@@ -12,7 +12,7 @@ class Encounter:
   from_mps: float
   speed_mps: float
   way_id: int
-  started_ns: int
+  started_ns: int | None
   crossed: bool = False
   declined: bool = False
   withdrawn: bool = False
@@ -40,10 +40,16 @@ class Advance:
         self.encounter.accepted_ns = None
 
   def propose(self, observations, policy, upcoming: Upcoming | None, *, current_way_id: int,
-              seconds: int, speed_mps: float | None, standstill: bool, can_offer: bool, now_ns: int):
+              seconds: int, speed_mps: float | None, standstill: bool, can_offer: bool,
+              confirmation_active: bool, now_ns: int):
     self.boundary_id, self.started_ns = '', None
     current = observations[sel.Source.MAP].candidate
     encounter = self.encounter
+    if encounter is not None and not encounter.declined and encounter.accepted_ns is None:
+      if not confirmation_active:
+        encounter.started_ns = None
+      elif encounter.started_ns is None:
+        encounter.started_ns = now_ns
     if current is None:
       if encounter is not None and not encounter.crossed:
         return {source: acc.Observation(acc.ObservationKind.UNKNOWN)

openpilot/starpilot/speed_limits/runtime.py

@@ -7,6 +7,7 @@ from dataclasses import dataclass, replace
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.selfdrive.car.cruise import V_CRUISE_MIN, V_CRUISE_MAX
+from openpilot.starpilot.aol.runtime import current_axis
 from openpilot.starpilot.speed_limits import acceptance as acc
 from openpilot.starpilot.speed_limits import action_arbitration as arb
 from openpilot.starpilot.speed_limits import composition as comp
@@ -384,6 +385,9 @@ class Runtime:
                      selfdrive.enabled and car_fresh and selfdrive_fresh)
     long_enabled = bool(car_fresh and controls_fresh and selfdrive_fresh and car_control_fresh and
                         selfdrive.enabled and sm['carControl'].enabled)
+    if 'aolAxisState' in sm.logMonoTime:
+      axis = current_axis(sm, now_ns=now_ns)
+      long_enabled = long_enabled and bool(axis is not None and axis.nativeAcknowledged and axis.desiredLongitudinal)
     lateral = bool(car_control_fresh and sm['carControl'].latActive and selfdrive_fresh)
     authority = _authority(CP, sm, host_long, lateral, long_enabled=long_enabled)
     authority = replace(authority, accelerator_override=bool(CP.openpilotLongitudinalControl and long_enabled and car_state.gasPressed))
@@ -415,7 +419,8 @@ class Runtime:
         current_way_id=self.map_source.way_id, seconds=self.settings.advance_seconds,
         speed_mps=float(car_state.vEgo) if ego_valid else None,
         standstill=bool(getattr(car_state, 'standstill', False)) or str(getattr(car_state, 'gearShifter', '')) == 'reverse',
-        can_offer=self.settings.enabled and authority.can_control and not host.force_decel, now_ns=now_ns)
+        can_offer=self.settings.enabled and authority.can_control and not host.force_decel,
+        confirmation_active=authority.can_control, now_ns=now_ns)
     return replace(frame, adoption_observations=proposals, confirmation_started_ns=self.advance.started_ns)
 
   def _action(self, request: Action | None, now_ns: int) -> tuple[acc.DriverAction | None, acc.AdoptRequest | None, int | None]:

openpilot/starpilot/speed_limits/tests/test_acceptance.py

@@ -42,12 +42,54 @@ class TestAcceptance(unittest.TestCase):
         self.assertIsNone(result.history_write)
         self.assertIsNone(result.control_target_mps)
 
+  def test_manual_confirmation_requires_longitudinal_engagement_for_every_source(self):
+    lateral_only = Authority(Mode.LATERAL_ONLY, LongitudinalOwner.NONE, True, False, False, False)
+    for source, authority in itertools.product(('dashboard', 'map', 'vision', 'online'), (DISENGAGED, lateral_only, PAUSED)):
+      with self.subTest(source=source, authority=authority):
+        candidate = replace(LOW, source=source)
+        inactive = step(saved(), valid(candidate), authority, Policy(), now_ns=0)
+        self.assertIsNone(inactive.state.pending)
+        self.assertFalse(inactive.state.timer_running)
+        self.assertEqual(inactive.display_candidate, candidate)
+        self.assertEqual(inactive.state.accepted, saved().accepted)
+        self.assertIsNone(inactive.history_write)
+        self.assertIsNone(inactive.control_target_mps)
+        engaged = step(inactive.state, valid(candidate), ACTIVE, Policy(), now_ns=20 * NS)
+        self.assertIsNotNone(engaged.state.pending)
+        self.assertEqual(engaged.state.pending.active_elapsed_ns, 0)
+
+  def test_disengagement_cancels_confirmation_without_declining_or_consuming_an_old_action(self):
+    pending = self.pending()
+    action = DriverAction('drive', 1, pending.state.pending.decision_id, ActionKind.ACCEPT)
+    inactive = step(pending.state, valid(LOW), DISENGAGED, Policy(), now_ns=5 * NS, action=action)
+    self.assertIsNone(inactive.state.pending)
+    self.assertFalse(inactive.state.timer_running)
+    self.assertFalse(inactive.state.rejected)
+    self.assertFalse(inactive.action_receipt.consumed)
+    self.assertEqual(inactive.state.accepted, pending.state.accepted)
+    resumed = step(inactive.state, valid(LOW), ACTIVE, Policy(), now_ns=20 * NS,
+                   action=replace(action, sequence_id=2))
+    self.assertIsNotNone(resumed.state.pending)
+    self.assertNotEqual(resumed.state.pending.decision_id, pending.state.pending.decision_id)
+    self.assertEqual(resumed.state.pending.active_elapsed_ns, 0)
+    self.assertFalse(resumed.action_receipt.consumed)
+    self.assertEqual(resumed.state.accepted, pending.state.accepted)
+
+  def test_accelerator_override_keeps_an_engaged_confirmation(self):
+    pending = self.pending()
+    overriding = replace(ACTIVE, longitudinal_active=False, accelerator_override=True)
+    held = step(pending.state, valid(LOW), overriding, Policy(), now_ns=NS)
+    self.assertEqual(held.state.pending.decision_id, pending.state.pending.decision_id)
+    accepted = self.act(held.state, ActionKind.ACCEPT, authority=overriding, now_ns=2 * NS)
+    self.assertTrue(accepted.action_receipt.consumed)
+    self.assertEqual(accepted.control_target_mps, LOW.speed_mps)
+
   def test_sources_share_confirmation_across_disengage_and_source_loss(self):
     for source, confirmation in itertools.product(('dashboard', 'map', 'vision', 'online'), (False, True)):
       policy = Policy(confirmation, confirmation)
       candidate = replace(LOW, source=source)
       first = step(new_session('drive'), valid(candidate), DISENGAGED, policy, now_ns=0)
-      self.assertEqual(first.state.pending is not None, confirmation)
+      self.assertIsNone(first.state.pending)
       self.assertEqual(first.history_write is None, confirmation)
       self.assertIsNone(first.control_target_mps)
       ready = step(first.state, valid(candidate), ACTIVE, policy, now_ns=NS)
@@ -190,13 +232,18 @@ class TestAcceptance(unittest.TestCase):
     self.assertIsNone(expired.history_write)
     self.assertEqual(expired.control_target_mps, HIGH.speed_mps)
 
-  def test_confirmation_expires_after_ten_seconds_even_without_active_longitudinal(self):
+  def test_disengagement_cancels_confirmation_instead_of_timing_out(self):
     start = self.pending()
     pause = step(start.state, valid(LOW), PAUSED, Policy(), now_ns=5 * NS)
-    self.assertEqual(pause.state.pending.active_elapsed_ns, 5 * NS)
-    expired = step(pause.state, valid(LOW), PAUSED, Policy(), now_ns=CONFIRMATION_NS)
+    self.assertIsNone(pause.state.pending)
+    inactive = step(pause.state, valid(LOW), PAUSED, Policy(), now_ns=CONFIRMATION_NS)
+    self.assertFalse(inactive.state.rejected)
+    self.assertIsNone(inactive.control_target_mps)
+    resumed = self.pending(state=inactive.state, now_ns=20 * NS)
+    self.assertNotEqual(resumed.state.pending.decision_id, start.state.pending.decision_id)
+    self.assertEqual(resumed.state.pending.active_elapsed_ns, 0)
+    expired = step(resumed.state, valid(LOW), ACTIVE, Policy(), now_ns=30 * NS)
     self.assertEqual(expired.basis, 'timeout_reject')
-    self.assertIsNone(expired.control_target_mps)
     self.assertEqual(expired.state.accepted.candidate, HIGH)
 
   def test_no_implicit_ttl_or_maximum_active_gap(self):
@@ -261,12 +308,13 @@ class TestAcceptance(unittest.TestCase):
     self.assertEqual(ready.control_target_mps, LOW.speed_mps)
 
   def test_zero_active_axes_does_not_imply_fully_disengaged(self):
-    result = self.pending(authority=PAUSED)
+    result = step(saved(), valid(LOW), PAUSED, Policy(), now_ns=0)
+    self.assertIsNone(result.state.pending)
     self.assertEqual(result.state.accepted.candidate, HIGH)
     self.assertIsNone(result.history_write)
     self.assertIsNone(result.control_target_mps)
 
-  def test_modes_stock_owner_and_inactive_long_never_gain_control_but_driver_is_still_asked(self):
+  def test_modes_stock_owner_and_inactive_long_cannot_confirm_or_gain_control(self):
     contexts = [
       Authority(Mode.OFF, LongitudinalOwner.NONE, False, False, False, False),
       Authority(Mode.LATERAL_ONLY, LongitudinalOwner.SYSTEM, True, False, False, False),
@@ -277,15 +325,16 @@ class TestAcceptance(unittest.TestCase):
     ]
     for authority in contexts:
       with self.subTest(authority=authority):
-        pending = self.pending(authority=authority)
+        pending = self.pending()
+        inactive = step(pending.state, valid(LOW), authority, Policy(), now_ns=NS)
         acted = self.act(pending.state, ActionKind.ACCEPT, authority=authority)
         lost = step(acted.state, ABSENT, authority, Policy(), now_ns=2 * NS)
-        for decision in (pending, acted, lost):
+        for decision in (inactive, acted, lost):
           self.assertIsNone(decision.control_target_mps)
-        self.assertIsNone(pending.history_write)
-        # The driver's choice is recorded even though nothing here may act on it.
-        self.assertTrue(acted.action_receipt.consumed)
-        self.assertEqual(acted.state.accepted.candidate, LOW)
+          self.assertIsNone(decision.state.pending)
+          self.assertIsNone(decision.history_write)
+        self.assertFalse(acted.action_receipt.consumed)
+        self.assertEqual(acted.state.accepted.candidate, HIGH)
         self.assertEqual(acted.state.last_action_sequence, 0)
 
   def test_combined_and_long_only_have_equal_longitudinal_decisions(self):
@@ -327,7 +376,7 @@ class TestAcceptance(unittest.TestCase):
 
   def test_repeated_and_out_of_order_actions_are_seen_but_not_consumed(self):
     pending = self.pending()
-    first = self.act(pending.state, ActionKind.ACCEPT, sequence=10, authority=PAUSED)
+    first = self.act(pending.state, ActionKind.ACCEPT, sequence=10)
     self.assertTrue(first.action_receipt.consumed)
     seen = replace(pending.state, last_action_sequence=10)
     for sequence in (10, 9):
@@ -662,11 +711,13 @@ class TestAcceptance(unittest.TestCase):
     shown = self.pending()
     request = replace(self.adopt_request(shown.state, sequence=5), presentation_id=999)
     paused = step(shown.state, valid(LOW), PAUSED, Policy(), now_ns=NS, adopt=request)
-    repeated_action = DriverAction("drive", 5, paused.state.pending.decision_id, ActionKind.ACCEPT)
+    self.assertIsNone(paused.state.pending)
+    repeated_action = DriverAction("drive", 5, shown.state.pending.decision_id, ActionKind.ACCEPT)
     repeated = step(paused.state, valid(LOW), ACTIVE, Policy(), now_ns=2 * NS, action=repeated_action)
     self.assertEqual(repeated.action_receipt.status, "already_seen")
     self.assertFalse(repeated.action_receipt.consumed)
-    fresh = step(repeated.state, valid(LOW), ACTIVE, Policy(), now_ns=3 * NS, action=replace(repeated_action, sequence_id=6))
+    fresh = step(repeated.state, valid(LOW), ACTIVE, Policy(), now_ns=3 * NS,
+                 action=replace(repeated_action, sequence_id=6, decision_id=repeated.state.pending.decision_id))
     self.assertTrue(fresh.action_receipt.consumed)
     old_adopt = step(fresh.state, valid(LOW), ACTIVE, Policy(), now_ns=4 * NS, adopt=request)
     self.assertFalse(old_adopt.adoption_receipt.consumed)

openpilot/starpilot/speed_limits/tests/test_advance.py

@@ -138,6 +138,7 @@ def test_confirmation_latches_on_slowdown_and_stop_then_times_out_once():
   drive = Drive(confirm=True)
   first = drive.tick().message.slcState
   start = drive.runtime.advance.encounter.started_ns
+  assert start is not None
   slower = drive.tick(speed=1., distance=59.).message.slcState
   stopped = drive.tick(speed=0., stopped=True, distance=59.).message.slcState
   assert first.decisionId == slower.decisionId == stopped.decisionId
@@ -152,6 +153,7 @@ def test_exact_deadline_action_wins_and_crossing_does_not_repeat(kind):
   drive = Drive(confirm=True)
   state = drive.tick().message.slcState
   start = drive.runtime.advance.encounter.started_ns
+  assert start is not None
   action = drive.action(state, kind)
   output = drive.tick(at=start+acc.CONFIRMATION_NS, action=action)
   assert output.message.slcState.actionStatus == kind
@@ -166,6 +168,7 @@ def test_recovery_renews_presentation_without_restarting_confirmation():
   drive = Drive(confirm=True)
   first = drive.tick().message.slcState
   start = drive.runtime.advance.encounter.started_ns
+  assert start is not None
   lost = drive.tick(alive=False)
   assert not lost.message.slcState.hasPending and lost.command is None
   recovered = drive.tick(at=start+9_000_000_000, producer=42).message.slcState
@@ -174,6 +177,30 @@ def test_recovery_renews_presentation_without_restarting_confirmation():
   assert not expired.message.slcState.hasPending and expired.command is None
 
 
+def test_disengagement_cancels_early_confirmation_and_reengagement_gets_a_fresh_deadline():
+  drive = Drive(confirm=True)
+  first = drive.tick().message.slcState
+  start = drive.now
+  drive.sm['carControl'].enabled = drive.sm['carControl'].longActive = False
+  drive.sm['selfdriveState'].enabled = False
+  drive.sm['controlsState'].longControlState = 'off'
+  inactive = drive.tick(at=start+2_000_000_000)
+  assert not inactive.message.slcState.hasPending and inactive.command is None
+  assert not drive.runtime.advance.encounter.declined
+  drive.sm['carControl'].enabled = drive.sm['carControl'].longActive = True
+  drive.sm['selfdriveState'].enabled = True
+  drive.sm['controlsState'].longControlState = 'pid'
+  resumed = drive.tick(at=start+20_000_000_000).message.slcState
+  assert resumed.hasPending and resumed.decisionId != first.decisionId
+  assert resumed.pendingSinceMonoTime == drive.now
+  restarted = drive.now
+  stale_action = drive.tick(action=drive.action(first, 'accept'))
+  assert stale_action.message.slcState.hasPending and stale_action.command is None
+  assert drive.tick(at=restarted+9_000_000_000).message.slcState.hasPending
+  assert not drive.tick(at=restarted+acc.CONFIRMATION_NS).message.slcState.hasPending
+  assert drive.runtime.advance.encounter.declined
+
+
 def test_fresh_different_boundary_supersedes_without_accepting_stale_action():
   drive = Drive(confirm=True)
   first = drive.tick().message.slcState

openpilot/starpilot/speed_limits/tests/test_confirmation_engagement.py

@@ -0,0 +1,73 @@
+"""Manual confirmations follow the longitudinal axis, including pedal override."""
+
+import pytest
+
+from openpilot.cereal import messaging
+from openpilot.starpilot.speed_limits.runtime import Action, Runtime
+from openpilot.starpilot.speed_limits.runtime_settings import parse
+from openpilot.starpilot.ui.tests.test_slc_ui_runtime import START, replay_inputs
+
+
+@pytest.mark.parametrize('mode', ['disengaged', 'aol', 'longitudinal', 'combined', 'accelerator'])
+def test_runtime_confirms_only_with_engaged_longitudinal_control(mode):
+  sm, cp = replay_inputs()
+  active = mode in ('longitudinal', 'combined')
+  engaged = active or mode == 'accelerator'
+  sm['selfdriveState'].enabled = sm['carControl'].enabled = engaged
+  sm['carControl'].longActive = active
+  sm['carControl'].latActive = mode in ('aol', 'combined')
+  sm['controlsState'].longControlState = 'pid' if active else 'off'
+  sm['carState'].gasPressed = mode == 'accelerator'
+  runtime = Runtime(parse({'SpeedLimitController': True, 'SLCConfirmation': True, 'SLCPriority1': 'Dashboard'}))
+  output = runtime.step(sm, cp, now_ns=START)
+  assert not output.result.errors
+  assert output.message.slcState.hasPending == engaged
+  assert output.message.slcState.observationKind == 'valid'
+  assert output.message.slcState.presentationId > 0
+  assert output.command is None
+
+
+def test_runtime_cancels_old_action_and_restarts_confirmation_after_disengagement():
+  sm, cp = replay_inputs()
+  runtime = Runtime(parse({'SpeedLimitController': True, 'SLCConfirmation': True, 'SLCPriority1': 'Dashboard'}))
+  first = runtime.step(sm, cp, now_ns=START).message.slcState
+  action = Action(runtime.session_id, 1, first.decisionId, first.presentationId, 'accept')
+  sm.advance(START+50_000_000)
+  sm['selfdriveState'].enabled = sm['carControl'].enabled = sm['carControl'].longActive = False
+  sm['carControl'].latActive = True
+  sm['controlsState'].longControlState = 'off'
+  inactive = runtime.step(sm, cp, now_ns=START+50_000_000, request=action)
+  assert not inactive.message.slcState.hasPending and not inactive.message.slcState.hasAccepted
+  assert inactive.command is None and not inactive.action_result.change
+  now = START+20_000_000_000
+  sm.advance(now)
+  sm['slcDashboardObservation'].observedMonoTime = now
+  sm['slcDashboardObservation'].validUntilMonoTime = now+1_000_000_000
+  sm['selfdriveState'].enabled = sm['carControl'].enabled = sm['carControl'].longActive = True
+  sm['controlsState'].longControlState = 'pid'
+  resumed = runtime.step(sm, cp, now_ns=now).message.slcState
+  assert resumed.hasPending and resumed.decisionId != first.decisionId
+  assert resumed.pendingSinceMonoTime == now and not resumed.hasAccepted
+
+
+@pytest.mark.parametrize('desired,acknowledged,fresh', [(False, True, True), (True, False, True),
+                                                       (True, True, False), (True, True, True)])
+def test_aol_pedal_override_requires_current_longitudinal_engagement(desired, acknowledged, fresh):
+  sm, cp = replay_inputs()
+  sm['carControl'].longActive = False
+  sm['carControl'].latActive = True
+  sm['carState'].gasPressed = True
+  sm['controlsState'].longControlState = 'off'
+  axis = messaging.new_message('aolAxisState').aolAxisState
+  axis.qualified, axis.nativeAcknowledged, axis.desiredLongitudinal = True, acknowledged, desired
+  axis.sessionId = 'aol-drive'
+  axis.observedMonoTime = START
+  axis.validUntilMonoTime = START+30_000_000
+  sm.data['aolAxisState'] = axis
+  sm.seen = {'aolAxisState': True}
+  sm.valid['aolAxisState'] = sm.alive['aolAxisState'] = True
+  sm.logMonoTime['aolAxisState'] = START if fresh else START-100_000_000
+  runtime = Runtime(parse({'SpeedLimitController': True, 'SLCConfirmation': True, 'SLCPriority1': 'Dashboard'}))
+  output = runtime.step(sm, cp, now_ns=START)
+  assert not output.result.errors
+  assert output.message.slcState.hasPending == (desired and acknowledged and fresh)

openpilot/starpilot/speed_limits/tests/test_runtime_settings.py

@@ -36,7 +36,7 @@ class RuntimeSettingsTests(unittest.TestCase):
                                 authority, policy, now_ns=1)
               self.assertEqual(result.state.accepted, accepted)
               self.assertIsNone(result.history_write)
-              self.assertIsNotNone(result.state.pending)
+              self.assertEqual(result.state.pending is not None, authority.can_control)
               self.assertEqual(result.control_target_mps, 20. if accepted is not None and authority.can_control else None)
 
   def test_fresh_vision_display_default_preserves_explicit_saved_choices(self):

openpilot/starpilot/ui/onroad_state.py

@@ -264,7 +264,8 @@ class OnroadRequest:
 def manual_slc_confirmation(profile: Profile, state: OnroadState) -> bool:
   obs = state.speed_limit
   widget = 'speed_limit' if profile == Profile.COMPACT else 'cruise_limits'
-  return bool(state.alert.size == AlertSize.NONE and state.toyota_cruise_status not in TOYOTA_NOTICE_STATUSES and
+  return bool((state.longitudinal_active or state.longitudinal_overridden) and
+              state.alert.size == AlertSize.NONE and state.toyota_cruise_status not in TOYOTA_NOTICE_STATUSES and
               (profile != Profile.COMPACT or not state.reverse_driver_camera and state.appearance.camera_view != CameraViewChoice.DRIVER) and
               obs.kind == ObservationKind.VALID and obs.action_enabled and not obs.automatic_adoption_pending and
               obs.pending_speed_limit_mps is not None and obs.pending_speed_limit_mps > 0 and

openpilot/starpilot/ui/tests/test_slc_confirmation.py

@@ -39,11 +39,31 @@ def test_closed_and_open_card_share_anchor_and_physically_scaled_targets(profile
     assert after[0] > 146
 
 
+@pytest.mark.parametrize('profile', list(Profile))
+@pytest.mark.parametrize('lateral', [False, True])
+def test_disengaged_and_aol_views_cancel_confirmation_but_keep_the_sign(profile, lateral):
+  current, emitted = state(), []
+  touch = OnroadInput(emitted.append, profile)
+  tap_sign(touch, current)
+  assert touch.present(current).slc_prompt_open
+  inactive = replace(current, longitudinal_active=False, lateral_active=lateral)
+  assert not manual_slc_confirmation(profile, inactive)
+  assert not touch.present(inactive).slc_prompt_open
+  assert not slc_controls(profile, inactive)
+  assert speed_limit_bounds(profile, inactive) is not None
+  assert not emitted
+
+
+@pytest.mark.parametrize('profile', list(Profile))
+def test_accelerator_override_keeps_confirmation_visible(profile):
+  current = replace(state(), longitudinal_active=False, longitudinal_overridden=True)
+  assert manual_slc_confirmation(profile, current)
+
+
 @pytest.mark.parametrize('profile', list(Profile))
 @pytest.mark.parametrize('action', [SlcActionKind.ACCEPT, SlcActionKind.REJECT])
 def test_touch_choice_remains_available_without_wheel_confirmation(profile, action):
-  current = replace(state(), longitudinal_active=False, slc_system_long_available=False,
-                    slc_wheel_confirmation_available=False)
+  current = replace(state(), slc_wheel_confirmation_available=False)
   emitted = []
   touch = OnroadInput(emitted.append, profile)
   tap_sign(touch, current)