SLC: Fix offset safety, retention, and UI

openpilot/selfdrive/controls/plannerd.py

@@ -2,6 +2,7 @@
 import os
 import json
 from collections import deque
+from dataclasses import replace
 
 from opendbc.car.structs import car
 from openpilot.common.params import Params
@@ -45,7 +46,7 @@ from openpilot.starpilot.speed_limits.runtime_settings import read_params as rea
 
 
 class SavedSlcLifecycle:
-  """Refresh saved admission; each replacement starts an independent SLC session."""
+  """Refresh saved admission; offset-only edits preserve the accepted drive limit."""
 
   def __init__(self, params, cp, environment):
     self.params, self.cp, self.environment = params, cp, dict(environment)
@@ -68,17 +69,22 @@ class SavedSlcLifecycle:
                      feature_enabled(self.params, self.cp, 'vision', {}, saved_slc=saved_slc))
     if configuration == self.configuration:
       return self.runtime, False
-    if self.runtime is not None:
-      self.runtime.reset()
+    if (self.runtime is not None and self.configuration is not None and
+        configuration == (self.configuration[0], replace(self.runtime.settings, offsets=settings.offsets),
+                          *self.configuration[2:])):
+      self.runtime.update_offsets(settings.offsets)
+    else:
+      if self.runtime is not None:
+        self.runtime.reset()
+      self.runtime = (SlcRuntime(settings, vision_enabled=vision,
+                                 vision_control_qualified=configuration[3],
+                                 vision_display_qualified=configuration[4])
+                      if configuration[0] else None)
     self.configuration = configuration
     if pending_actions is not None:
       pending_actions.clear()
     if pending_cruise is not None:
       pending_cruise.clear()
-    self.runtime = (SlcRuntime(settings, vision_enabled=vision,
-                               vision_control_qualified=configuration[3],
-                               vision_display_qualified=configuration[4])
-                    if configuration[0] else None)
     return self.runtime, True
 
 

openpilot/starpilot/speed_limits/README.md

@@ -35,9 +35,15 @@ it is the only numeric offset authority: old `Offset*` bytes and `IsMetric` are
 ignored for control, even if malformed. A malformed document or the explicit
 `needs_review` state disables SLC without falling back to legacy values or
 rewriting the saved control bit. The document helper exposes strict decoding,
-serialization and pure adoption of a currently valid legacy schedule. Native
-settings UI adoption and display-unit writes are a separate pending integration;
-this backend does not write Params.
+serialization and pure adoption of a currently valid legacy schedule. The shared
+native editor is available under Driving Controls on both screens. It shows the
+saved half-open boundaries, rounded to two decimals, without changing them.
+Offset-only edits preserve the drive session's accepted and rejected limits,
+renew pending action IDs, and apply the new adjustment on the next frame after
+the saved-settings refresh. Other configuration changes still replace the session.
+The editor disallows reductions that cannot produce a positive target anywhere
+in the band. At runtime, an offset that would remove the ceiling is ignored for
+that limit, retaining its unadjusted target instead.
 
 All transport additions use cereal's reserved custom messages. The official
 vehicle schema is unchanged. Slot 9 retains its six historical model-status fields

openpilot/starpilot/speed_limits/SPEED_DOMAIN.md

@@ -13,6 +13,10 @@ speed; resolution clamps each delta to `max(cluster - raw, 0)`. A nonvalid
 resolution with a context is malformed and triggers the override session-reset
 contract. Only a `VALID` context may enter `overrides.step(domain=...)`.
 
+A finite negative offset that would make the target nonpositive in cluster or
+planner coordinates is ignored for that limit. The unadjusted accepted limit
+continues to supply the ceiling; malformed schedules still fail validation.
+
 Call order is acceptance, domain resolution, override reduction, then
 `to_planner_coordinate(context, contribution_raw_mps, basis)`. The reducer
 compares current selected and retained intent against `accepted raw + offset`

openpilot/starpilot/speed_limits/runtime.py

@@ -2,7 +2,7 @@
 
 import math
 import uuid
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.services import SERVICE_LIST
@@ -181,6 +181,15 @@ class Runtime:
     self.command_target_mps = 0.0
     self.command_card_producer = ''
 
+  def update_offsets(self, offsets: sd.OffsetSchedule) -> None:
+    self.settings = replace(self.settings, offsets=offsets)
+    # Renew action IDs when the adjusted target changes, keeping accepted/rejected limits.
+    acceptance = replace(self.state.acceptance, observation=acc.Observation(acc.ObservationKind.UNKNOWN),
+                         pending=None, presentation=None, timer_running=False)
+    self.state = replace(self.state, acceptance=acceptance)
+    self._ui_receipt = None
+    self.command_status = ''
+
   def _event_current(self, event: CruiseEvent, sm, now_ns: int) -> bool:
     return (bool(event.producer_session) and 0 < event.observed_ns <= now_ns and
             now_ns - event.observed_ns <= int(2e9 / SERVICE_LIST['modelV2'].frequency) and
@@ -395,7 +404,7 @@ class Runtime:
       completed = arb.step(self.ledger, arb.CompleteAction(self.session_id, action_id), now_ns=now_ns)
       self.ledger = completed.state
     command = self._command(result, sm, CP, now_ns, old_accepted, old_pending, action, adopt)
-    message = self._message(now_ns, result, sm, frame.observations)
+    message = self._message(now_ns, result, sm, frame)
     if command is not None:
       self._command_ui_sequence = request.sequence_id if request is not None else 0
     message.slcState.commandActionSequenceId = getattr(self, '_command_ui_sequence', 0)
@@ -456,7 +465,7 @@ class Runtime:
     record.targetMps = target
     return command
 
-  def _message(self, now_ns: int, result: comp.Result, sm, observations=None):
+  def _message(self, now_ns: int, result: comp.Result, sm, frame: comp.Frame):
     message = messaging.new_message("slcState")
     message.logMonoTime = now_ns
     message.valid = not result.errors
@@ -469,7 +478,7 @@ class Runtime:
       enabled_sources.add(sel.Source.ONLINE)
     rows = state.init('sourceReadings', len(sel.Source))
     for row, source in zip(rows, sel.Source, strict=True):
-      observation = (observations or {}).get(source)
+      observation = frame.observations.get(source)
       row.source = source.value
       row.enabled = bool(self.settings.display and source in enabled_sources)
       row.observationKind = observation.kind.value if observation is not None else 'unknown'
@@ -510,12 +519,16 @@ class Runtime:
     state.overrideBasis = (result.override.basis if state.driverOverrideActive else 'none')
     state.retainedOverride = bool(result.state.override.persistent_selected_mps is not None)
     shown = result.state.acceptance.presentation
+    ego = frame.ego_pair
+    ego_delta = (max(0.0, ego.cluster_mps - ego.raw_mps)
+                 if ego.status is sd.DomainStatus.VALID and ego.cluster_mps is not None and ego.raw_mps is not None else 0.0)
     for candidate, value_field, flag_field in (
       (accepted.candidate if accepted is not None else None, 'acceptedAdjustedLimit', 'hasAcceptedAdjustedLimit'),
       (pending.candidate if pending is not None else None, 'pendingAdjustedLimit', 'hasPendingAdjustedLimit'),
       (shown.candidate if shown is not None else None, 'presentationAdjustedLimit', 'hasPresentationAdjustedLimit'),
     ):
-      value = sd.adjusted_limit(candidate.speed_mps, self.settings.offsets) if candidate is not None else None
+      value = (sd.adjusted_limit(candidate.speed_mps, self.settings.offsets, minimum_target_mps=ego_delta)
+               if candidate is not None else None)
       if value is not None:
         setattr(state, value_field, value)
         setattr(state, flag_field, True)

openpilot/starpilot/speed_limits/speed_domain.py

@@ -138,12 +138,14 @@ def offset_for_limit(speed_mps: float, schedule: OffsetSchedule) -> float | None
   return None
 
 
-def adjusted_limit(speed_mps: float, schedule: OffsetSchedule) -> float | None:
-  """Informational cluster limit using the same validated bands as control."""
+def adjusted_limit(speed_mps: float, schedule: OffsetSchedule, *, minimum_target_mps: float = 0.0) -> float | None:
+  """Cluster target; ignore a reduction that would remove the planner ceiling."""
   offset = offset_for_limit(speed_mps, schedule)
-  if offset is None:
+  if offset is None or not _number(minimum_target_mps):
     return None
   value = speed_mps + offset
+  if offset < 0 and value <= minimum_target_mps:
+    return speed_mps
   return value if _number(value, positive=True) else None
 
 
@@ -187,13 +189,16 @@ def resolve(decision: acc.Decision, schedule: OffsetSchedule, selected: SpeedPai
   offset = offset_for_limit(target, schedule)
   if offset is None:
     return DomainResolution(DomainStatus.UNAVAILABLE, reason="no configured offset band")
-  effective = target + offset
-  if not _number(effective, positive=True):
+  ego_delta = max(0.0, ego.cluster_mps - ego.raw_mps)
+  effective = adjusted_limit(target, schedule, minimum_target_mps=ego_delta)
+  if effective is None:
     return DomainResolution(DomainStatus.INVALID, reason="nonpositive effective target")
+  if effective == target:
+    offset = 0.0
   return DomainResolution(DomainStatus.VALID, DomainContext(
     decision.state.session_id, decision.state.last_timestamp_ns, target, offset, effective,
     selected.raw_mps, max(0.0, selected.cluster_mps - selected.raw_mps),
-    ego.raw_mps, max(0.0, ego.cluster_mps - ego.raw_mps)))
+    ego.raw_mps, ego_delta))
 
 
 def to_planner_coordinate(context: DomainContext, contribution_raw_mps: float | None, basis: str) -> PlannerCoordinate:

openpilot/starpilot/speed_limits/tests/test_speed_domain.py

@@ -60,6 +60,14 @@ class SpeedDomainTests(unittest.TestCase):
     self.assertEqual(pedal.cap_mps, 25.0)
     self.assertEqual(pedal.cluster_target_mps, 26.0)
 
+  def test_unsafe_negative_offsets_keep_the_unadjusted_ceiling(self):
+    for offset, delta in ((-20.0, 0.0), (-30.0, 0.0), (-19.0, 2.0)):
+      with self.subTest(offset=offset, delta=delta):
+        schedule = sd.OffsetSchedule((sd.OffsetBand(0.0, None, offset),))
+        _, coordinate = self.frame(0, schedule=schedule, ego_cluster=20.0 + delta)
+        self.assertEqual(coordinate.cluster_target_mps, 20.0)
+        self.assertEqual(coordinate.cap_mps, 20.0 - delta)
+
   def test_effective_threshold_blocks_early_intent_and_cluster_only_pedal_can_cross(self):
     offset_three = sd.OffsetSchedule((sd.OffsetBand(0.0, None, 3.0),))
     first, _ = self.frame(0, selected_raw=20.0, ego_raw=20.0, schedule=offset_three)

openpilot/starpilot/ui/onroad_state.py

@@ -252,19 +252,19 @@ def _default_slc_controls(profile: Profile, state: OnroadState) -> tuple[SlcCont
     def request(kind: SlcActionKind) -> SlcUiRequest:
       return SlcUiRequest(kind, observation.session_id or "", observation.decision_id or 0,
                           observation.presentation_id or 0, pending)
+    target = observation.pending_adjusted_limit_mps
+    label = f'Apply {round(target * (3.6 if state.metric else 2.2369362921))}' if target is not None and target > 0 else 'Apply'
+    # Name the speed each choice leaves you at: the new target, or the limit already in use.
+    current = observation.accepted_adjusted_limit_mps
+    keep = f'Keep {round(current * (3.6 if state.metric else 2.2369362921))}' if current is not None and current > 0 else 'Keep'
     if profile == Profile.LARGE:
       from openpilot.starpilot.ui.large_speed_geometry import WIDTH, ACTION_HEIGHT, ACTION_X, ACTION_Y, ACTION_HEADER_HEIGHT
-      target = observation.pending_adjusted_limit_mps
-      label = f'Apply {round(target * (3.6 if state.metric else 2.2369362921))}' if target is not None and target > 0 else 'Apply'
-      # Name the speed each choice leaves you at: the new target, or the limit already in use.
-      current = observation.accepted_adjusted_limit_mps
-      keep = f'Keep {round(current * (3.6 if state.metric else 2.2369362921))}' if current is not None and current > 0 else 'Keep'
       return (SlcControl(request(SlcActionKind.ACCEPT), label,
                          (ACTION_X, ACTION_Y + ACTION_HEADER_HEIGHT, ACTION_X + WIDTH // 2 - 6, ACTION_Y + ACTION_HEIGHT)),
               SlcControl(request(SlcActionKind.REJECT), keep,
                          (ACTION_X + WIDTH // 2 + 6, ACTION_Y + ACTION_HEADER_HEIGHT, ACTION_X + WIDTH, ACTION_Y + ACTION_HEIGHT)))
-    return (SlcControl(request(SlcActionKind.ACCEPT), "ACCEPT", (174, 180, 310, 234)),
-            SlcControl(request(SlcActionKind.REJECT), "REJECT", (320, 180, 456, 234)))
+    return (SlcControl(request(SlcActionKind.ACCEPT), label, (174, 180, 310, 234)),
+            SlcControl(request(SlcActionKind.REJECT), keep, (320, 180, 456, 234)))
   if (pending is None and observation.speed_limit_mps is not None and
       observation.presentation_id is not None and observation.presentation_id > 0):
     if profile == Profile.COMPACT:

openpilot/starpilot/ui/runtime_app.py

@@ -2115,6 +2115,10 @@ class StarMiciMainLayout(MiciMainLayout):
       from openpilot.starpilot.ui.models_compact import ModelsCompact
       ModelsCompact(self.star).open()
       return
+    if destination == Destination.DRIVING_CONTROLS:
+      from openpilot.starpilot.ui.feature_settings_compact import FeatureSettingsCompact
+      FeatureSettingsCompact(self.star).open(FeaturePage.HUB)
+      return
     if destination == Destination.APPEARANCE:
       from openpilot.starpilot.ui.appearance_compact import AppearanceCompact
       AppearanceCompact(self.star).open()

openpilot/starpilot/ui/settings_state.py

@@ -57,6 +57,7 @@ COMPACT_MENU = (
   (Destination.DEVICE, "Device", "icons_mici/settings/device_icon.png", 72, 58),
   (Destination.SOFTWARE, "Software", "icons_mici/settings/device/update.png", 64, 75),
   (Destination.DRIVING_MODEL, "Driving Model", "icons_mici/settings/device/lkas.png", 72, 56),
+  (Destination.DRIVING_CONTROLS, "Driving Controls", "icons_mici/settings/device/lkas.png", 72, 56),
   (Destination.APPEARANCE, "Visuals", "icons_mici/settings/device/cameras.png", 64, 64),
   (Destination.GALAXY, "Galaxy", "icons_mici/settings/galaxy.png", 64, 64),
   (Destination.PAIR, "Pair to Connect", "icons_mici/settings/comma_icon.png", 33, 60),

openpilot/starpilot/ui/slc_offset_feature.py

@@ -57,6 +57,12 @@ def _unit_factor(metric: bool) -> float:
   return od.KPH_TO_MPS if metric else od.MPH_TO_MPS
 
 
+def _minimum_offset(document: od.OffsetDocument, index: int, metric: bool) -> float:
+  minimum = -150 if metric else -99
+  boundary = -document.bounds_mps[index + 1] / _unit_factor(metric)
+  return max(minimum, math.floor(boundary) + 1) if math.isfinite(boundary) else minimum
+
+
 def _display(value: float) -> str:
   return f"{value:.2f}".rstrip("0").rstrip(".")
 
@@ -141,22 +147,20 @@ class SlcOffsetOwner:
     if isinstance(snap.document, od.OffsetDocument) and snap.metric is not None:
       unit = "km/h" if snap.metric else "mph"
       scale = 1 / _unit_factor(snap.metric)
-      display_bounds = ((0, 25, 35, 45, 55, 65, 75, 100) if not snap.metric and snap.document.bounds_mps == od.IMPERIAL_BOUNDS else
-                        (0, 30, 50, 60, 80, 100, 120, 140) if snap.metric and snap.document.bounds_mps == od.METRIC_BOUNDS else
-                        tuple(round(bound * scale) for bound in snap.document.bounds_mps))
+      display_bounds = tuple(_display(bound * scale) for bound in snap.document.bounds_mps)
       for index in range(7):
         # Round presentation only: the stored SI boundaries remain authoritative.
-        label = f"{display_bounds[index]}–{display_bounds[index + 1]} {unit}"
+        label = f"{display_bounds[index]}–<{display_bounds[index + 1]} {unit}"
         value = _display(snap.document.offsets_mps[index] * scale)
         rows.append(FeatureRow(OFFSET_KEYS[index], label + " offset", value, snap.raw_document,
                                step=1 if snap.raw_document is not None else 0, unit=unit,
-                               minimum=-150 if snap.metric else -99, maximum=150 if snap.metric else 99,
+                               minimum=_minimum_offset(snap.document, index, snap.metric), maximum=150 if snap.metric else 99,
                                available=available and snap.raw_document is not None,
                                reason="Legacy value; adopt to edit" if snap.raw_document is None else
-                                      "Speed ranges are rounded for display; saved limits keep their exact values",
+                                      "Upper boundary excluded; ranges rounded to 2 decimals. Unsafe reductions use the unadjusted limit",
                                related_source=snap.raw_unit, capability=capable, dependencies=deps,
                                display_unit=unit, default_value="0"))
-      rows.append(FeatureRow("", "Above final band", "Zero offset"))
+      rows.append(FeatureRow("", f"{display_bounds[-1]} {unit} and above", "Zero offset"))
     else:
       for index, key in enumerate(OFFSET_KEYS, start=1):
         label = f"Band {index} offset (invalid saved units)" if snap.metric is None else f"Band {index} offset"
@@ -214,14 +218,14 @@ class SlcOffsetOwner:
             return False
           index = OFFSET_KEYS.index(request.key)
           current = snap.document.offsets_mps[index]
-          shown = current / _unit_factor(snap.metric)
-          candidate_display = float(request.value) if request.direction == 0 else shown + request.direction
-          low, high = (-150.0, 150.0) if snap.metric else (-99.0, 99.0)
-          if not math.isfinite(candidate_display) or not low <= candidate_display <= high:
+          low, high = _minimum_offset(snap.document, index, snap.metric), 150.0 if snap.metric else 99.0
+          factor = _unit_factor(snap.metric)
+          candidate = (float(request.value) * factor if request.direction == 0 else
+                       max(low * factor, min(high * factor, current + request.direction * factor)))
+          if not math.isfinite(candidate) or not low * factor <= candidate <= high * factor:
             return False
           new_offsets = list(snap.document.offsets_mps)
-          new_offsets[index] = (candidate_display * _unit_factor(snap.metric) if request.direction == 0 else
-                                current + request.direction * _unit_factor(snap.metric))
+          new_offsets[index] = candidate
           result = od.OffsetDocument(snap.document.bounds_mps, tuple(new_offsets))
         value = od.to_value(result)
         if not self._same(request, snap, require_capability=True):

openpilot/starpilot/ui/tests/test_layout_preview_renderer.py

@@ -110,7 +110,7 @@ class LayoutPreviewRendererTests(unittest.TestCase):
     pending = sample_state("slc_pending", document)
     for profile in (Profile.LARGE, Profile.COMPACT):
       self.assertEqual([item.label for item in slc_controls(profile, pending)],
-                       ["Apply 54", "Keep 49"] if profile == Profile.LARGE else ["ACCEPT", "REJECT"])
+                       ["Apply 54", "Keep 49"])
       # The engaged sample already uses its accepted limit, so there is no redundant adopt action.
       self.assertEqual(slc_controls(profile, sample_state("engaged", document)), ())
     manual = sample_state("experimental", document)

openpilot/starpilot/ui/tests/test_runtime_navigation.py

@@ -37,7 +37,8 @@ class TestCompactSettingsNavigation(unittest.TestCase):
   def test_scrolled_device_and_developer_cards_emit_distinct_destinations(self):
     emitted = []
     touch = SettingsInput(Profile.COMPACT, emitted.append)
-    for index, destination in ((5, Destination.DEVICE), (9, Destination.GALAXY), (11, Destination.DEVELOPER)):
+    for destination in (Destination.DEVICE, Destination.GALAXY, Destination.DEVELOPER, Destination.DRIVING_CONTROLS):
+      index = next(index for index, item in enumerate(COMPACT_MENU) if item[0] == destination)
       state = SettingsState(compact_scroll_x=-422 * index)
       touch.press(45, 75, state)
       touch.release(45, 75, state)
@@ -52,10 +53,21 @@ class TestCompactSettingsNavigation(unittest.TestCase):
     touch.release(90, 75, state)
     self.assertEqual(emitted, [])
 
-  def test_compact_main_menu_matches_original_order(self):
+  def test_compact_main_menu_includes_shared_driving_controls(self):
     self.assertEqual([label.lower() for _, label, _, _, _ in COMPACT_MENU],
                      ["toggles", "network", "bluetooth", "force drive state", "vehicle", "device", "software",
-                      "driving model", "visuals", "galaxy", "pair to connect", "developer"])
+                      "driving model", "driving controls", "visuals", "galaxy", "pair to connect", "developer"])
+
+  def test_compact_driving_controls_opens_the_shared_feature_editor(self):
+    from openpilot.starpilot.ui.runtime_app import StarMiciMainLayout
+    from openpilot.starpilot.ui.feature_settings_state import FeaturePage
+    layout = StarMiciMainLayout.__new__(StarMiciMainLayout)
+    layout.star = Mock()
+    with patch('openpilot.starpilot.ui.feature_settings_compact.FeatureSettingsCompact') as editor:
+      layout._open_compact_destination(Destination.DRIVING_CONTROLS)
+    editor.assert_called_once_with(layout.star)
+    editor.return_value.open.assert_called_once_with(FeaturePage.HUB)
+    layout.star._unavailable.assert_not_called()
 
 
 class TestExistingToggleOwner(unittest.TestCase):

openpilot/starpilot/ui/tests/test_slc_offset_feature.py

@@ -1,6 +1,7 @@
 """SLC native offset actions and both unit paths against real temporary Params."""
 
 from pathlib import Path
+from dataclasses import replace
 from types import SimpleNamespace as NS
 from typing import Any, cast
 import tempfile
@@ -144,16 +145,52 @@ class SlcOffsetFeatureTests(unittest.TestCase):
     self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), adjusted)
     self.assertEqual((self.raw("Offset1"), self.raw("Offset7")), old)
 
-  def test_rounded_range_labels_do_not_rewrite_si_bounds(self):
+  def test_range_labels_follow_saved_si_bounds_without_rewriting_them(self):
     document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0.0,) * 7)
     self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
     original = self.raw(DOCUMENT_KEY)
     rows = [self.row(f"Offset{index}").label for index in range(1, 8)]
-    self.assertEqual(rows, [f"{low}–{high} mph offset" for low, high in
-                           ((0, 25), (25, 35), (35, 45), (45, 55), (55, 65), (65, 75), (75, 100))])
+    self.assertEqual(rows, [f"{low}–<{high} mph offset" for low, high in
+                           (("0", "25.05"), ("25.05", "34"), ("34", "43.84"), ("43.84", "53.91"),
+                            ("53.91", "63.98"), ("63.98", "74.04"), ("74.04", "98.87"))])
+    state = self.owner.snapshot("slc", parked=True, system_long=True, lateral_context=True, metric=False)
+    self.assertTrue(any(row.label == "98.87 mph and above" and row.value == "Zero offset" for row in state.rows))
     self.assertEqual(self.raw(DOCUMENT_KEY), original)
     self.assertTrue(self.units.change_units(True, None))
     self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
+    self.assertEqual(self.row("Offset4").label, "70.56–<86.76 km/h offset")
+
+  def test_editor_rejects_offsets_that_cannot_produce_a_positive_target_in_the_band(self):
+    for metric, offset in ((False, -30), (True, -45)):
+      with self.subTest(metric=metric):
+        self.params.put_bool("IsMetric", metric, block=True)
+        document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0.0,) * 7)
+        self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
+        original = self.raw(DOCUMENT_KEY)
+        row = self.row("Offset1")
+        request = row_change(row)
+        assert request is not None
+        self.assertGreater(row.minimum, offset)
+        self.assertFalse(self.owner.apply(replace(request, value=str(offset), direction=0)))
+        self.assertEqual(self.raw(DOCUMENT_KEY), original)
+        # Useful reductions remain supported, including in the first band.
+        self.assertTrue(self.owner.apply(replace(request, value="-1", direction=0)))
+
+  def test_stepper_can_repair_an_unsafe_saved_offset_and_clamps_at_the_band_minimum(self):
+    document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (-30 * od.MPH_TO_MPS, 0, 0, 0, 0, 0, 0))
+    self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
+    row = self.row('Offset1')
+    request = row_change(row)
+    assert request is not None
+    self.assertTrue(self.owner.apply(request))
+    self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).offsets_mps[0], row.minimum * od.MPH_TO_MPS)
+    # Unit conversion can leave a fractional value less than one step above the minimum.
+    document = od.OffsetDocument(od.IMPERIAL_BOUNDS, ((row.minimum + .5) * od.MPH_TO_MPS, 0, 0, 0, 0, 0, 0))
+    self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
+    request = row_change(self.row('Offset1'), -1)
+    assert request is not None
+    self.assertTrue(self.owner.apply(request))
+    self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).offsets_mps[0], row.minimum * od.MPH_TO_MPS)
 
   def test_onroad_adopted_offsets_edit_without_repair_authority(self):
     self.params.put_bool("IsMetric", False, block=True)

openpilot/starpilot/ui/tests/test_slc_retention.py

@@ -8,6 +8,8 @@ import pytest
 
 from openpilot.common.params import Params
 from openpilot.selfdrive.controls.lib import longitudinal_planner as native
+from openpilot.selfdrive.controls.plannerd import SavedSlcLifecycle
+from openpilot.starpilot.speed_limits import offset_document as od
 from openpilot.starpilot.speed_limits.runtime import Action, Runtime
 from openpilot.starpilot.speed_limits.runtime_settings import read_params
 from openpilot.starpilot.speed_limits.tests.test_maps_runtime import BOOT, packet
@@ -22,6 +24,100 @@ from openpilot.starpilot.ui.unified_speed_presentation import resolve_unified_sp
 MPH = .44704
 
 
+@pytest.mark.parametrize('document', [False, True])
+@pytest.mark.parametrize('offset,delta', [(-25, 0), (-30, 0), (-24, 2), (-5, 0)])
+def test_negative_offsets_cannot_remove_the_native_slc_ceiling(tmp_path, document, offset, delta):
+  drive = RetentionDrive(tmp_path)
+  params = Params(str(tmp_path))
+  if document:
+    params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (offset, 0, 0, 0, 0, 0, 0))), block=True)
+  else:
+    params.put('Offset1', float(offset), block=True)
+  drive.runtime = Runtime(read_params(params))
+  drive.sm['carState'].vEgoCluster += delta
+  output, _, target = drive.step(0, limit=25)
+  adjusted = 20 * MPH if offset == -5 else 25 * MPH
+  assert not output.result.errors
+  assert output.message.slcState.hasCeiling
+  assert output.message.slcState.acceptedAdjustedLimit == pytest.approx(adjusted)
+  assert output.message.slcState.effectiveClusterTarget == pytest.approx(adjusted)
+  assert target == pytest.approx((adjusted - delta) / MPH)
+
+
+@pytest.mark.parametrize('confirm', [False, True])
+@pytest.mark.parametrize('source_gap', [False, True])
+def test_live_offset_edits_keep_the_accepted_limit_and_native_ceiling(tmp_path, confirm, source_gap):
+  drive = RetentionDrive(tmp_path, confirm=confirm)
+  params = Params(str(tmp_path))
+  lifecycle = SavedSlcLifecycle(params, drive.cp, {'SLC_REPLAY_RUNTIME': '1'})
+  runtime, _ = lifecycle.refresh(NOW)
+  assert runtime is not None
+  drive.runtime = runtime
+  first, _, _ = drive.step(0)
+  output, _, _ = drive.step(50, request=Action(runtime.session_id, 1, first.message.slcState.decisionId, 0, 'accept') if confirm else None)
+  accepted = output.result.state.acceptance.accepted
+  session = runtime.session_id
+  for milliseconds, offset in ((1000, 5), (2000, -5)):
+    params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, offset, 0, 0, 0))), block=True)
+    actions, cruise = [object()], [object()]
+    refreshed, changed = lifecycle.refresh(NOW + milliseconds * 1_000_000, pending_actions=actions, pending_cruise=cruise)
+    assert changed and refreshed is runtime
+    assert not actions and not cruise
+    output, _, target = drive.step(milliseconds, received=not source_gap)
+    assert not output.result.errors
+    assert runtime.session_id == session
+    assert output.result.state.acceptance.accepted == accepted
+    assert not output.message.slcState.hasPending
+    assert output.message.slcState.hasCeiling
+    assert target == pytest.approx(45 + offset)
+
+
+@pytest.mark.parametrize('accepted_first', [False, True])
+def test_offset_edits_invalidate_old_prompts_without_accepting_a_new_limit(tmp_path, accepted_first):
+  drive = RetentionDrive(tmp_path, confirm=True)
+  params = Params(str(tmp_path))
+  lifecycle = SavedSlcLifecycle(params, drive.cp, {'SLC_REPLAY_RUNTIME': '1'})
+  runtime, _ = lifecycle.refresh(NOW)
+  assert runtime is not None
+  drive.runtime = runtime
+  first, _, _ = drive.step(0)
+  if accepted_first:
+    drive.step(50, request=Action(runtime.session_id, 1, first.message.slcState.decisionId, 0, 'accept'))
+  old, _, _ = drive.step(100, limit=50)
+  old_id = old.message.slcState.decisionId
+  params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, 5, 0, 0, 0))), block=True)
+  assert lifecycle.refresh(NOW + 1_000_000_000)[0] is runtime
+  stale = Action(runtime.session_id, 2, old_id, old.message.slcState.presentationId, 'accept')
+  output, _, target = drive.step(1000, limit=50, request=stale)
+  assert not output.result.errors
+  assert output.message.slcState.hasPending
+  assert output.message.slcState.decisionId != old_id
+  assert output.message.slcState.presentationId != old.message.slcState.presentationId
+  assert output.message.slcState.hasAccepted == accepted_first
+  assert target == pytest.approx(50 if accepted_first else 70)
+
+
+def test_offset_edits_preserve_rejected_limits(tmp_path):
+  drive = RetentionDrive(tmp_path, confirm=True)
+  params = Params(str(tmp_path))
+  lifecycle = SavedSlcLifecycle(params, drive.cp, {'SLC_REPLAY_RUNTIME': '1'})
+  runtime, _ = lifecycle.refresh(NOW)
+  assert runtime is not None
+  drive.runtime = runtime
+  first, _, _ = drive.step(0)
+  drive.step(50, request=Action(runtime.session_id, 1, first.message.slcState.decisionId, 0, 'accept'))
+  proposed, _, _ = drive.step(100, limit=50)
+  drive.step(150, limit=50, request=Action(runtime.session_id, 2, proposed.message.slcState.decisionId, 0, 'reject'))
+  rejected = runtime.state.acceptance.rejected
+  params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, 5, 0, 0, 0))), block=True)
+  assert lifecycle.refresh(NOW + 1_000_000_000)[0] is runtime
+  output, _, target = drive.step(1000, limit=50)
+  assert not output.message.slcState.hasPending
+  assert output.result.state.acceptance.rejected == rejected
+  assert output.message.slcState.acceptedSpeedLimit == pytest.approx(45 * MPH)
+  assert target == pytest.approx(50)
+
+
 class RetentionDrive:
   def __init__(self, path, *, keep=True, confirm=False, dashboard=False):
     params = Params(str(path))

openpilot/starpilot/ui/tests/test_slc_ui_actions.py

@@ -35,6 +35,17 @@ def native_state(*, pending: bool = True, session: str = "drive-session", decisi
 
 
 class TestSlcTouch(unittest.TestCase):
+  def test_both_profiles_name_the_adjusted_apply_and_keep_targets(self):
+    for metric, factor in ((False, .44704), (True, 1 / 3.6)):
+      current = state()
+      current = replace(current, metric=metric, speed_limit=replace(current.speed_limit,
+        pending_speed_limit_mps=45 * factor, pending_adjusted_limit_mps=50 * factor,
+        accepted_speed_limit_mps=35 * factor, accepted_adjusted_limit_mps=40 * factor))
+      for profile in Profile:
+        controls = slc_controls(profile, current)
+        self.assertEqual([control.label for control in controls], ['Apply 50', 'Keep 40'])
+        self.assertEqual(controls[0].request.candidate_speed_mps, 45 * factor)
+
   def test_shell_routes_visible_card_action_as_typed_slc_request(self):
     emitted: list[ShellRequest] = []
     shell = ShellInput(Profile.LARGE, emitted.append)

openpilot/starpilot/ui/tests/test_speed_stack.py

@@ -25,7 +25,7 @@ def test_offset_proposals_use_their_own_band_and_reject_invalid_values():
   assert adjusted_limit(19, schedule) == 20
   assert adjusted_limit(20, schedule) == 23
   assert adjusted_limit(float('nan'), schedule) is None
-  assert adjusted_limit(10, OffsetSchedule((OffsetBand(0, None, -20),))) is None
+  assert adjusted_limit(10, OffsetSchedule((OffsetBand(0, None, -20),))) == 10
 
 
 def test_numeric_card_and_sources_row_are_not_tap_targets_while_pending():