Block unresolved Elantra CAN interrupt faults

Hold the mock background upload until in-flight assertions finish and join it on every exit. Preserve strict completion and payload checks.

Type the dynamically populated startup fixture namespace as object and assert its main entry is callable before invocation. Preserve startup ordering assertions and refresh the canonical protected inventory receipt.

.github/workflows/tests.yaml

@@ -343,6 +343,8 @@ jobs:
       run: scons
     - name: Gate every assigned lifecycle module serially
       timeout-minutes: 10
+      env:
+        RAYLIB_BACKEND: headless
       run: python3 tools/ci/run_vehicle_tests.py --suite aol-lifecycle --shard-index "${{ matrix.shard }}" --shard-count 16 --output "ci-results/aol-lifecycle-${{ matrix.shard }}"
     - name: Save every lifecycle outcome and native build
       if: always()
@@ -443,7 +445,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        shard: [0, 1, 2, 3]
+        shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
     env:
       FUZZ_SEED: '0'
     steps:
@@ -453,7 +455,9 @@ jobs:
       run: scons
     - name: Run every assigned RELEASE target serially
       timeout-minutes: 20
-      run: python3 tools/ci/run_vehicle_tests.py --suite safety-release --shard-index "${{ matrix.shard }}" --shard-count 4 --output "ci-results/safety-release-${{ matrix.shard }}"
+      env:
+        RAYLIB_BACKEND: headless
+      run: python3 tools/ci/run_vehicle_tests.py --suite safety-release --shard-index "${{ matrix.shard }}" --shard-count 16 --output "ci-results/safety-release-${{ matrix.shard }}"
     - name: Save every RELEASE outcome and native build
       if: always()
       uses: actions/upload-artifact@v4
@@ -476,7 +480,7 @@ jobs:
         pattern: safety-release-*
         path: ci-results/safety-release-shards
     - name: Require every RELEASE case exactly once and passed
-      run: python3 tools/ci/lifecycle_shards.py ci-results/safety-release-shards --suite safety-release --count 4 --output ci-results/safety-release-complete.json
+      run: python3 tools/ci/lifecycle_shards.py ci-results/safety-release-shards --suite safety-release --count 16 --output ci-results/safety-release-complete.json
     - name: Save complete RELEASE verdict
       if: always()
       uses: actions/upload-artifact@v4

launch_chffrplus.sh

@@ -57,6 +57,37 @@ EOF
   fi
 }
 
+function update_is_ready {
+  [ -f "$1" ] && [ ! -L "$1" ]
+}
+
+# A validated legacy return records the OS it was staged against. Check it on
+# every boot before swapping trees, including manual reboots.
+# Keep this in sync with TRUSTED_LEGACY_RELEASES in updated.py.
+TRUSTED_LEGACY_STARPILOT_COMMIT="4a553da9ac22a14e34513c011aed04c1e93b4619"
+
+function staged_os_is_compatible {
+  local staged="$1" marker="$1/.git/starpilot-retain-agnos" commit
+  commit="$(git -C "$staged" rev-parse --verify 'HEAD^{commit}')" || return 1
+  if [ "$commit" = "$TRUSTED_LEGACY_STARPILOT_COMMIT" ]; then
+    [ -f "$marker" ] && [ -s "$marker" ] && [ ! -L "$marker" ] && [ -r /VERSION ] && [ "$(cat "$marker")" = "$(< /VERSION)" ]
+  else
+    [ ! -e "$marker" ] && [ ! -L "$marker" ]
+  fi
+}
+
+function reset_update_os_environment {
+  unset AGNOS_VERSION AGNOS_ACCEPTED_VERSIONS
+}
+
+function maybe_activate_legacy_agnos {
+  local repo="$1" verifier="$2" manifest="$3" commit
+  commit="$(git -C "$repo" rev-parse --verify 'HEAD^{commit}')" || return 1
+  if [ "$commit" != "$TRUSTED_LEGACY_STARPILOT_COMMIT" ] && [ ! -L "$manifest" ]; then
+    "$verifier" --verify "$manifest" && sudo reboot
+  fi
+}
+
 function launch {
   # Remove orphaned git lock if it exists on boot
   [ -f "$DIR/.git/index.lock" ] && rm -f "$DIR/.git/index.lock"
@@ -75,8 +106,10 @@ function launch {
     if [ $? -eq 0 ]; then
       echo "${DIR} has been modified, skipping overlay update installation"
     else
-      if [ -f "${STAGING_ROOT}/finalized/.overlay_consistent" ]; then
-        if [ ! -d /data/safe_staging/old_openpilot ]; then
+      if update_is_ready "${STAGING_ROOT}/finalized/.overlay_consistent"; then
+        if ! staged_os_is_compatible "${STAGING_ROOT}/finalized"; then
+          echo "Staged update requires a different or unverified AGNOS version; skipping installation."
+        elif [ ! -d /data/safe_staging/old_openpilot ]; then
           echo "Valid overlay update found, installing"
           LAUNCHER_LOCATION="${BASH_SOURCE[0]}"
 
@@ -84,10 +117,10 @@ function launch {
           mv "${STAGING_ROOT}/finalized" "$DIR"
           cd "$DIR"
           # releases before the openpilot/ move can't swap AGNOS on newer AGNOS, use ours
-          [ ! -L system/hardware/tici/agnos.json ] && /data/safe_staging/old_openpilot/openpilot/common/hardware/comma/agnos.py --verify system/hardware/tici/agnos.json && sudo reboot
+          maybe_activate_legacy_agnos "$DIR" /data/safe_staging/old_openpilot/openpilot/common/hardware/comma/agnos.py system/hardware/tici/agnos.json
 
           echo "Restarting launch script ${LAUNCHER_LOCATION}"
-          unset AGNOS_VERSION
+          reset_update_os_environment
           exec "${LAUNCHER_LOCATION}"
         else
           echo "openpilot backup found, not updating"

opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_handoff.py

@@ -244,14 +244,14 @@ class TestIoniq6PrepublicationHandoff(unittest.TestCase):
         cp = CarInterface.get_params(model, fp, [], False, False, False)
         ci = CarInterface(cp)
         packer = CANPacker(DBC[model][Bus.pt])
-        ci.update((1_000_000_000, []))
+        ci.update([(1_000_000_000, [])])
         sources = [(parser, state) for parser in ci.can_parsers.values()
                    for state in parser.message_states.values() if not state.ignore_alive and state.address != 0x2E0]
         for frame in range(120):
           packets = [packer.make_can_msg(state.name, parser.bus,
                                         {"COUNTER": frame % 256} if any(sig.name == "COUNTER" for sig in state.signals) else {})
                      for parser, state in sources]
-          state = ci.update((1_000_000_000 + frame * 100_000_000, packets))
+          state = ci.update([(1_000_000_000 + frame * 100_000_000, packets)])
         self.assertTrue(state.canValid)
         self.assertFalse(cp.openpilotLongitudinalControl)
         self.assertTrue(cp.pcmCruise)
@@ -261,7 +261,7 @@ class TestIoniq6PrepublicationHandoff(unittest.TestCase):
           packets = [packer.make_can_msg(msg.name, parser.bus,
                                         {"COUNTER": frame % 256} if any(sig.name == "COUNTER" for sig in msg.signals) else {})
                      for parser, msg in sources if msg.name != "MDPS"]
-          state = ci.update((1_000_000_000 + frame * 100_000_000, packets))
+          state = ci.update([(1_000_000_000 + frame * 100_000_000, packets)])
         self.assertFalse(state.canValid)
 
   def test_card_tags_aol_candidate_before_cp_publication_without_disabling_ecu(self):

opendbc_repo/opendbc/car/toyota/radar_interface.py

@@ -77,7 +77,12 @@ class RadarInterface(RadarInterfaceBase):
     if not self.rcp.can_valid:
       ret.errors.canError = True
 
-    if self.rcp.vl['STATUS_MSG']['RADAR_STATUS'] != 1 or self.rcp.vl['STATUS_MSG']['RADAR_PRE_FAULT'] != 0:
+    status = self.rcp.vl['STATUS_MSG']
+    pre_fault = status['RADAR_PRE_FAULT']
+    # Legacy pre-warning can rise on an empty road before the documented dirty-radar threshold.
+    legacy_pre_warning = (not self.CP.flags & ToyotaFlags.TSS2 and pre_fault == 3 and
+                          status['INVALID_COUNTER'] < 2659)
+    if status['RADAR_STATUS'] != 1 or pre_fault != 0 and not legacy_pre_warning:
       ret.errors.radarUnavailableTemporary = True
 
     for ii in sorted(updated_messages):

opendbc_repo/opendbc/safety/modes/hyundai_canfd.h

@@ -659,7 +659,9 @@ static void hyundai_canfd_rx_hook(const CANPacket_t *msg) {
           aol_optional_release_neutral = false;
           aol_optional_release_previous = 0U;
           aol_optional_release_pending = false;
-          aol_optional_release_latch = false;
+          if ((elapsed == 0U) || (elapsed > 100000U)) {
+            aol_optional_release_latch = false;
+          }
         }
         const bool raw_neutral = fresh && !main_button &&
           !(hyundai_canfd_alt_buttons ? GET_BIT(msg, 39U) : GET_BIT(msg, 23U)) && (cruise_button == HYUNDAI_BTN_NONE);

opendbc_repo/opendbc/safety/tests/test_hyundai_optional_release.py

@@ -68,9 +68,9 @@ class TestHyundaiOptionalRelease(unittest.TestCase):
     self.buttons(button)
     self.buttons()
 
-  def claimed(self):
+  def claimed(self, button=2):
     self.safety.aol_set_host_request(5)
-    self.release_gesture()
+    self.release_gesture(button)
     self.assert_rx_profile()
     self.safety.aol_set_host_request(7)
     self.buttons(4)
@@ -200,3 +200,35 @@ class TestHyundaiOptionalRelease(unittest.TestCase):
     self.assertEqual(self.safety.aol_get_request_mask(), 0, self.context)
     self.assertEqual(self.safety.aol_get_permission_mask(), 0, self.context)
     self.assertFalse(self.tx(1, 1), self.context)
+
+  def test_claimed_lateral_survives_timely_button_counter_interruption(self):
+    for raw in (0x815, 0x895):
+      for button in (1, 2):
+        for defect in ("replay", "gap"):
+          with self.subTest(raw=raw, button=button, defect=defect):
+            self.mode(raw)
+            self.claimed(button)
+            self.now += 20_000
+            self.safety.set_timer(self.now)
+            self.healthy()
+            counter = self.counters["CRUISE_BUTTONS"] % 16
+            self.buttons(counter=counter if defect == "replay" else (counter + 2) % 16)
+            self.assert_rx_profile()
+            self.assertEqual(self.safety.aol_get_permission_mask(), 0 if self.release else 1, self.context)
+            self.assertEqual(bool(self.tx(1, 1)), not self.release, self.context)
+            self.buttons(counter=(counter + 2) % 16)
+            self.safety.aol_set_host_request(5)
+            self.assertFalse(self.safety.get_controls_allowed(), self.context)
+            self.assertEqual(self.safety.aol_get_permission_mask(), 0 if self.release else 1, self.context)
+            self.safety.aol_set_host_request(0)
+            self.safety.aol_set_host_request(5)
+            self.assert_denied()
+
+  def test_claimed_lateral_still_requires_bounded_button_source_time(self):
+    for raw in (0x815, 0x895):
+      for elapsed in (0, 100_001):
+        with self.subTest(raw=raw, elapsed=elapsed):
+          self.mode(raw)
+          self.claimed()
+          self.buttons(counter=self.counters["CRUISE_BUTTONS"] % 16, advance=elapsed)
+          self.assert_denied()

opendbc_repo/opendbc/safety/tests/test_subaru_receive_guards.py

@@ -1,5 +1,6 @@
 """Malformed and contradictory physical sources withdraw receiver credits."""
 import unittest
+from opendbc.car.structs import CarParams
 from opendbc.safety.tests import test_subaru_software_cruise as software
 from opendbc.safety.tests import test_subaru_sng as sng
 from opendbc.safety.tests import test_subaru_stock_aol as stock
@@ -40,12 +41,36 @@ class TestSubaruReceiveGuards(unittest.TestCase):
 
   def test_stock_aol_truncated_optional_gear_withdraws(self):
     case = stock.TestSubaruStockAol()
+    self.addCleanup(case.tearDown)
     case.configure()
     case.arm()
     gear = case.frame('Transmission', {'Gear': 121})
     case.safety.safety_rx_hook(libsafety_py.make_CANPacket(gear.addr, gear.bus, bytes(gear.data[0:7])))
     self.assertEqual(case.safety.aol_get_permission_mask(), 0)
 
+  def test_stock_aol_cleanup_runs_on_success_failure_and_error(self):
+    """The manually owned helper must not leave its native AOL profile for the next test."""
+    safety = libsafety_py.libsafety
+    for outcome in ('success', 'failure', 'error'):
+      with self.subTest(outcome=outcome):
+        class OwnedCase(TestSubaruReceiveGuards):
+          def runTest(self, expected_outcome=outcome):
+            self.test_stock_aol_truncated_optional_gear_withdraws()
+            if expected_outcome == 'failure':
+              self.fail('deliberate failure after the real receiver assertions')
+            if expected_outcome == 'error':
+              raise RuntimeError('deliberate error after the real receiver assertions')
+
+        result = unittest.TestResult()
+        OwnedCase().run(result)
+        self.assertEqual(result.testsRun, 1)
+        self.assertEqual(len(result.failures), int(outcome == 'failure'))
+        self.assertEqual(len(result.errors), int(outcome == 'error'))
+        self.assertEqual(safety.get_current_safety_mode(), CarParams.SafetyModel.noOutput)
+        self.assertFalse(safety.get_controls_allowed())
+        self.assertEqual(safety.aol_get_request_mask(), 0)
+        self.assertEqual(safety.aol_get_permission_mask(), 0)
+
   def test_sng_conflicting_duplicate_withdraws_source_and_forwarding_lease(self):
     case = sng.TestSubaruSng()
     case.setup()

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -1352,6 +1352,27 @@ class SelfdriveD:
       self.aol_fault_reason = 'availabilityPause'
       self.aol_publication_retained_intent = intent
       return
+    from openpilot.starpilot.car.toyota.aol import qualified as toyota_aol_qualified
+    companion = getattr(self, 'aol_intent_companion', None)
+    if (healthy_expiry and toyota_aol_qualified(self.CP, marked_only=True) and
+        not was_enabled and not self.active and not getattr(self, 'aol_ever_active', False) and
+        not any((previous.desired_lateral, previous.desired_longitudinal, previous.lateral_active,
+                 previous.longitudinal_active, previous.native_acknowledged)) and
+        not self.aol_authority_lost and getattr(self, 'aol_fault_reason', 'none') == 'availabilityPause' and
+        availability_only(self.events.to_msg()) and not any(
+          event.userDisable and event.name != EventName.pcmDisable for event in self.events.to_msg()) and
+        intent is not None and intent is self.aol_last_intent and
+        intent.settingsQualified and intent.allowedLatch and intent.lateralArmed and
+        intent.carStateLogMonoTime == source_ns and
+        0 < intent.observedMonoTime <= now_ns <= intent.validUntilMonoTime and
+        companion is not None and companion.last_drain_ok and companion.session == intent.producerSessionId):
+      from openpilot.starpilot.aol.transport_pause import _healthy_transport_native
+      native = _healthy_transport_native(self.sm, self.CP, CS, self.aol_session_id, now_ns)
+      if native is not None and native.pandaInventory and not any((native.requestedLateral,
+          native.lateralAllowed, native.requestedLongitudinal, native.longitudinalAllowed)):
+        # Cold readiness has no granted command to lose; the expired source still withdraws all output.
+        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

openpilot/selfdrive/test/longitudinal_maneuvers/plant.py

@@ -98,6 +98,7 @@ class Plant:
     # always start at current speed
     model.modelV2.velocity.x = [float(self.speed)] + [float(self.speed + 0.5)] * (len(T_IDXS) - 1)
     model.modelV2.acceleration.x = [0.0] * len(T_IDXS)
+    model.modelV2.meta.disengagePredictions.gasPressProbs = [float(prob_throttle)] * 6
 
     control.controlsState.longControlState = LongCtrlState.pid if self.enabled else LongCtrlState.off
     ss.selfdriveState.experimentalMode = self.e2e

openpilot/selfdrive/ui/ui.py

@@ -34,10 +34,18 @@ def update_frame(preview=None, controllers=None, star=None, game_mode=None, arab
 
 
 def main():
-  camera_core_ui = COMMA_HARDWARE and HARDWARE.get_device_type() in ("mici", "tici", "tizi")
-  cores = {6} if camera_core_ui else {5}
+  device_type = HARDWARE.get_device_type() if COMMA_HARDWARE else ""
+  camera_core_ui = device_type in ("mici", "tici", "tizi")
+  background_ui = device_type == "mici"
+  cores = {0, 1, 2, 7} if background_ui else {6} if camera_core_ui else {5}
+  display_nice = 5 if background_ui else 0
   main_thread_configured = False
-  config_realtime_process(0, Priority.UI)
+  if background_ui:
+    drop_realtime()
+    os.setpriority(os.PRIO_PROCESS, 0, display_nice)
+    set_core_affinity(list(cores))
+  else:
+    config_realtime_process(0, Priority.UI)
 
   selection = select_ui(Profile.LARGE if BIG_UI else Profile.COMPACT, os.environ)
   print(f"UI: {'StarPilot' if selection.custom else 'upstream'} selected ({selection.reason})", file=sys.stderr, flush=True)
@@ -76,11 +84,10 @@ def main():
     for should_render, frame_time, cpu_time in gui_app.render(before_frame=lambda: update_frame(preview, controllers, star, poll_game_mode, poll_arabesque)):
       if should_render:
         if camera_core_ui and not main_thread_configured:
-          # Share the camera core with normal scheduling, without FIFO display
-          # work preempting camera work or waiting on the driving core.
+          # Display work stays below real-time driving and audio playback.
           drop_realtime()
-          os.setpriority(os.PRIO_PROCESS, 0, 0)
-          if os.sched_getscheduler(0) != os.SCHED_OTHER or os.getpriority(os.PRIO_PROCESS, 0) != 0:
+          os.setpriority(os.PRIO_PROCESS, 0, display_nice)
+          if os.sched_getscheduler(0) != os.SCHED_OTHER or os.getpriority(os.PRIO_PROCESS, 0) != display_nice:
             raise RuntimeError("Could not configure display scheduling")
           main_thread_configured = True
         # reaffine after power save offlines our core

openpilot/starpilot/aol/intent.py

@@ -241,6 +241,10 @@ class AolCardIntent:
       return
     if self._engagement_consumed or not active and not blocked:
       return
+    if (not blocked and getattr(self, 'defer_full_engagement_on_gas', False) and CS.gasPressed and
+        CS.cruiseState.available and
+        CS.gearShifter not in (GearShifter.park, GearShifter.neutral, GearShifter.reverse, GearShifter.unknown)):
+      return
     self._engagement_consumed = True
     if (blocked or not active or not self.settings.enabled or not CS.cruiseState.available or
         CS.gearShifter in (GearShifter.park, GearShifter.neutral, GearShifter.reverse, GearShifter.unknown)):

openpilot/starpilot/aol/panda_faults.py

@@ -21,15 +21,6 @@ def _diagnostic_mask(cp):
                          int(cp.flags) & int(HondaFlags.BOSCH) and qualified(cp, marked_only=True)) else 0
     if cp.brand == 'hyundai':
       from opendbc.car.hyundai.values import CAR, HyundaiFlags
-      if cp.carFingerprint == CAR.HYUNDAI_ELANTRA_2021:
-        from opendbc.car.hyundai.classic_scc_aol import qualified
-        # Firmware latches this IRQ diagnostic without revoking native authority.
-        # Limit compatibility to the recorded stock-SCC profile; keep raw faults;
-        # native permission, receive health, relay and heartbeat remain required.
-        recorded_flags = HyundaiFlags.SEND_LFA | HyundaiFlags.USE_FCA | HyundaiFlags.CHECKSUM_CRC8 | HyundaiFlags.HAS_LDA_BUTTON
-        return (1 << 2) if (str(cp.fingerprintSource) == 'fw' and int(cp.flags) == int(recorded_flags) and
-                           len(cp.safetyConfigs) == 1 and int(cp.safetyConfigs[0].safetyParam) == 0xC00 and
-                           cp.alternativeExperience == 32 and qualified(cp, marked_only=True)) else 0
       if cp.carFingerprint != CAR.HYUNDAI_SONATA_HYBRID:
         return 0
       from opendbc.car.hyundai.classic_scc_aol import qualified

openpilot/starpilot/aol/tests/test_elantra_irq_admission.py

@@ -1,4 +1,4 @@
-"""Exact stock-SCC Elantra IRQ diagnostics retain native authority vetoes."""
+"""Elantra CAN interrupt faults block steering regardless of otherwise valid authority."""
 from contextlib import ExitStack
 from dataclasses import replace
 from unittest import mock
@@ -32,7 +32,8 @@ def recorded_cp():
 @pytest.mark.parametrize('fault_source', ['none', 'panda', 'native', 'both'])
 @pytest.mark.parametrize('full', [False, True])
 @pytest.mark.parametrize('fault', ['none', 'rx', 'relay', 'denied', 'request', 'expired-wire'])
-def test_recorded_irq_through_selfdrived_and_controls_keeps_native_vetoes(full, fault, fault_source):
+def test_recorded_irq_through_selfdrived_and_controls_blocks_faulted_steering(full, fault, fault_source):
+  irq = fault_source != 'none'
   cp = recorded_cp()
   before = cp.to_dict()
   case = fixture.IoniqAvailabilityTests(methodName='runTest')
@@ -69,16 +70,20 @@ def test_recorded_irq_through_selfdrived_and_controls_keeps_native_vetoes(full,
       if fault == 'request':
         options.update(native_request_lat=False, native_request_long=False)
       axis, state, command = drive(tick, **options)
-    if fault == 'none':
+    if fault == 'none' and not irq:
       assert axis.lateralActive and command.latActive, 'HEALTHY_ELANTRA_AUTHORITY_REJECTED'
       assert log.OnroadEvent.EventName.controlsMismatch not in sd.events.names
       native = current_native(sd.sm, sd.CP, now_ns=stamp, axis_session_id=sd.aol_session_id)
-      assert native is not None and native.pandaInventory[0].faults == (4 if fault_source in ('native', 'both') else 0)
-      assert list(sd.sm['pandaStates'][0].faults) == ([log.PandaState.FaultType.interruptRateCan1] if fault_source in ('panda', 'both') else [])
+      assert native is not None and native.pandaInventory[0].faults == 0
+      assert list(sd.sm['pandaStates'][0].faults) == []
       assert not axis.longitudinalActive and not command.longActive
       assert state.enabled == full
     else:
       assert not axis.lateralActive and not command.latActive, 'ELANTRA_HARDWARE_FAULT_MUST_BLOCK'
+      if irq and fault == 'none':
+        assert log.OnroadEvent.EventName.controlsMismatch in sd.events.names
+        expected_faults = [log.PandaState.FaultType.interruptRateCan1] if fault_source in ('panda', 'both') else []
+        assert list(sd.sm['pandaStates'][0].faults) == expected_faults
       assert not axis.longitudinalActive and not command.longActive
     assert cp.to_dict() == before
 
@@ -86,7 +91,7 @@ def test_recorded_irq_through_selfdrived_and_controls_keeps_native_vetoes(full,
 @pytest.mark.parametrize('name,value', tuple(log.PandaState.FaultType.schema.enumerants.items()))
 def test_other_and_mixed_faults_still_block(name, value):
   cp = recorded_cp()
-  expected = name != 'interruptRateCan1'
+  expected = True
   assert blocking_fault_mask(1 << value, cp) == expected
   assert blocking_fault_mask((1 << value) | 4, cp) == expected
   assert blocking_fault_list([value], cp) == expected

openpilot/starpilot/aol/tests/test_ev6_cancel_retention.py

@@ -100,3 +100,48 @@ def test_ev6_card_cancel_join_requires_its_dedicated_physical_owner(restriction)
   owner._last_latch_edge_ns = now - 10_000_000
   assert retained_shared_long_cancel_intent(cp, cs, sm, feedback, owner,
     cancel_source_ns=now, now_ns=now, producer='authority-card') == (restriction == 'healthy')
+
+
+@pytest.mark.parametrize('ev6', (False, True))
+@pytest.mark.parametrize('restriction', ('healthy', 'no_source', 'old_source', 'lat_denied', 'rx',
+                                       'heartbeat', 'same_health', 'wrong_session'))
+def test_cancel_can_join_new_raw_long_off_health_on_its_first_physical_sample(ev6, restriction):
+  from openpilot.starpilot.aol.wire import encode_safety
+
+  cp, cs, sm, feedback, _, native, now = captured_context()
+  if ev6:
+    cp = ev6_params()
+    cp.safetyConfigs[0].safetyParam |= 0x800
+    cp.alternativeExperience = 32
+    feedback.cp = cp_identity(cp)
+  slot = replace(native.pandaInventory[0], controlsAllowed=True)
+  native = replace(native, longitudinalAllowed=True, pandaInventory=(slot,),
+                   sourcePandaStatesMonoTime=now - 10_000_000)
+  source = now
+  if restriction == 'no_source':
+    source = 0
+  elif restriction == 'old_source':
+    source = now - 30_000_001
+  elif restriction == 'lat_denied':
+    native = replace(native, lateralAllowed=False)
+  elif restriction == 'same_health':
+    native = replace(native, sourcePandaStatesMonoTime=now)
+  elif restriction == 'wrong_session':
+    native = replace(native, axisSessionId='foreign')
+  from openpilot.cereal import messaging
+  pandas = messaging.new_message('pandaStates', 1)
+  pandas.pandaStates[0] = sm['pandaStates'][0]
+  pandas.pandaStates[0].safetyRxChecksInvalid = restriction == 'rx'
+  pandas.pandaStates[0].heartbeatLost = restriction == 'heartbeat'
+  sm['pandaStates'] = pandas.as_reader().pandaStates
+  sm['aolSafetyWire'] = encode_safety(native)
+  assert feedback.longitudinal_cancel_ns == 0
+  cleared = feedback.observe(sm, cp, cs, now, latched=True, cancel_source_ns=source,
+                             intent_producer_session='authority-card')
+  assert cleared == (restriction != 'healthy')
+  if restriction == 'healthy':
+    assert feedback.longitudinal_cancel_ns == now
+    cs.buttonEvents = []
+    assert not feedback.observe(sm, cp, cs, now + 10_000_000, latched=True,
+                                intent_producer_session='authority-card')
+    assert feedback.longitudinal_cancel_ns == now

openpilot/starpilot/aol/tests/test_full_engagement.py

@@ -194,3 +194,49 @@ def test_auxiliary_off_after_pending_observation_cannot_be_undone_by_activation(
   assert not owner.allowed_latch
   observe(owner, cs, 3)
   assert not owner.allowed_latch
+
+
+@pytest.mark.parametrize('factory', (ClassicLongCardIntent, ClassicSccCardIntent))
+@pytest.mark.parametrize('main', (0, AOL_TOGGLE))
+@pytest.mark.parametrize('lkas', (0, AOL_TOGGLE))
+def test_classic_hkg_gas_override_defers_arming_until_release(factory, main, lkas):
+  owner, cs = factory(CP, replace(SETTINGS, main_action=main, lkas_action=lkas)), state()
+  observe(owner, cs, 1, enabled=False, active=False)
+  cs.gasPressed = True
+  observe(owner, cs, 2)
+  observe(owner, cs, 3)
+  assert not owner.allowed_latch
+  cs.gasPressed = False
+  observe(owner, cs, 4)
+  assert owner.allowed_latch and owner._last_latch_edge_ns == 4
+  cs.brakePressed = True
+  owner.update(cs, now_ns=5, fault_active=False)
+  observe(owner, cs, 5, enabled=False, active=False)
+  assert owner.allowed_latch and owner.output(cs)[0]
+
+
+@pytest.mark.parametrize('factory', (ClassicLongCardIntent, ClassicSccCardIntent))
+@pytest.mark.parametrize('veto', ('cancel', 'lkas', 'fault', 'rejection', 'invalid', 'main_off', 'gear', 'disabled', 'calibration'))
+def test_classic_hkg_pending_gas_engagement_cannot_replay_after_withdrawal(factory, veto):
+  owner, cs = factory(CP, SETTINGS), state()
+  observe(owner, cs, 1, enabled=False, active=False)
+  cs.gasPressed = True
+  observe(owner, cs, 2)
+  if veto in ('cancel', 'lkas'):
+    cs.buttonEvents = [car.CarState.ButtonEvent(type=veto, pressed=True)]
+  elif veto == 'invalid':
+    cs.canValid = False
+  elif veto == 'main_off':
+    cs.cruiseState.available = False
+  elif veto == 'gear':
+    cs.gearShifter = 'park'
+  elif veto == 'disabled':
+    owner.settings = replace(SETTINGS, enabled=False)
+  elif veto == 'calibration':
+    owner._calibration_rearm_required = True
+  observe(owner, cs, 3, fault=veto == 'fault', rejection=3 if veto == 'rejection' else 0)
+  owner.settings, owner._calibration_rearm_required = SETTINGS, False
+  cs.gasPressed, cs.canValid = False, True
+  cs.gearShifter, cs.cruiseState.available, cs.buttonEvents = 'drive', True, []
+  observe(owner, cs, 4)
+  assert not owner.allowed_latch

openpilot/starpilot/aol/tests/test_hkg_shared_cancel_guard.py

@@ -256,3 +256,54 @@ def test_simultaneous_cancel_keeps_default_timestamp_side_effects(opted_in, mapp
   owner.update(cs, now_ns=2, fault_active=False)
   assert not owner.allowed_latch
   assert owner._last_latch_edge_ns == (0 if opted_in and mapped else 2)
+
+
+@pytest.mark.parametrize('identity', (CAR.GENESIS_GV60_EV_1ST_GEN, CAR.HYUNDAI_IONIQ_5,
+                                     CAR.HYUNDAI_KONA_EV_2ND_GEN))
+@pytest.mark.parametrize('fault', ('none', 'no_cancel', 'lat_denied', 'rx', 'heartbeat', 'expired', 'same_health'))
+def test_shared_long_cancel_health_can_precede_native_status(identity, fault):
+  from opendbc.car import gen_empty_fingerprint
+  from opendbc.car.hyundai.canfd_stock_aol import qualified_long
+  from openpilot.starpilot.aol.transport_pause import cp_identity
+
+  cp, cs, sm, feedback, owner, native, now = captured_context()
+  if identity == CAR.GENESIS_GV60_EV_1ST_GEN:
+    fingerprint = gen_empty_fingerprint()
+    fingerprint[0][0x1cf] = 8
+    cp = CarInterface.get_params(identity, fingerprint, [], True, False, False)
+    cp.safetyConfigs[0].safetyParam |= 0x800
+    cp.alternativeExperience = 32
+  else:
+    cp.carFingerprint = identity
+  assert qualified_long(cp, marked_only=True)
+  feedback.cp = cp_identity(cp)
+  config = cp.safetyConfigs[0]
+  slot = replace(native.pandaInventory[0], safetyModel=int(config.safetyModel.raw), safetyParam=config.safetyParam,
+                 controlsAllowed=True)
+  native = replace(native, safetyModel=int(config.safetyModel.raw), safetyParam=config.safetyParam,
+                   longitudinalAllowed=True, pandaInventory=(slot,), sourcePandaStatesMonoTime=now)
+  sm['aolSafetyWire'] = encode_safety(native)
+  pandas = messaging.new_message('pandaStates', 1, valid=True, logMonoTime=now)
+  panda = pandas.pandaStates[0]
+  panda.safetyModel, panda.safetyParam, panda.alternativeExperience = config.safetyModel, config.safetyParam, 32
+  panda.controlsAllowed = True
+  sm['pandaStates'] = pandas.as_reader().pandaStates
+  if fault == 'no_cancel':
+    cs.buttonEvents = []
+  assert not feedback.observe(sm, cp, cs, now, latched=True, cancel_source_ns=now,
+                              intent_producer_session='authority-card')
+  assert bool(feedback.longitudinal_cancel_ns) == (fault != 'no_cancel')
+  original_receipt = feedback.longitudinal_cancel_ns
+  cs.buttonEvents = []
+  pandas.pandaStates[0].controlsAllowed = False
+  pandas.pandaStates[0].safetyRxChecksInvalid = fault == 'rx'
+  pandas.pandaStates[0].heartbeatLost = fault == 'heartbeat'
+  sm['pandaStates'] = pandas.as_reader().pandaStates
+  sm.logMonoTime['pandaStates'] = now + (0 if fault == 'same_health' else 10_000_000)
+  if fault == 'lat_denied':
+    sm['aolSafetyWire'] = encode_safety(replace(native, lateralAllowed=False))
+  later = now + (201_000_000 if fault == 'expired' else 10_000_000)
+  retained = feedback._longitudinal_cancel_current(sm, cp, cs, sm['aolAxisState'], later,
+                                                  latched=True, cancel_source_ns=0)
+  assert (retained is not None) == (fault == 'none')
+  assert feedback.longitudinal_cancel_ns == original_receipt

openpilot/starpilot/aol/tests/test_toyota_cold_source_expiry.py

@@ -0,0 +1,114 @@
+"""Cold Toyota readiness is not a previously granted steering command."""
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pytest
+from openpilot.cereal import log, messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.transport_pause import TransportPauseFeedback
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_toyota_moving_pcm_off_deadline import cp_for, moving
+from openpilot.starpilot.aol.tests.test_toyota_parking_brake_deadline import refresh_saved_settings
+
+
+BASE = 10_000_000_000
+
+
+@pytest.mark.parametrize('profile', ('prius_long', 'prius_stock', 'camera_long', 'camera_stock'))
+@pytest.mark.parametrize('gas', (False, True))
+def test_cold_source_expiry_waits_without_inventing_authority(profile, gas):
+  with ExitStack() as cleanup:
+    cleanup.enter_context(OpenpilotPrefix())
+    cleanup.enter_context(patch.dict('os.environ', {'SIMULATION': '0', 'REPLAY': '0'}))
+    case = fixture.IoniqAvailabilityTests(methodName='runTest')
+    cleanup.callback(case.doCleanups)
+    sd, _, drive = case._authority_callers(False, cp=cp_for(profile), genuine_events=True)
+    refresh_saved_settings(sd)
+    feedback = TransportPauseFeedback(sd.CP)
+    feedback_sm = messaging.SubMaster(['aolAxisState', 'onroadEvents', 'aolSafetyWire', 'pandaStates'])
+
+    def card_feedback(now):
+      samples = [sd.pm.messages[name] for name in ('aolAxisState', 'onroadEvents')]
+      for name in ('aolSafetyWire', 'pandaStates'):
+        sample = messaging.new_message(name, len(sd.sm[name]), valid=sd.sm.valid[name], logMonoTime=sd.sm.logMonoTime[name])
+        setattr(sample, name, sd.sm[name])
+        samples.append(sample.as_reader())
+      feedback_sm.update_msgs(now / 1e9, samples)
+      assert not feedback.observe(feedback_sm, sd.CP, moving(), now, latched=True,
+                                  intent_producer_session='authority-card')
+
+    update_events, publish = sd.update_events, sd.publish_axis_proof
+
+    def startup_events(cs):
+      sd.sm.valid['longitudinalPlan'] = False
+      return update_events(cs)
+
+    def delayed_publication(cs, **kwargs):
+      with patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', return_value=BASE + 31_148_802):
+        return publish(cs, **kwargs)
+
+    with patch.object(sd, 'update_events', side_effect=startup_events), \
+         patch.object(sd, 'publish_axis_proof', side_effect=delayed_publication):
+      axis, host, command = drive(0, observed_state=moving(), gas=gas, allowed=False,
+                                  native_request_lat=False, native_request_long=False)
+    assert log.OnroadEvent.EventName.commIssue in sd.events.names
+    assert log.OnroadEvent.EventName.controlsMismatch not in sd.events.names, 'TOYOTA_COLD_SOURCE_EXPIRY'
+    assert axis.sourceCarStateMonoTime == 0 and str(axis.faultReason) == 'availabilityPause'
+    assert not axis.nativeAcknowledged and not axis.desiredLateral and not axis.desiredLongitudinal
+    assert not axis.lateralActive and not axis.longitudinalActive and not host.enabled
+    assert not command.latActive and not command.longActive and not sd.aol_authority_lost
+    assert sd.aol_publication_retained_intent.validUntilMonoTime == BASE + 200_000_000
+    card_feedback(BASE + 31_148_802)
+    # A fresh source is required; requesting LAT alone still cannot produce it.
+    axis, host, command = drive(4, observed_state=moving(), gas=gas, allowed=False,
+                                native_request_lat=False, native_request_long=False)
+    assert axis.sourceCarStateMonoTime == BASE + 40_000_000
+    assert not axis.lateralActive and not command.latActive and not host.enabled
+    card_feedback(BASE + 40_000_000)
+    axis, host, command = drive(5, observed_state=moving(), gas=gas, allowed=True,
+                                native_request_lat=True, native_request_long=False)
+    assert axis.lateralActive and command.latActive and not host.enabled and not command.longActive
+    assert log.OnroadEvent.EventName.controlsMismatch not in sd.events.names
+    card_feedback(BASE + 50_000_000)
+    assert feedback.ever_lateral_active
+
+
+@pytest.mark.parametrize('restriction', ('previously_active', 'native_requested', 'native_granted', 'native_invalid',
+                                        'rx_fault', 'relay_fault', 'expired_source', 'can_invalid', 'critical'))
+def test_cold_wait_never_waives_authority_or_health_failures(restriction):
+  with ExitStack() as cleanup:
+    cleanup.enter_context(OpenpilotPrefix())
+    cleanup.enter_context(patch.dict('os.environ', {'SIMULATION': '0', 'REPLAY': '0'}))
+    case = fixture.IoniqAvailabilityTests(methodName='runTest')
+    cleanup.callback(case.doCleanups)
+    sd, _, drive = case._authority_callers(False, cp=cp_for('prius_long'), genuine_events=True)
+    refresh_saved_settings(sd)
+    if restriction == 'previously_active':
+      axis, _, command = drive(0, observed_state=moving(), native_request_long=False)
+      assert axis.lateralActive and command.latActive
+    original_events, publish = sd.update_events, sd.publish_axis_proof
+
+    def startup_events(cs):
+      sd.sm.valid['longitudinalPlan'] = False
+      return original_events(cs)
+
+    def delayed_publication(cs, **kwargs):
+      delay = 201_000_000 if restriction == 'expired_source' else 31_148_802
+      with patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', return_value=BASE + 10_000_000 + delay):
+        return publish(cs, **kwargs)
+
+    state = moving()
+    if restriction == 'can_invalid':
+      state.canValid = False
+    options = {'allowed': restriction == 'native_granted', 'native_request_lat': restriction == 'native_requested',
+               'native_request_long': False, 'native_valid': restriction != 'native_invalid',
+               'panda_fault': restriction == 'rx_fault', 'relay_fault': restriction == 'relay_fault'}
+    if restriction == 'critical':
+      options['critical_event'] = log.OnroadEvent.EventName.canError
+    with patch.object(sd, 'update_events', side_effect=startup_events), \
+         patch.object(sd, 'publish_axis_proof', side_effect=delayed_publication):
+      axis, host, command = drive(1, observed_state=state, **options)
+    assert log.OnroadEvent.EventName.controlsMismatch in sd.events.names
+    assert not axis.lateralActive and not axis.longitudinalActive
+    assert not host.enabled and not command.latActive and not command.longActive
+    assert str(axis.faultReason) == 'critical'

openpilot/starpilot/aol/transport_pause.py

@@ -805,7 +805,7 @@ class TransportPauseFeedback:
     return native
 
   def _longitudinal_cancel_current(self, sm, cp, cs, axis, now_ns, *, latched, cancel_source_ns, withdrawal_required=True) -> SafetyState | None:
-    from opendbc.car.hyundai.ev6_aol import qualified
+    from opendbc.car.hyundai.canfd_stock_aol import qualified_long as qualified
     from openpilot.starpilot.car.hyundai.aol import qualified_ccnc_long_aol as ev9_qualified
     from opendbc.car.structs import car
 
@@ -833,7 +833,7 @@ class TransportPauseFeedback:
       return None
     native = current_native(sm, cp, now_ns=now_ns, axis_session_id=self.session)
     if (native is None or not native.pandaInventory or not native.requestedLateral or not native.lateralAllowed or
-        not native.requestedLongitudinal or withdrawal_required and native.longitudinalAllowed or
+        not native.requestedLongitudinal or
         native.observedMonoTime < self.lateral_native_ns or
         not native_inventory_matches_cp(native, cp) or
         (native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) != self.lateral_inventory):
@@ -849,6 +849,12 @@ class TransportPauseFeedback:
       not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and
       (str(c.safetyModel) not in ('silent', 'noOutput') or not p.controlsAllowed)
       for p, c in zip(pandas, cp.safetyConfigs, strict=True))
+    raw_withdrawal = bool((self.longitudinal_cancel_ns or fresh_cancel) and
+      0 < native.sourcePandaStatesMonoTime < stamp and
+      all(not p.controlsAllowed for p, c in zip(pandas, cp.safetyConfigs, strict=True)
+          if str(c.safetyModel) not in ('silent', 'noOutput')))
+    if withdrawal_required and native.longitudinalAllowed and not raw_withdrawal:
+      return None
     return native if current_pandas else None
 
 

openpilot/starpilot/car/hyundai/classic_long_intent.py

@@ -3,6 +3,8 @@ from openpilot.starpilot.aol.intent import AolCardIntent
 
 
 class ClassicLongCardIntent(AolCardIntent):
+  defer_full_engagement_on_gas = True
+
   def __init__(self, cp, settings):
     super().__init__(settings, explicit_latch=True, independent_cruise_buttons=(
       independent_buttons_qualified(cp, marked_only=False)))

openpilot/starpilot/car/hyundai/classic_scc_intent.py

@@ -4,6 +4,8 @@ from openpilot.starpilot.car.hyundai.forte_intent import ForteCardIntent
 
 
 class ClassicSccCardIntent(ForteCardIntent):
+  defer_full_engagement_on_gas = True
+
   observe_stock_engagement = True
 
   def __init__(self, cp, settings):

openpilot/starpilot/car/tesla/tests/fixtures/custom_reviewed_schema_additions.json

@@ -1,6 +1,6 @@
 {
-  "reviewed_schema_commit": "cd0f46b9d65a9c5e47002f57e5ad6625c9757c63",
-  "reviewed_schema_sha256": "5effa2eb1fbbada3d1bbaae0acd3d31d780d8ec0e6ee1e214fedbc38ae400182",
+  "reviewed_schema_commit": "63beafc6af294d0759b47dc3bc546f24082a2a5d",
+  "reviewed_schema_sha256": "ca01ef4f6d2b15c5fac59ce237ec6e49f1167e2408903c9088ee322ca490de31",
   "incumbent_fixture_sha256": "a36ab54c00b5bbe61d520873d24aadfb8f33babd70c31a07046516d5545838d9",
   "types": {
     "0xf35cc4560bbf6ec2": {
@@ -18,7 +18,7 @@
       },
       "new_storage": {
         "dataWordCount": 3,
-        "pointerCount": 6,
+        "pointerCount": 7,
         "preferredListEncoding": "inlineComposite",
         "isGroup": false,
         "discriminantCount": 0,
@@ -59,11 +59,31 @@
             "hadExplicitDefault": false
           },
           "name": "toyotaCruise",
-          "codeOrder": 38,
+          "codeOrder": 39,
           "discriminantValue": 65535,
           "ordinal": {
             "explicit": 38
           }
+        },
+        {
+          "slot": {
+            "offset": 6,
+            "type": {
+              "struct": {
+                "typeId": 15172299076723568810
+              }
+            },
+            "defaultValue": {
+              "schemaValueHex": "00000000040000000000000002000100100000000000000000000000000000000000000000000000"
+            },
+            "hadExplicitDefault": false
+          },
+          "name": "steeringLimitInfo",
+          "codeOrder": 38,
+          "discriminantValue": 65535,
+          "ordinal": {
+            "explicit": 39
+          }
         }
       ]
     },
@@ -656,7 +676,8 @@
   },
   "published_addition_commits": {
     "slc": "ab8368f287b78b42f8590722878c7d5834609e19",
-    "flm": "b151ca11cb4d88f32d7adfdbcff50baf29f4fbbc"
+    "flm": "b151ca11cb4d88f32d7adfdbcff50baf29f4fbbc",
+    "tesla_cooperative_steering": "63beafc6af294d0759b47dc3bc546f24082a2a5d"
   },
-  "review_basis": "Published SLC and FLM append-only changes. Every incumbent compiled field descriptor and enum prefix was independently checked unchanged; only the exact seven typed deltas below are permitted."
+  "review_basis": "Published SLC, FLM, and Tesla cooperative-steering additions. Every incumbent wire field descriptor and enum prefix was independently checked unchanged; only the exact typed deltas listed here are permitted. Tesla adds steeringLimitInfo @39 at pointer slot 6 using the existing SteeringLimitInfo type. ToyotaCruise @38 retains pointer slot 5, type, and default; only its declaration codeOrder changes from 38 to 39."
 }

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

@@ -115,6 +115,47 @@ def test_additive_feedback_survives_incumbent_reader_relay(historical):
     assert recovered.feedbackPublication.to_dict() == expected_feedback
 
 
+def test_additive_steering_limits_preserve_defaults_and_survive_incumbent_relay(historical):
+  original = historical.StarPilotCarState.new_message(
+    pedalMaxRegen=True, pedalLongActive=True, sourceCarStateMonoTime=123456,
+    gps={'latitude': 41.25, 'longitude': -87.5, 'hasFix': True})
+  incumbent = original.to_dict()
+  with custom.StarPilotCarState.from_bytes(original.to_bytes()) as decoded:
+    assert not decoded.steeringLimitInfo.valid
+    assert decoded.steeringLimitInfo.monoTime == 0
+    current = decoded.as_builder()
+  current.feedbackPublication = {'version': 1, 'publicationSequence': 27, 'reportedTorque': 0.25}
+  current.toyotaCruise = {'version': 1, 'status': 'active', 'targetKph': 72., 'sessionId': 'incumbent'}
+  current.steeringLimitInfo = {
+    'valid': True, 'modelLimitErrorDeg': 1.25, 'resumeLimitErrorDeg': 2.5,
+    'cooperativeLimitErrorDeg': 3.75, 'cooperativeOffsetDeg': -0.5,
+    'monoTime': 123456789, 'combinedLimitErrorDeg': 4.0}
+  expected = current.to_dict()
+  with historical.StarPilotCarState.from_bytes(current.to_bytes()) as old:
+    assert old.to_dict() == incumbent
+    relayed = old.as_builder().to_bytes()
+  with custom.StarPilotCarState.from_bytes(relayed) as recovered:
+    assert recovered.to_dict() == expected
+
+
+@pytest.mark.parametrize(('before', 'after'), [
+  ('dashboardSpeedLimit @3 :Float32;', 'dashboardSpeedLimit @3 :Float64;'),
+  ('toyotaCruise @38 :ToyotaCruise;', 'toyotaCruise @38 :StarPilotCarControl.SteeringLimitInfo;'),
+  ('steeringLimitInfo @39 :StarPilotCarControl.SteeringLimitInfo;', 'steeringLimitInfo @39 :StarPilotCarState.Gps;'),
+  ('steeringLimitInfo @39 :StarPilotCarControl.SteeringLimitInfo;',
+   'steeringLimitInfo @39 :StarPilotCarControl.SteeringLimitInfo; unreviewed @40 :Bool;'),
+])
+def test_reviewed_steering_addition_does_not_allow_other_schema_changes(historical, tmp_path, before, after):
+  source = Path(CEREAL_PATH, 'custom.capnp').read_text()
+  assert source.count(before) == 1
+  path = tmp_path / 'unreviewed.capnp'
+  path.write_text(source.replace(before, after))
+  parser = capnp.SchemaParser()
+  changed = parser.load(str(path), imports=[CEREAL_PATH, str(files('opendbc') / 'car')])
+  with pytest.raises(AssertionError):
+    compare_node(historical.StarPilotCarState, changed.StarPilotCarState)
+
+
 @pytest.mark.parametrize(('fields', 'compatible'), [
   ('value @0 :UInt32; label @1 :Text;', True),
   ('value @0 :UInt32; label @1 :Text; extra @2 :Bool;', False),

openpilot/starpilot/car/toyota/tests/test_legacy_radar_pre_warning.py

@@ -0,0 +1,138 @@
+"""Legacy Toyota pre-warning is distinct from dirty radar and CAN failure."""
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pytest
+from opendbc.can import CANPacker
+from opendbc.car import Bus, gen_empty_fingerprint
+from opendbc.car.toyota.interface import CarInterface
+from opendbc.car.toyota.radar_interface import RadarInterface
+from opendbc.car.toyota.values import CAR, DBC
+from openpilot.cereal import messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_toyota_moving_pcm_off_deadline import cp_for
+
+
+LEGACY = (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT, CAR.TOYOTA_RAV4H, CAR.LEXUS_RX)
+
+
+def params(candidate, alpha):
+  if candidate == CAR.TOYOTA_PRIUS_RETROFIT:
+    return cp_for('prius_long' if alpha else 'prius_stock')
+  return CarInterface.get_params(candidate, gen_empty_fingerprint(), [], alpha, False, False)
+
+
+def batch(radar, packer, tick, *, status=1, pre=3, counter=1358, target=False, omit=None):
+  frames = []
+  for address in radar.RADAR_A_MSGS + radar.RADAR_B_MSGS:
+    if address == omit:
+      continue
+    first = address == radar.RADAR_A_MSGS[0]
+    fields = ({'LONG_DIST': 30. if target and first else 255., 'VALID': int(target and first),
+               'LAT_DIST': 0., 'REL_SPEED': -1.} if address in radar.RADAR_A_MSGS else {'SCORE': 100})
+    frames.append(packer.make_can_msg(address, 1, fields))
+  if omit != 'status':
+    frames.append(packer.make_can_msg('STATUS_MSG', 1, {
+      'RADAR_STATUS': status, 'RADAR_PRE_FAULT': pre, 'INVALID_COUNTER': counter}))
+  return [(1_000_000_000 + tick * 50_000_000, frames)]
+
+
+def feed(radar, packer, **kwargs):
+  result = None
+  for tick in range(30):
+    result = radar.update(batch(radar, packer, tick, **kwargs))
+  assert result is not None
+  return result
+
+
+@pytest.mark.parametrize('candidate', LEGACY)
+@pytest.mark.parametrize('alpha', (False, True))
+@pytest.mark.parametrize('counter', (1358, 1872, 2643))
+@pytest.mark.parametrize('target', (False, True))
+def test_legacy_ready_radar_pre_warning_keeps_live_measurements(candidate, alpha, counter, target):
+  cp = params(candidate, alpha)
+  radar = RadarInterface(cp)
+  packer = CANPacker(DBC[cp.carFingerprint][Bus.radar])
+  result = feed(radar, packer, counter=counter, target=target)
+  assert radar.rcp.can_valid
+  assert not result.errors.radarUnavailableTemporary, 'LEGACY_RADAR_PRE_WARNING'
+  assert not result.errors.canError
+  assert len(result.points) == int(target)
+  if target:
+    assert result.points[0].dRel == 30. and result.points[0].vRel == -1.
+
+
+@pytest.mark.parametrize('profile', ('prius_long', 'camera_long'))
+@pytest.mark.parametrize('condition', ('initializing', 'fault', 'unknown_status', 'unknown_pre', 'dirty_threshold',
+                                       'saturated_counter', 'missing_status', 'missing_track'))
+def test_real_radar_faults_and_receive_checks_remain(profile, condition):
+  cp = cp_for(profile)
+  radar = RadarInterface(cp)
+  packer = CANPacker(DBC[cp.carFingerprint][Bus.radar])
+  options = {'target': True}
+  if condition == 'initializing':
+    options['status'] = 0
+  elif condition == 'fault':
+    options['status'] = 3
+  elif condition == 'unknown_status':
+    options['status'] = 2
+  elif condition == 'unknown_pre':
+    options['pre'] = 1
+  elif condition == 'dirty_threshold':
+    options['counter'] = 2659
+  elif condition == 'saturated_counter':
+    options['counter'] = 4095
+  elif condition == 'missing_status':
+    options['omit'] = 'status'
+  else:
+    options['omit'] = radar.RADAR_A_MSGS[0]
+  result = feed(radar, packer, **options)
+  assert result.errors.canError or result.errors.radarUnavailableTemporary
+  if condition.startswith('missing'):
+    assert result.errors.canError and not radar.rcp.can_valid
+  else:
+    assert result.errors.radarUnavailableTemporary
+
+
+def test_tss2_pre_fault_behavior_is_unchanged():
+  cp = cp_for('camera_long')
+  radar = RadarInterface(cp)
+  result = feed(radar, CANPacker(DBC[cp.carFingerprint][Bus.radar]))
+  assert result.errors.radarUnavailableTemporary
+
+
+@pytest.mark.parametrize('profile', ('prius_long', 'prius_stock'))
+def test_actual_radar_to_selfdrived_keeps_pre_warning_distinct_from_fault(profile):
+  with ExitStack() as cleanup:
+    cleanup.enter_context(OpenpilotPrefix())
+    cleanup.enter_context(patch.dict('os.environ', {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}))
+    cp = cp_for(profile)
+    case = fixture.IoniqAvailabilityTests(methodName='runTest')
+    cleanup.callback(case.doCleanups)
+    sd, _, drive = case._authority_callers(True, cp=cp, genuine_events=True)
+    radar = RadarInterface(cp)
+    packer = CANPacker(DBC[cp.carFingerprint][Bus.radar])
+    observed = fixture.car_state()
+    observed.cruiseState.available = observed.cruiseState.enabled = True
+    original = sd.update_events
+    errors = [feed(radar, packer).errors]
+
+    def events(cs):
+      message = messaging.new_message('radarState', valid=True)
+      message.radarState.radarErrors = errors[0].as_reader()
+      sd.sm.data['radarState'] = message.radarState.as_reader()
+      return original(cs)
+
+    with patch.object(sd, 'update_events', side_effect=events):
+      for tick in range(5):
+        axis, host, command = drive(tick, observed_state=observed)
+        assert host.enabled and axis.lateralActive and command.latActive
+        assert not sd.sm['radarState'].radarErrors.radarUnavailableTemporary, 'LEGACY_RADAR_PRE_WARNING'
+        if cp.openpilotLongitudinalControl:
+          assert axis.longitudinalActive and command.longActive
+      errors[0] = feed(radar, packer, status=3, counter=2659).errors
+      axis, host, command = drive(5, observed_state=observed, native_request_long=False, long_allowed=False)
+      if cp.openpilotLongitudinalControl:
+        assert not host.enabled and not axis.longitudinalActive and not command.longActive
+      assert axis.lateralActive and command.latActive

openpilot/starpilot/galaxy/setting_descriptions.py

@@ -100,7 +100,7 @@ DESCRIPTIONS = {
   'DeveloperSidebar': 'Shows live developer metrics on the right of the Big driving screen. Off gives that space back to the road view.',
   'DriverCamera': 'Shows the driver camera while reversing.',
   'StockConfidenceBallWidget': 'Uses StarPilot driving widgets in place of the stock confidence ball.',
-  'HidePersonalityAlerts': 'Hides the alert shown when the driving personality changes.',
+  'HidePersonalityAlerts': 'Hides the alert shown when the driving personality changes on the comma and Android Auto.',
   'ShowBrakeStatus': 'Colors the steering wheel to show braking and acceleration.',
   'HideLeadMarker': 'Shows the lead-vehicle indicator on the driving screen.',
   'LeadInfo': 'Shows the lead vehicle distance or speed beside its indicator.',

openpilot/starpilot/galaxy/tests/test_projection_layout.mjs

@@ -138,6 +138,8 @@ OnroadLayoutPage.methods.setProjectionLeadInfo.call(leadVm, "Both")
 assert.equal(leadVm.state.draft.leadInfo, "Speed")
 assert.equal(leadChanges.length, 2)
 assert.equal(validDocument({ ...native.document, signalBorder: true }, native.metadata), false, "the comma screen keeps its C4 settings")
+assert.match(OnroadLayoutPage.template, /Personality alerts follow the comma screen setting\./)
+assert.doesNotMatch(OnroadLayoutPage.template, /setProjectionDisplay\('hidePersonalityAlerts', value\)/)
 assert.match(OnroadLayoutPage.template, /Red Blind-Spot Border/)
 assert.match(OnroadLayoutPage.template, /Amber Signal Border/)
 const borderChanges = []

openpilot/starpilot/galaxy/tests/test_starport_send.py

@@ -119,18 +119,34 @@ class SenderJobTest(unittest.TestCase):
 
   def test_sends_in_the_background_and_drops_the_snapshot_when_too_big(self):
     bundles, sent = [], []
+    sending, release = threading.Event(), threading.Event()
 
     def bundle(system_snapshot=True):
       bundles.append(system_snapshot)
       return "logs.zip", b"x" * (DISCORD_FILE_LIMIT + 1 if system_snapshot else 10)
 
-    job = StarPortLogSender(bundle, sender=lambda *args: sent.append(args), summary=lambda: SUMMARY, url=lambda: URL)
-    self.assertEqual(job.start("hello")["state"], "sending")
+    def sender(*args):
+      sending.set()
+      if not release.wait(2):
+        raise TimeoutError("The test did not release the send")
+      sent.append(args)
+
+    job = StarPortLogSender(bundle, sender=sender, summary=lambda: SUMMARY, url=lambda: URL)
+    try:
+      self.assertEqual(job.start("hello")["state"], "sending")
+      self.assertTrue(sending.wait(2))
+      self.assertEqual(job.status()["state"], "sending")
+      self.assertEqual(sent, [])
+    finally:
+      release.set()
+      if job._thread is not None:
+        job._thread.join(2)
+        self.assertFalse(job._thread.is_alive())
     status = self.wait(job)
     self.assertEqual(status["state"], "sent")
     self.assertIn("snapshot was left out", status["message"])
     self.assertEqual(bundles, [True, False])
-    self.assertEqual(sent[0][2], "hello")
+    self.assertEqual(sent, [(b"x" * 10, "logs.zip", "hello", SUMMARY)])
 
   def test_one_send_at_a_time_and_failures_are_reported(self):
     release = threading.Event()

openpilot/starpilot/galaxy/web/js/onroad-layout-template.js

@@ -355,8 +355,7 @@ export const ONROAD_LAYOUT_TEMPLATE = `
               <div class="gx-layout__subhead"><h4>StarPort screen options</h4></div>
               <p class="gx-note">Only the StarPort screen uses these. The comma screen has its own options on the Comma tab. Save to apply.</p>
               <h5 class="gx-layout__display-group">Alerts and borders</h5>
-              <GalaxySettingRow feedback="draft" :row="switchRow('Hide Personality Alerts', state.draft.hidePersonalityAlerts === true)" :index="0"
-                :disabled="!editable || !!state.drag || !!state.layerDrag" :save-value="(index, value) => setProjectionDisplay('hidePersonalityAlerts', value)" />
+              <p class="gx-note">Personality alerts follow the comma screen setting.</p>
               <GalaxySettingRow feedback="draft" :row="switchRow('Red Blind-Spot Border', state.draft.blindSpotBorder === true)" :index="1"
                 :disabled="!editable || !!state.drag || !!state.layerDrag" :save-value="(index, value) => setProjectionDisplay('blindSpotBorder', value)" />
               <GalaxySettingRow feedback="draft" :row="switchRow('Amber Signal Border', state.draft.signalBorder === true)" :index="2"

openpilot/starpilot/lateral/hkg_vehicle_policy.py

@@ -156,7 +156,8 @@ class HKGVehicleTorquePolicy(HKGShapedTorquePolicy):
     if self.profile == 'kia_stinger_2022':
       return value * shaping.get_kia_stinger_2022_center_taper_scale(setpoint, speed)
     if self.profile == 'kia_ev6':
-      return value * shaping.get_kia_ev6_low_speed_center_taper_scale(setpoint, speed) * shaping.get_kia_ev6_center_output_scale(setpoint, speed)
+      return (value * shaping.get_kia_ev6_low_speed_center_taper_scale(setpoint, speed) *
+              shaping.get_kia_ev6_center_output_scale(setpoint, speed, self.future_lateral_accel))
     if self.profile == 'kia_carnival':
       return (
         value

openpilot/starpilot/lateral/hkg_vehicle_shaping.py

@@ -544,6 +544,8 @@ KIA_EV6_LOW_SPEED_CENTER_TAPER_SPEED_WIDTH = 1.4
 
 KIA_EV6_CENTER_OUTPUT_TAPER_MAX = 0.14
 
+KIA_EV6_HIGHWAY_CENTER_DAMPING = 0.08
+
 KIA_EV6_CENTER_OUTPUT_TAPER_LAT = 0.30
 
 KIA_EV6_CENTER_OUTPUT_TAPER_LAT_WIDTH = 0.08
@@ -1298,12 +1300,17 @@ def get_kia_ev6_low_speed_center_taper_scale(desired_lateral_accel: float, v_ego
   reduction = KIA_EV6_LOW_SPEED_CENTER_TAPER_MAX * speed_weight * center_weight
   return 1.0 - reduction
 
-def get_kia_ev6_center_output_scale(desired_lateral_accel: float, v_ego: float) -> float:
+def get_kia_ev6_center_output_scale(desired_lateral_accel: float, v_ego: float,
+                                  future_lateral_accel: float | None = None) -> float:
   speed_weight = _kia_ev6_sigmoid((v_ego - KIA_EV6_CENTER_OUTPUT_TAPER_SPEED) /
                                   KIA_EV6_CENTER_OUTPUT_TAPER_SPEED_WIDTH)
   center_weight = _kia_ev6_sigmoid((KIA_EV6_CENTER_OUTPUT_TAPER_LAT - abs(desired_lateral_accel)) /
                                    KIA_EV6_CENTER_OUTPUT_TAPER_LAT_WIDTH)
   reduction = _flm_vehicle_knob("hyundai_kia_ev6.center_output_taper_max", KIA_EV6_CENTER_OUTPUT_TAPER_MAX) * speed_weight * center_weight
+  future_accel = abs(desired_lateral_accel) if future_lateral_accel is None else max(abs(desired_lateral_accel), abs(future_lateral_accel))
+  future_center_weight = _kia_ev6_sigmoid((KIA_EV6_CENTER_OUTPUT_TAPER_LAT - future_accel) /
+                                         KIA_EV6_CENTER_OUTPUT_TAPER_LAT_WIDTH)
+  reduction += KIA_EV6_HIGHWAY_CENTER_DAMPING * speed_weight * future_center_weight
   return 1.0 - reduction
 
 

openpilot/starpilot/lateral/lane_centering.py

@@ -30,6 +30,7 @@ MAX_LANE_WIDTH = 4.8
 MAX_OFFSET = 0.3
 MIN_CENTER_TO_LINE = 1.1
 MAX_CORRECTION = 0.004 * 0.30
+MAX_CORRECTION_JERK = 1.0
 CENTER_ERROR_DEADBAND = 0.08
 SINGLE_LINE_DEADBAND_MIN = 0.04
 E2E_MAX_PATH_STD = 0.35
@@ -296,6 +297,17 @@ class LaneCenteringController:
       self._correction = 0.0
     return self._result(base, reason)
 
+  def _tracking_correction(self, base: float, correction: float, measured: float, speed: float, delay: float) -> float:
+    if self._tracker.path is None:
+      return correction
+    x, y = self._tracker.path
+    source = self._clearance_observation
+    offset = source.settings.offset_m if source is not None else 0.
+    if self._tracker.side is None and self._tracker.width is not None:
+      bound = min(MAX_OFFSET, max(0., self._tracker.width * .5 - MIN_CENTER_TO_LINE))
+      offset = clip_scalar(offset, -bound, bound)
+    return limit_tracking((x, y + offset), base, correction, measured, speed, delay)
+
   def guard_applied(self, model, base: float, candidate: float, measured: float, speed: float, delay) -> float:
     if self._body_required and (self._body is None or not self._body.valid):
       self.reset()
@@ -306,14 +318,7 @@ class LaneCenteringController:
     delta = candidate - base
     allowed = 0. if checked_delay is None else delta
     if checked_delay is not None:
-      if self._tracker.path is not None:
-        x, y = self._tracker.path
-        source = self._clearance_observation
-        offset = source.settings.offset_m if source is not None else 0.
-        if self._tracker.side is None and self._tracker.width is not None:
-          bound = min(MAX_OFFSET, max(0., self._tracker.width * .5 - MIN_CENTER_TO_LINE))
-          offset = clip_scalar(offset, -bound, bound)
-        allowed = limit_tracking((x, y + offset), base, allowed, measured, speed, checked_delay)
+      allowed = self._tracking_correction(base, allowed, measured, speed, checked_delay)
       if self._body is not None:
         allowed = limit_correction(model, base, allowed, measured, speed, checked_delay, self._body)
     if allowed != delta:
@@ -386,6 +391,12 @@ class LaneCenteringController:
     target = clip_scalar(raw, -0.004, 0.004) * 0.30
     if strength != 1.0:
       target = clip_scalar(target * strength, -MAX_CORRECTION, MAX_CORRECTION)
-    self._correction += (target - self._correction) * (1.0 - math.exp(-dt / ACQUIRE_TAU))
+    measured = _finite_number(observation.measured_curvature)
+    delay = _finite_number(observation.lateral_delay_s)
+    if measured is not None and delay is not None:
+      target = self._tracking_correction(base, target, measured, speed, delay)
+    step = (target - self._correction) * (1.0 - math.exp(-dt / ACQUIRE_TAU))
+    max_step = MAX_CORRECTION_JERK * dt / speed / speed
+    self._correction += clip_scalar(step, -max_step, max_step)
     self._correction = clip_scalar(self._correction, -MAX_CORRECTION, MAX_CORRECTION)
     return self._result(base, "qualified")

openpilot/starpilot/lateral/tests/test_hkg_vehicle_policy.py

@@ -136,3 +136,61 @@ def test_kona_2022_friction_does_not_change_sibling_policies(monkeypatch, vehicl
   controller = LatControlTorque(cp.as_reader(), interfaces[vehicle](cp), .01, controller_mode=ControllerMode.STARPILOT)
   threshold, scale = controller.starpilot_extension.policy.friction_context(.1, 0., 0., 120. / 3.6)
   assert threshold > 0. and scale > 0.
+
+
+@pytest.mark.parametrize('alpha', (False, True))
+@pytest.mark.parametrize('direction', (-1., 1.))
+def test_ev6_highway_center_damping_preserves_controller_and_curve_response(monkeypatch, alpha, direction):
+  from types import SimpleNamespace
+  import math
+  from openpilot.cereal import log
+  from opendbc.car.vehicle_model import VehicleModel
+  from openpilot.starpilot.lateral import hkg_vehicle_shaping as shaping
+
+  cp = interfaces[CAR.KIA_EV6].get_non_essential_params(CAR.KIA_EV6)
+  cp.openpilotLongitudinalControl, cp.pcmCruise = alpha, not alpha
+  cs = SimpleNamespace(vEgo=26., steeringAngleDeg=0., steeringPressed=False)
+  params = log.VehicleParameters.new_message(angleOffsetDeg=0., roll=0.)
+  selected = shaping.KIA_EV6_HIGHWAY_CENTER_DAMPING
+  assert selected == .08
+
+  def run(taper, turn, mode=ControllerMode.STARPILOT):
+    monkeypatch.setattr(shaping, 'KIA_EV6_HIGHWAY_CENTER_DAMPING', taper)
+    controller = LatControlTorque(cp.as_reader(), interfaces[cp.carFingerprint](cp), .01, controller_mode=mode)
+    vm = VehicleModel(cp)
+    outputs, stages = [], []
+    for tick in range(300):
+      acceleration = direction * (1.5 if turn else .12 * math.sin(2 * math.pi * .4 * tick * .01))
+      value, _, state = controller.update(True, cs, vm, params, False, acceleration / cs.vEgo**2, False, .386)
+      assert math.isfinite(value) and abs(value) <= controller.steer_max + 1e-6
+      outputs.append(value)
+      stages.append((state.p, state.i, state.d, state.f))
+    off, _, state = controller.update(False, cs, vm, params, False, 0., False, .386)
+    assert off == 0. and not state.active
+    return outputs, stages, controller.torque_params.to_dict(), (controller.pid.pos_limit, controller.pid.neg_limit, controller.steer_max)
+
+  actual, baseline = run(selected, False), run(0., False)
+  assert actual[1:] == baseline[1:]
+  assert sum(x*x for x in actual[0]) < .88 * sum(x*x for x in baseline[0])
+  assert all(a*b >= -1e-12 for a, b in zip(actual[0], baseline[0], strict=True))
+  actual_turn, baseline_turn = run(selected, True), run(0., True)
+  assert actual_turn[1:] == baseline_turn[1:]
+  assert actual_turn[0] == pytest.approx(baseline_turn[0], rel=1e-6, abs=1e-8)
+  assert run(selected, False, ControllerMode.STANDARD) == run(0., False, ControllerMode.STANDARD)
+
+
+def test_ev6_taper_keeps_low_speed_and_substantial_curve_effect_small():
+  from openpilot.starpilot.lateral import hkg_vehicle_shaping as shaping
+  for speed in (0., 3., 26., 36.):
+    for acceleration in (-2., -1., 0., 1., 2.):
+      actual = shaping.get_kia_ev6_center_output_scale(acceleration, speed)
+      speed_weight = shaping._kia_ev6_sigmoid((speed - shaping.KIA_EV6_CENTER_OUTPUT_TAPER_SPEED) /
+                                              shaping.KIA_EV6_CENTER_OUTPUT_TAPER_SPEED_WIDTH)
+      center_weight = shaping._kia_ev6_sigmoid((shaping.KIA_EV6_CENTER_OUTPUT_TAPER_LAT - abs(acceleration)) /
+                                               shaping.KIA_EV6_CENTER_OUTPUT_TAPER_LAT_WIDTH)
+      old = 1. - .14 * speed_weight * center_weight
+      assert .78 <= actual <= old <= 1.
+      if speed <= 3.:
+        assert old - actual < .002
+      if abs(acceleration) >= 1.:
+        assert old - actual < .00002

openpilot/starpilot/lateral/tests/test_lane_centering.py

@@ -483,11 +483,70 @@ class TestSingleLineTracking(unittest.TestCase):
       controller = LaneCenteringController()
       source = observation(m, base_curvature=0., measured_curvature=0., lateral_delay_s=.1)
       result = controller.update(source)
-      self.assertGreater(-direction * result.correction, 0.)
+      self.assertEqual(result.correction, 0.)
       assert result.candidate_curvature is not None
       self.assertEqual(controller.guard_applied(m, 0., result.candidate_curvature, 0., 20., .1), 0.)
       self.assertEqual(controller._correction, 0.)
 
+  def test_centered_planned_curves_do_not_build_an_opposing_nudge(self):
+    from openpilot.starpilot.lateral.lane_clearance import _center_motion
+    for curvature in (-.006, -.002, .002, .006):
+      x, y, _ = _center_motion(curvature, curvature, 25., .1)
+      x, y = np.r_[0., x, 50.], np.r_[0., y, y[-1]]
+      m = model()
+      for index, shift in ((1, -1.8), (2, 1.8)):
+        m.laneLines[index].x, m.laneLines[index].y = x.tolist(), (y + shift).tolist()
+      m.position.x, m.position.y = x.tolist(), (y - math.copysign(.6, curvature)).tolist()
+      m.position.yStd = [.1] * len(x)
+      controller = LaneCenteringController()
+      source = observation(m, base_curvature=curvature, measured_curvature=curvature,
+                           speed_mps=25., lateral_delay_s=.1)
+      self.assertAlmostEqual(controller.update(source).correction, 0., delta=1e-8)
+
+  def test_curve_spacing_limit_shapes_the_target_before_it_ramps(self):
+    from openpilot.starpilot.lateral.lane_clearance import _center_motion, limit_tracking
+    from openpilot.starpilot.lateral.lane_centering import ACQUIRE_TAU
+    for sign in (-1., 1.):
+      base, delta, speed = sign * .004, sign * .00004, 25.
+      x, y, _ = _center_motion(base + delta, base, speed, .1)
+      x, y = np.r_[0., x, 50.], np.r_[0., y, y[-1]]
+      m = model()
+      for index, shift in ((1, -1.8), (2, 1.8)):
+        m.laneLines[index].x, m.laneLines[index].y = x.tolist(), (y + shift).tolist()
+      m.position.x, m.position.y = x.tolist(), (y - sign * .7).tolist()
+      m.position.yStd = [.1] * len(x)
+      source = observation(m, base_curvature=base, measured_curvature=base, speed_mps=speed, lateral_delay_s=.1)
+      controller = LaneCenteringController()
+      first = controller.update(source)
+      self.assertGreater(sign * first.correction, 0.)
+      self.assertLess(abs(first.correction), abs(delta) * .04)
+      raw = raw_correction(m, speed, 0., 0.)
+      assert raw is not None
+      path = controller._tracker.path
+      assert path is not None
+      target = limit_tracking(path, base, raw * .3 * 1.5, base, speed, .1)
+      self.assertAlmostEqual(first.correction, target * (1. - math.exp(-.01 / ACQUIRE_TAU)), delta=1e-12)
+      assert first.candidate_curvature is not None
+      applied = controller.guard_applied(m, base, first.candidate_curvature, base, speed, .1) - base
+      self.assertAlmostEqual(applied, first.correction, delta=1e-12)
+
+  def test_nudge_build_and_reversal_respect_lateral_acceleration_rate(self):
+    for speed in (5., 20., 35.):
+      controller = LaneCenteringController()
+      previous = 0.
+      for tick in range(160):
+        sign = 1. if tick < 80 else -1.
+        dt = (.01, .02, .05)[tick % 3]
+        source = observation(model(left=-1.8 + sign * 1.4, right=1.8 + sign * 1.4),
+                             speed_mps=speed, elapsed_seconds=dt, model_sample_id=tick,
+                             settings=LaneCenteringSettings(True, e2e_authority=0., strength=3.))
+        result = controller.update(source)
+        self.assertLessEqual(abs(result.correction - previous) * speed * speed / dt, 1. + 1e-12)
+        previous = result.correction
+      self.assertLess(previous, 0.)
+      self.assertEqual(controller.update(replace(source, driver_override=True)).correction, 0.)
+      self.assertEqual(controller.update(replace(source, lateral_active=False)).correction, 0.)
+
   def test_two_line_tracking_interpolates_shared_coordinates_and_bounds_offset(self):
     m = model(left=-1.31, right=1.31)
     m.laneLines[2].x = [float(x) + .25 for x in XS]

openpilot/starpilot/lateral/tests/test_lane_runtime.py

@@ -162,6 +162,8 @@ class LaneRuntimeTests(unittest.TestCase):
           feed(controls, 1_000_000_000 + tick * 10_000_000, tick)
           controls.state_control()
       self.assertIsNotNone(controls.last_lane_centering_result)
+      assert controls.last_lane_centering_result is not None
+      self.assertEqual(controls.last_lane_centering_result.correction, 0.)
       self.assertEqual(controls.lane_centering_applied, 0.)
       self.assertAlmostEqual(controls.desired_curvature, .0002)
 

openpilot/starpilot/replay/tests/test_render.py

@@ -574,11 +574,11 @@ def test_published_readiness_survives_buffering_but_rejects_seeks_and_old_epochs
 
 
 @pytest.mark.parametrize('matching_frame', (False, True))
-def test_epoch_reset_preserves_native_camera_stream_and_eligible_paused_frame(monkeypatch, matching_frame):
+@pytest.mark.parametrize('streams,expected_streams', (({0, 2}, [0, 2]), ([2, 0], [2, 0])))
+def test_epoch_reset_preserves_native_camera_stream_and_eligible_paused_frame(monkeypatch, matching_frame, streams, expected_streams):
   monkeypatch.setattr('openpilot.starpilot.ui.runtime_snapshot.RuntimeSnapshotAdapter', Mock())
   monkeypatch.setattr('openpilot.starpilot.ui.pip_render.PiPRenderer', Mock())
   core: Any = object.__new__(Compositor)
-  streams = {0, 2}
   client = Mock(frame_id=5 if matching_frame else 4, timestamp_sof=9_000_000_000, timestamp_eof=9_010_000_000)
   client.is_connected.return_value = True
   frame = SimpleNamespace(frame_id=5)
@@ -610,7 +610,7 @@ def test_epoch_reset_preserves_native_camera_stream_and_eligible_paused_frame(mo
     callback()
   assert core.render_receipts == {}
   assert core.camera.client is client
-  assert core.camera.available_streams == streams
+  assert core.camera.available_streams == expected_streams
   assert core.camera.available_streams is not streams
   core.camera._initialize_textures.assert_called_once()
   assert core.camera.frame is (frame if matching_frame else None)

openpilot/starpilot/software/fast_update.py

@@ -30,6 +30,15 @@ def required_os_version(launch):
   return versions[0] if len(versions) == 1 else None
 
 
+def legacy_os_compatible(launch, current_os):
+  required_os = required_os_version(launch)
+  accepted = re.findall(r'^\s*export\s+AGNOS_ACCEPTED_VERSIONS=["\']([^"\'\n]+)["\']\s*$', launch, re.MULTILINE)
+  accepted_os = {required_os}
+  if len(accepted) == 1:
+    accepted_os.update(accepted[0].split())
+  return bool(required_os and current_os and current_os in accepted_os)
+
+
 def validate_legacy_return(git, revision, branch):
   if branch not in ('Dom', 'StarPilot'):
     raise FastUpdateIncompatible('Legacy return is only available for Dom and StarPilot 6.x releases')
@@ -243,12 +252,7 @@ def fast_update(repo, branch, *, params, parked, expected_commit, current_os=Non
     validate_revision(repo, target)
   launch = git('show', f'{target}:launch_env.sh')
   required_os = required_os_version(launch)
-  accepted_os = {required_os}
-  if legacy_return:
-    accepted = re.findall(r'^\s*export\s+AGNOS_ACCEPTED_VERSIONS=["\']([^"\'\n]+)["\']\s*$', launch, re.MULTILINE)
-    if len(accepted) == 1:
-      accepted_os.update(accepted[0].split())
-  if not required_os or not current_os or current_os not in accepted_os:
+  if not (legacy_os_compatible(launch, current_os) if legacy_return else required_os and current_os == required_os):
     raise FastUpdateIncompatible('Update requires a different or unverified operating system')
   if target == previous and branch == initial_branch and not git('status', '--porcelain', '--untracked-files=no').strip():
     notify('complete', 'Already up to date.')

openpilot/starpilot/system/starport/daemon.py

@@ -213,6 +213,7 @@ def main() -> int:
   from openpilot.common.hardware import COMMA_HARDWARE
   placement = None
   if COMMA_HARDWARE:
+    os.setpriority(os.PRIO_PROCESS, 0, 5)
     from openpilot.starpilot.system.starport.placement import RendererPlacement
     placement = RendererPlacement(idle=False, report=lambda event: print(json.dumps(event), flush=True))
     placement.start()

openpilot/starpilot/system/starport/projection_layout.py

@@ -137,7 +137,8 @@ def layout_metadata_for_viewport(viewport):
   return profile
 
 
-# Android Auto display switches, independent of the comma screen's preferences.
+# Android Auto display switches. The legacy personality field remains in saved layouts for compatibility;
+# the shared comma preference controls that notice on every screen.
 DISPLAY_DEFAULTS = {'blindSpotBorder': False, 'signalBorder': False, 'blindSpotCamera': False,
                     'hidePersonalityAlerts': False, 'wheelPedalFeedback': False,
                     'leadIndicator': False, 'leadInfo': 'Off', 'speedSources': False,
@@ -147,7 +148,7 @@ LEAD_INFO_CHOICES = ('Off', 'Distance', 'Speed')
 
 
 def projection_appearance(appearance, document):
-  """The comma's onroad appearance with every car-screen choice taken from the AA layout instead."""
+  """Apply screen-specific choices while retaining the shared personality-alert preference."""
   from dataclasses import replace
   from openpilot.starpilot.ui.appearance_preferences import LeadInfoMode
   lead_indicator = document.get('leadIndicator', False) is True
@@ -159,7 +160,6 @@ def projection_appearance(appearance, document):
                  show_lead_indicator=lead_indicator, lead_info_mode=lead_info_mode,
                  show_blindspot_border=document.get('blindSpotBorder', False) is True,
                  show_signal_border=document.get('signalBorder', False) is True,
-                 hide_personality_alerts=document.get('hidePersonalityAlerts', False) is True,
                  wheel_pedal_feedback=document.get('wheelPedalFeedback', False) is True)
 
 

openpilot/starpilot/system/starport/tests/test_placement.py

@@ -1,4 +1,12 @@
 import os
+import argparse
+import ast
+import json
+from pathlib import Path
+import sys
+import types
+import unittest
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -162,3 +170,49 @@ def test_background_pool_tracks_online_cores_without_using_driving_cores(monkeyp
 def test_rejects_invalid_core_pools(cores):
   with pytest.raises(ValueError):
     RendererPlacement(core=cores)
+
+
+class PlacementReached(Exception):
+  pass
+
+
+class TestStarPortStartup(unittest.TestCase):
+  def test_transport_priority_before_workers(self):
+    tree = ast.parse((Path(__file__).parents[1] / 'daemon.py').read_text())
+    main = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'main')
+    for device in ('mici', 'tici', 'tizi', 'pc'):
+      with self.subTest(device=device):
+        order = []
+        hardware = types.ModuleType('openpilot.common.hardware')
+        hardware.COMMA_HARDWARE = device != 'pc'
+        hardware.HARDWARE = types.SimpleNamespace(get_device_type=lambda device=device: device)
+        placement = types.ModuleType('openpilot.starpilot.system.starport.placement')
+        instance = Mock()
+
+        def start(order=order):
+          order.append('start')
+          raise PlacementReached
+
+        instance.start.side_effect = start
+        placement.RendererPlacement = Mock(return_value=instance)
+        fake_os = Mock(PRIO_PROCESS=0)
+        fake_os.setpriority.side_effect = lambda *args, order=order: order.append(('priority', args))
+        fake_os.unlink.side_effect = PlacementReached
+        namespace: dict[str, object] = {'argparse': argparse, '__doc__': 'fixture', 'os': fake_os, 'json': json,
+                                        'STARPORT_SOCKET_PATH': 'fixture'}
+        exec(compile(ast.Module(body=[main], type_ignores=[]), '<actual-starport-main>', 'exec'), namespace)
+        run_main = namespace['main']
+        assert callable(run_main)
+        modules = {'openpilot.common.hardware': hardware, 'openpilot.starpilot.system.starport.placement': placement}
+        with patch.dict(sys.modules, modules), patch.object(sys, 'argv', ['starportd']):
+          with self.assertRaises(PlacementReached):
+            run_main()
+        if device != 'pc':
+          self.assertEqual(order, [('priority', (0, 0, 5)), 'start'])
+        else:
+          fake_os.setpriority.assert_not_called()
+        if device != 'pc':
+          placement.RendererPlacement.assert_called_once()
+          self.assertIs(placement.RendererPlacement.call_args.kwargs['idle'], False)
+        else:
+          placement.RendererPlacement.assert_not_called()

openpilot/starpilot/system/starport/tests/test_projection_layout.py

@@ -216,6 +216,19 @@ class TestProjectionLayout(unittest.TestCase):
         with self.subTest(key=key, value=value), self.assertRaises(ValueError):
           validate_layout({**document, key: value}, SCREEN)
 
+  def test_legacy_personality_choice_cannot_override_shared_preference(self):
+    from openpilot.starpilot.system.starport.projection_layout import projection_appearance
+    from openpilot.starpilot.ui.appearance_preferences import OnroadAppearance
+    for old_value in (False, True):
+      saved = validate_layout({**default_layout(SCREEN), 'hidePersonalityAlerts': old_value,
+                               'wheelPedalFeedback': True}, SCREEN)
+      round_trip = decode_layout(json.dumps(saved).encode(), SCREEN)
+      self.assertEqual(round_trip, saved)
+      for shared_value in (False, True):
+        actual = projection_appearance(OnroadAppearance(hide_personality_alerts=shared_value), round_trip)
+        self.assertIs(actual.hide_personality_alerts, shared_value)
+        self.assertTrue(actual.wheel_pedal_feedback)
+
   def test_lead_info_starts_off_and_takes_only_known_choices(self):
     document = default_layout(SCREEN)
     self.assertEqual((document['leadIndicator'], document['leadInfo']), (False, 'Off'))

openpilot/starpilot/system/starport/tests/test_projection_layout_runtime.py

@@ -302,13 +302,13 @@ class TestProjectionLayoutRuntime(unittest.TestCase):
     appearance = rendered[-1].appearance
     self.assertEqual((appearance.hide_personality_alerts, appearance.wheel_pedal_feedback, appearance.camera_view,
                       appearance.driver_camera_on_reverse, rendered[-1].reverse_driver_camera),
-                     (False, False, CameraViewChoice.DRIVER, True, False), 'the car screen never switches to the driver camera')
-    self.assertIs(adapter.appearance_override(comma).hide_personality_alerts, False, 'the adapter hides notices by AA choice')
+                     (True, False, CameraViewChoice.DRIVER, True, False), 'the car screen never switches to the driver camera')
+    self.assertIs(adapter.appearance_override(comma).hide_personality_alerts, True, 'the shared comma preference controls notices')
     view.apply_layout({**layout, 'hidePersonalityAlerts': True, 'wheelPedalFeedback': True})
     self.assertEqual(invalidated, [1])
     view.render()
     self.assertEqual((rendered[-1].appearance.hide_personality_alerts, rendered[-1].appearance.wheel_pedal_feedback), (True, True))
-    self.assertIs(adapter.appearance_override(OnroadAppearance()).hide_personality_alerts, True)
+    self.assertIs(adapter.appearance_override(OnroadAppearance()).hide_personality_alerts, False, 'stale AA value cannot override the shared choice')
 
   def test_lead_marker_and_info_follow_the_aa_layout_only(self):
     from openpilot.starpilot.ui.appearance_preferences import LeadInfoMode, OnroadAppearance

openpilot/starpilot/ui/tests/test_appearance_settings.py

@@ -127,6 +127,65 @@ class AppearanceSettingsTests(unittest.TestCase):
     self.assertFalse(onroad_appearance(self.params).hide_personality_alerts)
     self.assertEqual(path.read_bytes(), b"bad")
 
+  def test_personality_toggle_controls_native_and_existing_projection_layouts(self):
+    from openpilot.starpilot.system.starport.projection_layout import default_layout_for_viewport, projection_appearance, validate_layout_for_viewport
+    from openpilot.starpilot.ui.tests.test_runtime_snapshot import RuntimeSnapshotAdapter
+
+    viewport = (2880, 1080)
+    for profile in (Profile.LARGE, Profile.COMPACT):
+      for hidden in (True, False):
+        request = row_change(self.row("HidePersonalityAlerts", profile))
+        assert request is not None
+        self.assertEqual(request.value, "On" if hidden else "Off")
+        self.assertTrue(self.owner.apply(request))
+        for old_aa_choice in (False, True):
+          layout = default_layout_for_viewport(viewport)
+          layout.update(hidePersonalityAlerts=old_aa_choice, wheelPedalFeedback=True)
+          saved = validate_layout_for_viewport(layout, viewport)
+          self.assertIs(saved['hidePersonalityAlerts'], old_aa_choice)
+          self.assertTrue(saved['wheelPedalFeedback'])
+          ui = ui_fake()
+          ui.params = self.params
+          source = ui.sm.messages['selfdriveState']
+          source.personality = NS(raw=1)
+          source.alertSize, source.alertType = NS(raw=1), 'personalityChanged/warning'
+          source.alertText1 = 'Driving Personality: Standard'
+          for projection in (False, True):
+            with self.subTest(profile=profile, hidden=hidden, legacy_aa=old_aa_choice, projection=projection):
+              adapter = RuntimeSnapshotAdapter(ui)
+              if projection:
+                adapter.appearance_override = lambda appearance, document=saved: projection_appearance(appearance, document)
+              alert = adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad.alert
+              self.assertEqual(alert.size, AlertSize.NONE if hidden else AlertSize.SMALL)
+              source.alertSize, source.alertStatus = NS(raw=3), NS(raw=2)
+              source.alertText1 = 'Take control'
+              safety = adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad.alert
+              self.assertEqual((safety.size, safety.text1), (AlertSize.FULL, 'Take control'))
+              source.alertSize, source.alertStatus = NS(raw=1), NS(raw=0)
+              source.alertText1 = 'Driving Personality: Standard'
+
+  def test_personality_toggle_refreshes_on_projection_without_reconnect(self):
+    from openpilot.starpilot.system.starport.projection_layout import default_layout_for_viewport, projection_appearance
+    from openpilot.starpilot.ui.tests.test_runtime_snapshot import RuntimeSnapshotAdapter
+
+    ui = ui_fake()
+    ui.params = self.params
+    source = ui.sm.messages['selfdriveState']
+    source.personality = NS(raw=1)
+    source.alertSize, source.alertType = NS(raw=1), 'personalityChanged/warning'
+    source.alertText1 = 'Driving Personality: Standard'
+    # A conflicting old AA-only preference must never pin either state.
+    layout = {**default_layout_for_viewport((2880, 1080)), 'hidePersonalityAlerts': True}
+    adapter = RuntimeSnapshotAdapter(ui)
+    adapter.appearance_override = lambda appearance: projection_appearance(appearance, layout)
+    for second, hidden in enumerate((False, True, False)):
+      self.params.put_bool('HidePersonalityAlerts', hidden, block=True)
+      now = NOW + second * 1_000_000_000
+      ui.sm.logMonoTime['selfdriveState'] = now
+      ui.sm.recv_time['selfdriveState'] = now / 1e9
+      alert = adapter.build(ShellMode.ONROAD, now_ns=now).onroad.alert
+      self.assertEqual(alert.size, AlertSize.NONE if hidden else AlertSize.SMALL)
+
   def test_developer_sidebar_external_changes_refresh_without_restart(self):
     from openpilot.starpilot.ui.tests.test_runtime_snapshot import RuntimeSnapshotAdapter
     ui = ui_fake()

openpilot/starpilot/ui/tests/test_mici_transitions.py

@@ -9,6 +9,7 @@ import pytest
 
 from openpilot.selfdrive.ui.mici.layouts import main
 from openpilot.selfdrive.ui.mici.onroad import cameraview
+from openpilot.selfdrive.ui.onroad import cameraview as base_cameraview
 
 
 @pytest.fixture
@@ -152,7 +153,8 @@ def camera(monkeypatch):
 
   monkeypatch.setattr(cameraview, 'VisionIpcClient', new_client)
   monkeypatch.setattr(cameraview, 'COMMA_HARDWARE', True)
-  monkeypatch.setattr(cameraview, 'destroy_egl_image', lambda image: events.append(('egl', image)))
+  monkeypatch.setattr(base_cameraview, 'COMMA_HARDWARE', True)
+  monkeypatch.setattr(base_cameraview, 'destroy_egl_image', lambda image: events.append(('egl', image)))
   monkeypatch.setattr(cameraview.rl, 'unload_texture', lambda texture: events.append(('texture', texture.id)))
   monkeypatch.setattr(cameraview.rl, 'unload_shader', lambda shader: events.append(('shader', shader.id)))
   monkeypatch.setattr(cameraview.rl, 'get_time', lambda: 1.0)

openpilot/starpilot/ui/tests/test_spacepilot_timing.py

@@ -1,11 +1,12 @@
 """Cold native entry and preempted clock reads from the recorded d5 exit paths."""
 
+import time
 from types import SimpleNamespace as NS
 from unittest.mock import Mock
 
 import pytest
 
-from openpilot.starpilot.ui.spacepilot import ignition, native, session
+from openpilot.starpilot.ui.spacepilot import guard, ignition, native, session
 from openpilot.starpilot.ui.tests.test_spacepilot_session import Clock, Vehicle, game_ui, offroad, setup as setup
 from openpilot.starpilot.ui.tests.test_spacepilot_state import mature_gate, read
 from openpilot.system.ui.lib.application import gui_app
@@ -20,11 +21,25 @@ def clock_reads(monkeypatch, *, delayed=1, backwards=False, offset=2_000_000_000
     boots.append(now + offset)
     now += max(delta, 0)
   mono = Mock(side_effect=monos)
-  monkeypatch.setattr(session.time, 'monotonic_ns', mono)
-  monkeypatch.setattr(session.time, 'clock_gettime_ns', Mock(side_effect=boots))
+  # Keep background UI readers from consuming the scripted clock pairs.
+  monkeypatch.setattr(guard, 'time', NS(monotonic_ns=mono, clock_gettime_ns=Mock(side_effect=boots), CLOCK_BOOTTIME=time.CLOCK_BOOTTIME))
   return mono
 
 
+def test_clock_reads_leave_unrelated_process_clocks_untouched(monkeypatch):
+  process_mono, process_boot = time.monotonic_ns, time.clock_gettime_ns
+  mono = clock_reads(monkeypatch, delayed=3)
+  assert time.monotonic_ns is process_mono
+  assert time.clock_gettime_ns is process_boot
+  time.monotonic_ns()
+  time.clock_gettime_ns(time.CLOCK_BOOTTIME)
+  assert mono.call_count == guard.time.clock_gettime_ns.call_count == 0
+  with pytest.raises(ValueError, match='stable device clock'):
+    guard.sample_clocks()
+  assert mono.call_count == 6
+  assert guard.time.clock_gettime_ns.call_count == 3
+
+
 def witness_ui(monkeypatch):
   ui = game_ui()
   Clock().refresh(ui, 0)

openpilot/starpilot/ui/tests/test_startup.py

@@ -77,20 +77,29 @@ class TestStartup(unittest.TestCase):
            patch.object(ui.messaging, "PubMaster"), patch.object(ui, "config_realtime_process") as realtime, \
            patch.object(ui, "drop_realtime") as normal, patch.object(ui.os, "setpriority") as nice, \
            patch.object(ui.os, "sched_getscheduler", return_value=0, create=True), \
-           patch.object(ui.os, "SCHED_OTHER", 0, create=True), patch.object(ui.os, "getpriority", return_value=0), \
+           patch.object(ui.os, "SCHED_OTHER", 0, create=True), patch.object(ui.os, "getpriority", return_value=5 if device == "mici" else 0), \
            patch.object(ui.os, "sched_getaffinity", return_value={0}, create=True), \
            patch.object(ui, "set_core_affinity") as affinity:
         ui.main()
-        realtime.assert_called_once_with(0, ui.Priority.UI)
-        if hardware:
-          normal.assert_called_once_with()
-          nice.assert_called_once_with(ui.os.PRIO_PROCESS, 0, 0)
+        if device == "mici":
+          realtime.assert_not_called()
+          self.assertEqual(normal.call_count, 2)
+          self.assertEqual(nice.call_count, 2)
+          self.assertTrue(all(call.args == (ui.os.PRIO_PROCESS, 0, 5) for call in nice.call_args_list))
+          self.assertEqual(set(affinity.call_args_list[0].args[0]), {0, 1, 2, 7})
+          self.assertTrue(all(set(call.args[0]) == {0, 1, 2, 7} for call in affinity.call_args_list[1:]))
+          self.assertEqual(affinity.call_count, 3)
         else:
-          normal.assert_not_called()
-          nice.assert_not_called()
-        self.assertEqual(affinity.call_count, 2 if hardware else 0)
-        if hardware:
-          self.assertTrue(all(call.args == ([6],) for call in affinity.call_args_list))
+          realtime.assert_called_once_with(0, ui.Priority.UI)
+          if hardware:
+            normal.assert_called_once_with()
+            nice.assert_called_once_with(ui.os.PRIO_PROCESS, 0, 0)
+          else:
+            normal.assert_not_called()
+            nice.assert_not_called()
+          self.assertEqual(affinity.call_count, 2 if hardware else 0)
+          if hardware:
+            self.assertTrue(all(call.args == ([6],) for call in affinity.call_args_list))
 
   def test_slc_publisher_gate_is_independent_of_ui_selection(self):
     self.assertTrue(startup.slc_action_transport_enabled({"STARPILOT_UI_DEV": "1"}))

openpilot/system/updated/tests/test_fast_request.py

@@ -77,6 +77,7 @@ class TestFastRequest(unittest.TestCase):
     module.set_consistent_flag = Mock()
     module.init_overlay = Mock()
     module.validate_revision = Mock()
+    module.validate_update_revision = Mock(return_value=False)
     module.AGNOS = False
     self.updater.target_branch = "SecretGoodStarPilot"
     self.updater._checked_branch = "SecretGoodStarPilot"
@@ -84,7 +85,7 @@ class TestFastRequest(unittest.TestCase):
     for failure in (False, True):
       with self.subTest(failure=failure):
         module.init_overlay.reset_mock()
-        module.validate_revision.reset_mock()
+        module.validate_update_revision.reset_mock()
         self.writes.clear()
         self.values.pop("LastUpdateTime", None)
         self.helper._control_request("download")
@@ -105,9 +106,9 @@ class TestFastRequest(unittest.TestCase):
         self.assertEqual("LastUpdateTime" in self.values, not failure)
         module.finalize_update.assert_called_once()
         module.init_overlay.assert_called_once_with()
-        self.assertEqual(module.validate_revision.call_args_list, [
-          mock.call(module.OVERLAY_MERGED, 'a' * 40),
-          mock.call(module.OVERLAY_MERGED, 'HEAD'),
+        self.assertEqual(module.validate_update_revision.call_args_list, [
+          mock.call(module.OVERLAY_MERGED, 'a' * 40, 'SecretGoodStarPilot'),
+          mock.call(module.OVERLAY_MERGED, 'HEAD', 'SecretGoodStarPilot'),
         ])
 
   def test_fast_request_skips_staging_and_startup_delay(self):

openpilot/system/updated/tests/test_retained_agnos.py

@@ -44,6 +44,93 @@ class TestRetainedAgnos(unittest.TestCase):
     patcher.start()
     self.addCleanup(patcher.stop)
 
+  def test_readiness_flag_never_follows_or_replaces_existing_marker(self):
+    updated = load_updater()
+    updated.FINALIZED = str(self.root)
+    marker = self.root / '.overlay_consistent'
+    target = self.root / 'untouched'
+    target.write_text('keep')
+    marker.symlink_to(target)
+    with patch.object(updated.os, 'sync'):
+      with self.assertRaises(FileExistsError):
+        updated.set_consistent_flag(True)
+      self.assertEqual(target.read_text(), 'keep')
+      updated.set_consistent_flag(False)
+      updated.set_consistent_flag(True)
+      self.assertTrue(marker.is_file())
+      self.assertFalse(marker.is_symlink())
+      with self.assertRaises(FileExistsError):
+        updated.set_consistent_flag(True)
+
+  def test_launcher_readiness_rejects_symlink(self):
+    function = shell_function(ROOT / 'launch_chffrplus.sh', 'update_is_ready')
+    target = self.root / 'target'
+    target.touch()
+    marker = self.root / 'marker'
+    marker.symlink_to(target)
+    command = function + f'\nupdate_is_ready {shlex.quote(str(marker))}'
+    self.assertNotEqual(subprocess.run(['bash', '-c', command]).returncode, 0)
+    marker.unlink()
+    marker.touch()
+    self.assertEqual(subprocess.run(['bash', '-c', command]).returncode, 0)
+
+  def test_launch_discards_inherited_os_version_overrides(self):
+    function = shell_function(ROOT / 'launch_chffrplus.sh', 'reset_update_os_environment')
+    env_file = self.root / 'target_launch_env.sh'
+    env_file.write_text('if [ -z "$AGNOS_VERSION" ]; then export AGNOS_VERSION="19.8.1"; fi\n' +
+                        'if [ -z "$AGNOS_ACCEPTED_VERSIONS" ]; then export AGNOS_ACCEPTED_VERSIONS="19.8.1 20.2.1"; fi\n')
+    script = function + '\nexport AGNOS_VERSION=wrong AGNOS_ACCEPTED_VERSIONS=wrong\n' + \
+      f'reset_update_os_environment\nsource {shlex.quote(str(env_file))}\nprintf "%s|%s" "$AGNOS_VERSION" "$AGNOS_ACCEPTED_VERSIONS"'
+    result = subprocess.check_output(['bash', '-c', script], text=True)
+    self.assertEqual(result, '19.8.1|19.8.1 20.2.1')
+
+  def test_legacy_launch_checks_exact_commit_and_os_before_swap_or_slot_activation(self):
+    repo = self.root / 'staged'
+    repo.mkdir()
+    subprocess.run(['git', 'init', '-q', str(repo)], check=True)
+    subprocess.run(['git', '-C', str(repo), '-c', 'user.name=Test', '-c', 'user.email=test@example.invalid',
+                    'commit', '--allow-empty', '-qm', 'fixture'], check=True)
+    commit = subprocess.check_output(['git', '-C', str(repo), 'rev-parse', 'HEAD'], text=True).strip()
+    version = self.root / 'VERSION'
+    version.write_text('20.2.1\n')
+    marker = repo / '.git/starpilot-retain-agnos'
+    launcher = ROOT / 'launch_chffrplus.sh'
+    text = launcher.read_text()
+    audited = next(iter(load_updater().TRUSTED_LEGACY_RELEASES['StarPilot']))
+    self.assertIn(f'TRUSTED_LEGACY_STARPILOT_COMMIT="{audited}"', text)
+    check = shell_function(launcher, 'staged_os_is_compatible').replace('/VERSION', str(version))
+    activate = shell_function(launcher, 'maybe_activate_legacy_agnos')
+    verifier = self.root / 'verifier'
+    log = self.root / 'calls'
+    verifier.write_text(f'#!/bin/sh\necho verify >> {shlex.quote(str(log))}\n')
+    verifier.chmod(0o755)
+    manifest = self.root / 'agnos.json'
+    manifest.write_text('[]')
+    common = f'TRUSTED_LEGACY_STARPILOT_COMMIT={commit}\n{check}\n{activate}\n'
+    def admission():
+      return subprocess.run(['bash', '-c', common + f'staged_os_is_compatible {shlex.quote(str(repo))}'], check=False).returncode
+    self.assertNotEqual(admission(), 0)  # Missing marker is not a modern update.
+    marker.write_text('wrong\n')
+    self.assertNotEqual(admission(), 0)
+    marker.unlink()
+    marker.symlink_to(version)
+    self.assertNotEqual(admission(), 0)
+    marker.unlink()
+    marker.write_text('20.2.1\n')
+    self.assertEqual(admission(), 0)
+    version.write_text('changed\n')
+    self.assertNotEqual(admission(), 0)  # Manual reboot must recheck the OS.
+    shell = common + f'sudo() {{ echo reboot >> {shlex.quote(str(log))}; }}\n' + \
+      f'maybe_activate_legacy_agnos {shlex.quote(str(repo))} {shlex.quote(str(verifier))} {shlex.quote(str(manifest))}\n'
+    subprocess.run(['bash', '-c', shell], check=True)
+    self.assertFalse(log.exists())  # A pinned legacy install never activates a slot.
+    marker.unlink()
+    modern = common.replace(f'TRUSTED_LEGACY_STARPILOT_COMMIT={commit}', 'TRUSTED_LEGACY_STARPILOT_COMMIT=other')
+    self.assertEqual(subprocess.run(['bash', '-c', modern + f'staged_os_is_compatible {shlex.quote(str(repo))}']).returncode, 0)
+    subprocess.run(['bash', '-c', shell.replace(f'TRUSTED_LEGACY_STARPILOT_COMMIT={commit}',
+                                                  'TRUSTED_LEGACY_STARPILOT_COMMIT=other')], check=True)
+    self.assertEqual(log.read_text().splitlines(), ['verify', 'reboot'])
+
   def test_retained_match_never_flashes_or_queries_slot(self):
     with patch.dict(os.environ, {'AGNOS_VERSION': 'unexpected', 'AGNOS_UPDATE_POLICY': 'auto'}):
       self.updater.handle_agnos_update()

openpilot/system/updated/tests/test_vendored_update.py

@@ -105,6 +105,7 @@ class TestVendoredUpdate(unittest.TestCase):
     self.git(self.merged, "config", "fetch.recurseSubmodules", "true")
 
     self.updated = load_updater()
+    self.updated.TRUSTED_LEGACY_RELEASES = {}
     self.updated.OVERLAY_MERGED = str(self.merged)
     self.updated.METADATA_ROOT = self.root / 'metadata.git'
     self.updated.FINALIZED = str(self.finalized)
@@ -205,6 +206,137 @@ class TestVendoredUpdate(unittest.TestCase):
     self.assertEqual(self.git(self.merged, 'rev-parse', 'HEAD'), self.initial_commit)
     self.assertFalse(self.updater.params.get_bool('DoReboot'))
 
+  def make_legacy_release(self, branch='StarPilot'):
+    self.git(self.source, 'checkout', '-b', branch)
+    (self.source / 'upstream-sync.json').unlink()
+    files = {'prebuilt': '', 'launch_env.sh': 'export AGNOS_VERSION="19.8.1"\nexport AGNOS_ACCEPTED_VERSIONS="19.8.1 20.2.1"\n',
+             'launch_openpilot.sh': '#!/bin/sh\n', 'launch_chffrplus.sh': '#!/bin/sh\n',
+             'selfdrive/ui/lib/starpilot_version.py': 'STARPILOT_DISPLAY_VERSION = "6.7.9"\n',
+             'selfdrive/car/card.py': 'pass\n', 'selfdrive/modeld/modeld.py': 'pass\n'}
+    for name, content in files.items():
+      path = self.source / name
+      path.parent.mkdir(parents=True, exist_ok=True)
+      path.write_text(content)
+      if name in ('launch_openpilot.sh', 'launch_chffrplus.sh'):
+        path.chmod(0o755)
+    target = self.commit_source('Legacy prebuilt release')
+    self.updated.TRUSTED_LEGACY_RELEASES[branch] = frozenset({target})
+    return target
+
+  @patch('openpilot.starpilot.drive_state.evidence.PhysicalSource', return_value=Mock(update_ready=Mock(return_value=True)))
+  def test_download_legacy_starpilot_stages_without_reboot_or_flashing(self, _physical):
+    installed = self.root / 'installed'
+    self.git(self.root, 'clone', '--no-local', str(self.merged), str(installed))
+    self.updated.BASEDIR = str(installed)
+    self.git(installed, 'remote', 'set-url', 'origin', str(self.source))
+    target = self.make_legacy_release()
+    self.updated.AGNOS = True
+    self.updated.HARDWARE.get_os_version.return_value = '20.2.1'
+    with patch.object(self.updater, 'set_params'):
+      self.assertEqual(self.updater.fast_update('StarPilot', self.initial_commit, target, download_only=True), 'downloaded')
+    self.assertEqual(self.git(installed, 'rev-parse', 'HEAD'), self.initial_commit)
+    self.assertEqual(self.git(self.finalized, 'rev-parse', 'HEAD'), target)
+    self.assertTrue(self.updater.update_ready)
+    self.assertFalse(self.updater.params.get_bool('DoReboot'))
+    self.updated.handle_agnos_update.assert_not_called()
+    self.updated.HARDWARE.get_os_version.return_value = 'unverified'
+    with self.assertRaises(self.updated.FastUpdateIncompatible):
+      self.updated.validate_update_revision(str(self.finalized), 'HEAD', 'StarPilot')
+    self.updater.params.put('UpdaterState', 'idle')
+    self.updater.params.put_bool('IsOffroad', True)
+    settings = self.updater.install_settings
+    settings.save(True, '* * * * *', settings.read()['revision'], lambda: True)
+    with patch.object(self.updated, 'system_time_valid', return_value=True):
+      self.assertFalse(self.updater.maybe_install())
+    self.assertFalse(self.updater.params.get_bool('DoReboot'))
+
+  def test_legacy_wrong_os_or_incomplete_release_rejected_before_checkout(self):
+    target = self.make_legacy_release()
+    self.updater.params.put('UpdaterTargetBranch', 'StarPilot')
+    self.updated.HARDWARE.get_os_version.return_value = 'wrong'
+    with self.assertRaises(self.updated.FastUpdateIncompatible):
+      self.updater.fetch_update()
+    self.assertEqual(self.git(self.merged, 'rev-parse', 'HEAD'), self.initial_commit)
+    self.assertFalse(any(command[:2] == ['git', 'checkout'] for command in self.commands))
+    self.assertFalse((self.finalized / '.overlay_consistent').exists())
+    self.updated.HARDWARE.get_os_version.return_value = '19.8.1'
+    (self.source / 'prebuilt').unlink()
+    incomplete = self.commit_source('Incomplete legacy release')
+    self.updated.TRUSTED_LEGACY_RELEASES['StarPilot'] = frozenset({incomplete})
+    self.updater._branches_checked = False
+    with self.assertRaises(self.updated.FastUpdateIncompatible):
+      self.updater.fetch_update()
+    self.assertFalse((self.finalized / '.overlay_consistent').exists())
+    self.assertNotEqual(target, self.git(self.source, 'rev-parse', 'HEAD'))
+
+  def test_legacy_exception_does_not_accept_other_branch_or_malformed_manifest(self):
+    target = self.make_legacy_release()
+    self.updated.HARDWARE.get_os_version.return_value = '19.8.1'
+    with self.assertRaises(ValueError):
+      self.updated.validate_update_revision(str(self.source), target, 'other')
+    (self.source / 'upstream-sync.json').write_text('{}')
+    target = self.commit_source('Invalid modern manifest')
+    with self.assertRaises(ValueError):
+      self.updated.validate_update_revision(str(self.source), target, 'StarPilot')
+
+  def test_legacy_return_rejects_gitlinks_and_submodules(self):
+    self.make_legacy_release()
+    self.updated.HARDWARE.get_os_version.return_value = '19.8.1'
+    self.git(self.source, 'update-index', '--add', '--cacheinfo', f'160000,{self.initial_commit},unsafe')
+    self.git(self.source, 'commit', '-m', 'Invalid legacy gitlink')
+    self.updated.TRUSTED_LEGACY_RELEASES['StarPilot'] = frozenset({self.git(self.source, 'rev-parse', 'HEAD')})
+    with self.assertRaises(self.updated.FastUpdateIncompatible):
+      self.updated.validate_update_revision(str(self.source), 'HEAD', 'StarPilot')
+    self.git(self.source, 'rm', '--cached', 'unsafe')
+    (self.source / '.gitmodules').write_text('[submodule "unsafe"]\npath = unsafe\n')
+    self.commit_source('Invalid legacy submodule metadata')
+    self.updated.TRUSTED_LEGACY_RELEASES['StarPilot'] = frozenset({self.git(self.source, 'rev-parse', 'HEAD')})
+    with self.assertRaises(self.updated.FastUpdateIncompatible):
+      self.updated.validate_update_revision(str(self.source), 'HEAD', 'StarPilot')
+
+  def test_untrusted_legacy_commit_is_rejected_even_if_it_looks_complete(self):
+    target = self.make_legacy_release()
+    self.updated.TRUSTED_LEGACY_RELEASES.clear()
+    self.updated.HARDWARE.get_os_version.return_value = '19.8.1'
+    with self.assertRaises(ValueError):
+      self.updated.validate_update_revision(str(self.source), target, 'StarPilot')
+    audited = load_updater().TRUSTED_LEGACY_RELEASES
+    self.assertEqual(audited, {'StarPilot': frozenset({'4a553da9ac22a14e34513c011aed04c1e93b4619'})})
+
+  def test_reserved_readiness_marker_rejected_in_modern_and_legacy_trees(self):
+    for kind in ('modern-file', 'modern-symlink', 'legacy-file', 'legacy-symlink'):
+      with self.subTest(kind=kind):
+        if kind.startswith('legacy') and self.git(self.source, 'branch', '--show-current') != 'StarPilot':
+          self.make_legacy_release()
+          self.updated.HARDWARE.get_os_version.return_value = '19.8.1'
+        marker = self.source / '.overlay_consistent'
+        if kind.endswith('symlink'):
+          marker.symlink_to('prebuilt')
+        else:
+          marker.write_text('premature')
+        target = self.commit_source(kind)
+        branch = 'StarPilot' if kind.startswith('legacy') else BRANCH
+        self.updated.TRUSTED_LEGACY_RELEASES[branch] = frozenset({target})
+        with self.assertRaisesRegex(ValueError, 'reserved update readiness'):
+          self.updated.validate_update_revision(str(self.source), target, branch)
+        marker.unlink()
+        self.commit_source('Remove marker')
+
+  def test_copy_interruption_never_copies_an_untracked_readiness_marker(self):
+    marker = self.merged / '.overlay_consistent'
+    marker.symlink_to('panda/README.md')
+    original_copy = shutil.copy2
+    def fail_during_copy(src, dst, *args, **kwargs):
+      if Path(src).name == 'README.md':
+        raise OSError('interrupted copy')
+      return original_copy(src, dst, *args, **kwargs)
+    with patch.object(self.updated.shutil, 'copy2', side_effect=fail_during_copy):
+      with self.assertRaises(shutil.Error):
+        self.updated.finalize_update()
+    ready = self.finalized / '.overlay_consistent'
+    self.assertFalse(ready.exists())
+    self.assertFalse(ready.is_symlink())
+
   def test_metadata_refresh_does_not_wake_background_but_check_does(self):
     (self.source / 'new.txt').write_text('available')
     self.commit_source('New update')

openpilot/system/updated/updated.py

@@ -25,7 +25,8 @@ from openpilot.common.vendor_manifest import validate_revision
 from openpilot.starpilot.software.update_control import UpdaterControlError, UpdaterControlServer
 from openpilot.starpilot.software.preferences import AUTOMATIC_DOWNLOADS, download_permitted
 from openpilot.starpilot.software.fast_update import (
-  FastUpdateError, FastUpdateIncompatible, fetch_files, required_os_version, run as run_git,
+  FastUpdateError, FastUpdateIncompatible, fetch_files, legacy_os_compatible, required_os_version, run as run_git,
+  validate_legacy_return,
 )
 from openpilot.starpilot.software.history import MAX_COMMITS, METADATA_ROOT, commit_history
 from openpilot.starpilot.software.auto_install import InstallSettings, cron_matches, scheduled_restart
@@ -161,7 +162,8 @@ def set_consistent_flag(consistent: bool) -> None:
   os.sync()
   consistent_file = Path(os.path.join(FINALIZED, ".overlay_consistent"))
   if consistent:
-    consistent_file.touch()
+    fd = os.open(consistent_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o644)
+    os.close(fd)
   elif not consistent:
     consistent_file.unlink(missing_ok=True)
   os.sync()
@@ -261,6 +263,33 @@ def init_overlay() -> None:
   cloudlog.info(f"git diff output:\n{git_diff}")
 
 
+# Audited legacy release, pinned by Git commit identity. Other releases must
+# satisfy the modern source manifest contract.
+TRUSTED_LEGACY_RELEASES = {'StarPilot': frozenset({'4a553da9ac22a14e34513c011aed04c1e93b4619'})}
+LEGACY_OS_MARKER = '.git/starpilot-retain-agnos'
+
+
+def validate_update_revision(repo: str, revision: str, branch: str) -> bool:
+  """Validate a modern source tree or an explicitly supported legacy return."""
+  def git(*args):
+    return run(['git', *args], repo, download='show' in args)
+
+  if git('ls-tree', revision, '--', '.overlay_consistent').strip():
+    raise ValueError('Source tree contains reserved update readiness marker')
+  manifest_entry = git('ls-tree', revision, '--', 'upstream-sync.json').strip()
+  commit = git('rev-parse', '--verify', f'{revision}^{{commit}}').strip()
+  if not manifest_entry and commit in TRUSTED_LEGACY_RELEASES.get(branch, ()):
+    validate_legacy_return(git, revision, branch)
+    launch = git('show', f'{revision}:launch_env.sh')
+    if not legacy_os_compatible(launch, HARDWARE.get_os_version()):
+      raise FastUpdateIncompatible('Update requires a different or unverified operating system')
+    return True
+  if manifest_entry.split()[:1] == ['100644']:
+    git('show', f'{revision}:upstream-sync.json')
+  validate_revision(repo, revision)
+  return False
+
+
 def finalize_update(progress=None) -> None:
   """Take the current OverlayFS merged view and finalize a copy outside of
   OverlayFS, ready to be swapped-in at BASEDIR. Copy using shutil.copytree"""
@@ -268,7 +297,8 @@ def finalize_update(progress=None) -> None:
   # Remove the update ready flag and any old updates
   cloudlog.info("creating finalized version of the overlay")
   set_consistent_flag(False)
-  validate_revision(OVERLAY_MERGED, "HEAD")
+  branch = run(["git", "symbolic-ref", "--short", "HEAD"], OVERLAY_MERGED).strip()
+  legacy_return = validate_update_revision(OVERLAY_MERGED, "HEAD", branch)
 
   # Copy the merged overlay view and set the update ready flag
   if os.path.exists(FINALIZED):
@@ -302,13 +332,25 @@ def finalize_update(progress=None) -> None:
       reported = now
     return destination
   started = time.monotonic()
-  shutil.copytree(OVERLAY_MERGED, FINALIZED, symlinks=True, copy_function=copy_file)
+  def ignore_readiness_marker(directory, names):
+    return {'.overlay_consistent'} if Path(directory) == Path(OVERLAY_MERGED) else set()
+  shutil.copytree(OVERLAY_MERGED, FINALIZED, symlinks=True, copy_function=copy_file, ignore=ignore_readiness_marker)
   cloudlog.event('update copy complete', duration=time.monotonic() - started, files=copied)
 
   if progress is not None:
     progress('Verifying the prepared update...', None)
   run(["git", "reset", "--hard", "--no-recurse-submodules"], FINALIZED)
-  validate_revision(FINALIZED, "HEAD")
+  validate_update_revision(FINALIZED, "HEAD", branch)
+  marker = Path(FINALIZED, LEGACY_OS_MARKER)
+  marker.unlink(missing_ok=True)
+  if legacy_return:
+    # Bind installation to the exact OS validated during staging. The launcher
+    # rechecks this before swapping trees, even for an ordinary manual reboot.
+    os_version = HARDWARE.get_os_version()
+    if not legacy_os_compatible(run(['git', 'show', 'HEAD:launch_env.sh'], FINALIZED), os_version):
+      raise FastUpdateIncompatible('Update requires a different or unverified operating system')
+    with marker.open('x') as f:
+      f.write(os_version + '\n')
 
   if progress is not None:
     progress('Syncing the prepared update to storage...', None)
@@ -381,11 +423,11 @@ class Updater:
       if (not idle() or self.params.get('UpdaterState') != 'idle' or
           self._check_thread is not None or not self.update_ready or Path(FINALIZED, '.overlay_consistent').is_symlink()):
         return False
-      validate_revision(FINALIZED, 'HEAD')
+      validate_update_revision(FINALIZED, 'HEAD', self.get_branch(FINALIZED))
       return True
     try:
       return scheduled_restart(self.params, self.install_settings, ready, parked, now)
-    except (OSError, ValueError, subprocess.SubprocessError):
+    except (OSError, ValueError, FastUpdateError, subprocess.SubprocessError):
       cloudlog.exception('Scheduled update restart was not admitted')
       return False
     finally:
@@ -760,10 +802,9 @@ class Updater:
     fetched_commit = selected_commit or self.branches[branch]
     run(['git', 'fetch', '--progress', '--depth=1', '--filter=blob:none', '--no-tags', '--no-recurse-submodules',
          'origin', f'+{fetched_commit}:refs/remotes/origin/{branch}'], OVERLAY_MERGED, progress=git_progress)
-    # Fetch the manifest under the download guard before read-only validation.
-    if run(["git", "ls-tree", fetched_commit, "--", "upstream-sync.json"], OVERLAY_MERGED).split()[:1] == ['100644']:
-      run(["git", "show", f"{fetched_commit}:upstream-sync.json"], OVERLAY_MERGED, download=True)
-    validate_revision(OVERLAY_MERGED, fetched_commit)
+    # Validate before checkout or executing any target code. Legacy returns are
+    # restricted to known release branches and must already support this OS.
+    legacy_return = validate_update_revision(OVERLAY_MERGED, fetched_commit, branch)
     def git(*args, stdin=None):
       return run(['git', *args], OVERLAY_MERGED, progress=git_progress, stdin=stdin)
     fetch_files(git, fetched_commit)
@@ -786,11 +827,11 @@ class Updater:
       ["git", "clean", "-xdff"],
     ]
     r = [run(cmd, OVERLAY_MERGED, download=cmd[1] in ('checkout', 'reset'), progress=apply_progress) for cmd in cmds]
-    validate_revision(OVERLAY_MERGED, "HEAD")
+    validate_update_revision(OVERLAY_MERGED, "HEAD", branch)
     cloudlog.info("git reset success: %s", '\n'.join(r))
 
     # TODO: show agnos download progress
-    if AGNOS:
+    if AGNOS and not legacy_return:
       handle_agnos_update()
 
     # Create the finalized, ready-to-swap update

panda/board/obj/gitversion.h

@@ -1,4 +1,4 @@
 #pragma once
 
 extern const uint8_t gitversion[29];
-const uint8_t gitversion[29] = "DEV-d42ddc3a-Pf38f504f-DEBUG";
+const uint8_t gitversion[29] = "DEV-2496b40f-Pf38f504f-DEBUG";

panda/board/obj/version

@@ -1 +1 @@
-DEV-d42ddc3a-Pf38f504f-DEBUG
+DEV-2496b40f-Pf38f504f-DEBUG

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "7769f133116304f49b7b3419263f618e02451e31af54f2dc3354feea9e76cb16", "schema_version": 1}
+{"protected_inventory_sha256": "0edacc0cc483a2f6415617e8ca98c43e0b0106c2016ccb7490a5aae5f766bf1d", "schema_version": 1}

tools/ci/custom_pytest_workloads.json

@@ -16,117 +16,117 @@
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-32-18-4-2]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-32-20-3-5]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-40-18-0-6]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-18-0-0]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-18-4-2]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-20-3-5]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-40-18-0-6]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-18-0-0]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-18-4-2]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-20-3-5]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-18-0-0]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-18-4-2]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-20-3-5]": {
       "basis": "Existing exact capability guard, preserved from8b9338a0; skipped remains unavailable, never passed",
       "guard": "gpu_env",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753"
+      "source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6"
     },
     "openpilot/starpilot/ui/tests/test_onroad_map.py::test_overlay_draws_heading_up_roads_route_and_soft_edges": {
       "basis": "Existing map fixture alias gl=test_gpu_nv12.gpu; both consumer and shared guard definition pinned; unavailable remains SKIPPED, never passed",
       "guard": "gpu_env",
-      "guard_source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753",
+      "guard_source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6",
       "guard_source_path": "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "b3784c7cc0ed58db03aab594c335341a2a11dd1a"
+      "source_blob": "e6d250fc454f44298286e05acd361bf11ffb6063"
     },
     "openpilot/starpilot/ui/tests/test_onroad_map.py::test_overlay_opacity_fades_map_background": {
       "basis": "Existing map fixture alias gl=test_gpu_nv12.gpu; both consumer and shared guard definition pinned; unavailable remains SKIPPED, never passed",
       "guard": "gpu_env",
-      "guard_source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753",
+      "guard_source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6",
       "guard_source_path": "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "b3784c7cc0ed58db03aab594c335341a2a11dd1a"
+      "source_blob": "e6d250fc454f44298286e05acd361bf11ffb6063"
     },
     "openpilot/starpilot/ui/tests/test_onroad_map.py::test_overlay_without_gps_waits_without_drawing_roads": {
       "basis": "Existing map fixture alias gl=test_gpu_nv12.gpu; both consumer and shared guard definition pinned; unavailable remains SKIPPED, never passed",
       "guard": "gpu_env",
-      "guard_source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753",
+      "guard_source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6",
       "guard_source_path": "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "b3784c7cc0ed58db03aab594c335341a2a11dd1a"
+      "source_blob": "e6d250fc454f44298286e05acd361bf11ffb6063"
     },
     "openpilot/starpilot/ui/tests/test_onroad_map.py::test_route_stays_bright_at_minimum_map_opacity": {
       "basis": "Existing map fixture alias gl=test_gpu_nv12.gpu; both consumer and shared guard definition pinned; unavailable remains SKIPPED, never passed",
       "guard": "gpu_env",
-      "guard_source_blob": "242e8c56fe75b4eb9422bfa2424a91d8d98c8753",
+      "guard_source_blob": "b78ee1c0d246edcb22141b4db1b7500d4deb45c6",
       "guard_source_path": "openpilot/starpilot/system/starport/tests/test_gpu_nv12.py",
       "reason": "Skipped: STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context",
-      "source_blob": "b3784c7cc0ed58db03aab594c335341a2a11dd1a"
+      "source_blob": "e6d250fc454f44298286e05acd361bf11ffb6063"
     }
   },
   "independent_functions": {

tools/ci/lifecycle_build_outputs.json

@@ -147,6 +147,10 @@
     "recipe": "openpilot/selfdrive/pandad/SConscript",
     "required_tracked": false
   },
+  "openpilot/selfdrive/pandad/tests/test_panda_faults": {
+    "recipe": "openpilot/selfdrive/pandad/SConscript",
+    "required_tracked": false
+  },
   "openpilot/selfdrive/pandad/tests/test_panda_owner": {
     "recipe": "openpilot/selfdrive/pandad/SConscript",
     "required_tracked": false

tools/ci/tests/test_custom_pytest_shards.py

@@ -151,6 +151,53 @@ def test_changed_source_in_all_jobs_denied_against_checkout(jobs):
     shards.aggregate(directory, 2)
 
 
+@pytest.mark.parametrize('mutation', [None, 'revision', 'tree', 'source', 'source_mode', 'recipe',
+                                     'extra_input', 'missing_output', 'native_bytes', 'native_mode'])
+def test_real_build_comparison_keeps_checkout_and_native_guards(jobs, monkeypatch, mutation):
+  from tools.ci.lifecycle_shards import same_build_inputs
+  directory, paths, catalog = jobs
+  name = 'openpilot/selfdrive/pandad/tests/test_panda_faults'
+  recipe = 'openpilot/selfdrive/pandad/SConscript'
+  inventory_path = Path(shards.__file__).with_name('lifecycle_build_outputs.json')
+  inventory = json.loads(inventory_path.read_text())
+  record = {'sha256': 'a' * 64, 'mode': 0o644}
+  expected = source()
+  expected['build'] = {
+    'inputs': {'revision': 'commit', 'tree': 'tree', 'tracked_inputs': {name + '.cc': copy.deepcopy(record), recipe: copy.deepcopy(record)},
+               'changes': {}, 'untracked': {}, 'recipes': {recipe: copy.deepcopy(record)},
+               'output_inventory_sha256': shards.checksum(inventory_path.read_bytes())},
+    'declared_outputs': {path: copy.deepcopy(record) if row['required_tracked'] else None for path, row in inventory.items()},
+  }
+  monkeypatch.setattr(shards, 'source_snapshot', lambda root=shards.ROOT: copy.deepcopy(expected))
+  monkeypatch.setattr(shards, 'same_build_inputs', same_build_inputs)
+  for path in paths:
+    built = copy.deepcopy(expected)
+    built['build']['declared_outputs'][name] = {'sha256': 'b' * 64, 'mode': 0o755}
+    if mutation in ('revision', 'tree'):
+      built['build']['inputs'][mutation] = 'different'
+    elif mutation == 'source':
+      built['build']['inputs']['tracked_inputs'][name + '.cc']['sha256'] = 'c' * 64
+    elif mutation == 'source_mode':
+      built['build']['inputs']['tracked_inputs'][name + '.cc']['mode'] = 0o755
+    elif mutation == 'recipe':
+      built['build']['inputs']['recipes'][recipe] = {'sha256': 'c' * 64, 'mode': 0o644}
+    elif mutation == 'extra_input':
+      built['build']['inputs']['untracked'][name + '_other'] = record
+    elif mutation == 'missing_output':
+      del built['build']['declared_outputs'][name]
+    after = copy.deepcopy(built)
+    if mutation == 'native_bytes':
+      after['build']['declared_outputs'][name]['sha256'] = 'c' * 64
+    elif mutation == 'native_mode':
+      after['build']['declared_outputs'][name]['mode'] = 0o644
+    rewrite(path, 'receipt.json', lambda d, before=built, after=after: d.update(source_before=before, source_after=after))
+  if mutation is None:
+    assert shards.aggregate(directory, 2)['cases_passed'] == len(catalog)
+  else:
+    with pytest.raises(ValueError):
+      shards.aggregate(directory, 2)
+
+
 @pytest.mark.parametrize('count', [24, 32])
 def test_exact_parameter_nodes_whole_other_modules_and_bounded_heavy_groups(count):
   heavy = 'openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py'
@@ -309,6 +356,14 @@ def unavailable_observation(guard):
           'real_xapk': {'path': shards.REAL_XAPK, 'exists': False}}[guard]
 
 
+def test_audited_capability_fingerprints_match_current_sources():
+  """A binding migration must re-audit pins without broadening any skip guard."""
+  for node, row in shards.policy()['capability_skips'].items():
+    assert shards.blob_id(shards.ROOT / node.split('::')[0]) == row['source_blob'], node
+    if 'guard_source_path' in row:
+      assert shards.blob_id(shards.ROOT / row['guard_source_path']) == row['guard_source_blob'], node
+
+
 def skipped_reports(node, reason):
   return {node: [{'phase': 'setup', 'status': 'skipped', 'duration': .01, 'reason': reason},
                  {'phase': 'teardown', 'status': 'passed', 'duration': .01}]}

tools/ci/tests/test_headless_ci_environment.py

@@ -0,0 +1,52 @@
+"""Hosted collection uses the real display-free Raylib backend, like unit tests."""
+
+import os
+from pathlib import Path
+import re
+import unittest
+from unittest.mock import patch
+
+from tools.ci.vehicle_pytest import serial_environment
+
+
+ROOT = Path(__file__).resolve().parents[3]
+RUNNERS = (
+  ('unit_tests', 'Run unit tests'),
+  ('custom_pytest', 'Test every assigned custom pytest node serially'),
+  ('aol_lifecycles', 'Gate every assigned lifecycle module serially'),
+  ('safety_release_contracts', 'Run every assigned RELEASE target serially'),
+)
+
+
+class TestHeadlessCiEnvironment(unittest.TestCase):
+  def runner_step(self, workflow, job_name, step_name):
+    jobs = dict(re.findall(r'^  ([a-z_]+):\n(.*?)(?=^  [a-z_]+:|\Z)', workflow, re.M | re.S))
+    steps = dict(re.findall(r'^    - name: ([^\n]+)\n(.*?)(?=^    - |\Z)', jobs[job_name], re.M | re.S))
+    return steps[step_name]
+
+  def assert_headless_step(self, step):
+    self.assertRegex(step, r'(?m)^      env:\n(?:        .*\n)*?        RAYLIB_BACKEND: headless\n')
+
+  def test_hosted_collection_selects_native_headless_backend(self):
+    workflow = (ROOT / '.github/workflows/tests.yaml').read_text()
+    for job_name, step_name in RUNNERS:
+      with self.subTest(job=job_name):
+        self.assert_headless_step(self.runner_step(workflow, job_name, step_name))
+
+  def test_missing_or_desktop_backend_is_rejected(self):
+    workflow = (ROOT / '.github/workflows/tests.yaml').read_text()
+    for job_name, step_name in RUNNERS:
+      step = self.runner_step(workflow, job_name, step_name)
+      for replacement in ('', '        RAYLIB_BACKEND: desktop\n'):
+        with self.subTest(job=job_name, replacement=replacement), self.assertRaises(AssertionError):
+          self.assert_headless_step(step.replace('        RAYLIB_BACKEND: headless\n', replacement))
+
+  def test_serial_vehicle_pytest_preserves_backend(self):
+    with patch.dict(os.environ, {'RAYLIB_BACKEND': 'headless'}):
+      with serial_environment():
+        self.assertEqual(os.environ['RAYLIB_BACKEND'], 'headless')
+      self.assertEqual(os.environ['RAYLIB_BACKEND'], 'headless')
+
+
+if __name__ == '__main__':
+  unittest.main()

tools/ci/tests/test_lifecycle_build_identity.py

@@ -142,6 +142,62 @@ class BuildIdentityTests(unittest.TestCase):
         helper.build_identity(self.root)
       path.write_bytes(original)
 
+  def test_panda_faults_is_an_exact_optional_output_with_pinned_inputs(self):
+    name = 'openpilot/selfdrive/pandad/tests/test_panda_faults'
+    recipe = 'openpilot/selfdrive/pandad/SConscript'
+    self.assertEqual(self.inventory.get(name), {'recipe': recipe, 'required_tracked': False})
+    actual_recipe = Path(__file__).parents[3].joinpath(recipe).read_text()
+    self.assertIn("env.Program('tests/test_panda_faults', ['tests/test_panda_faults.cc']", actual_recipe)
+    output = self.root / name
+    source = self.root / (name + '.cc')
+    source.write_text('native faults test source\n')
+    subprocess.run([*GIT, 'rm', '--cached', name], cwd=self.root, check=True, capture_output=True)
+    subprocess.run([*GIT, 'add', name + '.cc'], cwd=self.root, check=True)
+    subprocess.run([*GIT, '-c', 'user.name=Test', '-c', 'user.email=test@example.com', 'commit', '-qm', 'Faults test source'],
+                   cwd=self.root, check=True)
+    output.unlink()
+    checkout = helper.build_identity(self.root)
+    output.write_bytes(b'independently built faults executable')
+    output.chmod(0o755)
+    built = helper.build_identity(self.root)
+    self.assertTrue(helper.same_build_inputs(checkout, built))
+    self.assertIsNone(checkout['declared_outputs'][name])
+    self.assertEqual(built['declared_outputs'][name], helper.file_record(output))
+    self.assertNotIn(name, built['inputs']['untracked'])
+    self.assertEqual(built['inputs']['tracked_inputs'][name + '.cc'], helper.file_record(source))
+    self.assertEqual(built['inputs']['recipes'][recipe], helper.file_record(self.root / recipe))
+    for mutation in ('bytes', 'mode'):
+      with self.subTest(output_mutation=mutation):
+        if mutation == 'bytes':
+          output.write_bytes(b'another independent faults build')
+        else:
+          output.chmod(0o644)
+        rebuilt = helper.build_identity(self.root)
+        self.assertTrue(helper.same_build_inputs(built, rebuilt))
+        self.assertNotEqual(built, rebuilt)  # The per-job exact before/after guard must still reject this.
+    for path in (source, self.root / recipe, self.root / 'opendbc_repo/native_input.h'):
+      with self.subTest(input_mutation=str(path.relative_to(self.root))):
+        original = path.read_bytes()
+        path.write_bytes(b'changed real input')
+        self.assertFalse(helper.same_build_inputs(built, helper.build_identity(self.root)))
+        path.write_bytes(original)
+    source.chmod(0o755)
+    self.assertFalse(helper.same_build_inputs(built, helper.build_identity(self.root)))
+    source.chmod(0o644)
+    for name_suffix in ('_other', '.input.h'):
+      neighbor = output.with_name(output.name + name_suffix)
+      neighbor.write_bytes(b'undeclared neighboring input')
+      self.assertFalse(helper.same_build_inputs(built, helper.build_identity(self.root)))
+      neighbor.unlink()
+    for metadata in ('missing', 'malformed'):
+      forged = copy.deepcopy(built)
+      if metadata == 'missing':
+        del forged['declared_outputs'][name]
+      else:
+        forged['declared_outputs'][name] = {'sha256': 'bad', 'mode': 0o755}
+      with self.assertRaises(ValueError):
+        helper.same_build_inputs(checkout, forged)
+
   def test_missing_declared_artifact_is_denied(self):
     (self.root / 'msgq_repo/msgq/_ipc.so').unlink()
     with self.assertRaises(ValueError):

tools/ci/tests/test_lifecycle_shards.py

@@ -2,6 +2,7 @@ import copy
 import hashlib
 import importlib.util
 import json
+import re
 from pathlib import Path
 import tempfile
 import unittest
@@ -145,11 +146,11 @@ class LifecycleShardTests(unittest.TestCase):
     camera = 'openpilot/starpilot/car/toyota/tests/test_camera_2aa_callers.py'
     self.assertFalse(any(camera in target for shard in shards for target in shard))
 
-  def documents(self, suite="aol-lifecycle"):
+  def documents(self, suite="aol-lifecycle", count=4):
     targets, full = self.catalog()
     partitioner = helper.release_partition if suite == "safety-release" else helper.partition
     matcher = helper.release_matches if suite == "safety-release" else helper.matches
-    partition = partitioner(targets, full, 4)
+    partition = partitioner(targets, full, count)
     source: dict = dict.fromkeys(('revision', 'inputs', 'manifest_sha256', 'dependencies', 'runner_sha256',
                                     'suite_runner_sha256', 'pytest_runner_sha256', 'shard_runner_sha256'), 'same')
     inventory_bytes = Path(__file__).parents[1].joinpath('lifecycle_build_outputs.json').read_bytes()
@@ -164,7 +165,7 @@ class LifecycleShardTests(unittest.TestCase):
     docs = []
     for index, assigned in enumerate(partition):
       ids = [identity for identity in full if any(matcher(identity, target) for target in assigned)]
-      doc = {'suite': suite, 'status': 'completed', 'exit_code': 0, 'shard': {'index': index, 'count': 4},
+      doc = {'suite': suite, 'status': 'completed', 'exit_code': 0, 'shard': {'index': index, 'count': count},
              'targets': assigned, 'selected_test_ids': ids, 'full_catalog_targets': targets, 'full_catalog_test_ids': full,
              'collection_errors': [], 'gate_errors': [], 'source': source, 'source_after': source,
              'native': {'can_sources': can_sources, 'pycapnp': {'sha256': 'b' * 64}, 'safety': {'sha256': '1' * 64,
@@ -174,7 +175,7 @@ class LifecycleShardTests(unittest.TestCase):
       docs.append((doc, result))
     return docs
 
-  def check(self, docs, artifact_mutation=None):
+  def check(self, docs, artifact_mutation=None, count=4):
     with tempfile.TemporaryDirectory() as directory:
       for index, (doc, result) in enumerate(docs):
         path = Path(directory) / str(index)
@@ -193,7 +194,7 @@ class LifecycleShardTests(unittest.TestCase):
         (path / 'coverage.json').write_text(json.dumps(doc))
       if artifact_mutation is not None:
         artifact_mutation(Path(directory))
-      return helper.aggregate(directory, 4, docs[0][0]["suite"])
+      return helper.aggregate(directory, count, docs[0][0]["suite"])
 
   def test_release_partition_accepts_real_class_selectors_and_import_aliases(self):
     targets = ['opendbc_repo/opendbc/safety/tests/a.py::TestRelease', 'b.py::test_case[one]', 'c.py']
@@ -243,6 +244,78 @@ class LifecycleShardTests(unittest.TestCase):
     with self.assertRaises(ValueError):
       self.check(docs)
 
+  def test_release_shard_counts_preserve_whole_targets_and_strict_aggregate(self):
+    targets = [f'module_{i}.py' for i in range(16)]
+    full = [f'module_{i}.Test.test_{j}' for i in range(16) for j in range(i + 1)]
+    for count in (8, 16):
+      with patch.object(self, 'catalog', return_value=(targets, full)):
+        docs = self.documents('safety-release', count=count)
+      self.assertEqual(len(docs), count)
+      self.assertTrue(all(doc['selected_test_ids'] for doc, _ in docs))
+      self.assertCountEqual([target for doc, _ in docs for target in doc['targets']], targets)
+      self.assertCountEqual([identity for doc, _ in docs for identity in doc['selected_test_ids']], full)
+      verdict = self.check(copy.deepcopy(docs), count=count)
+      self.assertEqual(verdict['cases_passed'], len(full))
+      self.assertEqual(verdict['cases_total'], len(full))
+      self.assertEqual(verdict['cases_skipped'], 0)
+      mutations = {
+        'missing_artifact': lambda d: d.pop(),
+        'duplicate_artifact': lambda d: d.append(copy.deepcopy(d[0])),
+        'duplicate_index': lambda d: d[0][0]['shard'].update(index=1),
+        'wrong_count': lambda d: d[0][0]['shard'].update(count=4),
+        'wrong_target': lambda d: d[0][0].update(targets=['foreign.py']),
+        'missing_case': lambda d: d[0][1]['results'].pop(),
+        'duplicate_case': lambda d: d[0][1]['results'].append(d[0][1]['results'][0]),
+        'cancelled_shard': lambda d: d[0][0].update(status='cancelled'),
+        'cancelled_case': lambda d: d[0][1]['results'][0].update(status='cancelled'),
+        'debug_registry': lambda d: d[0][0]['native']['safety'].update(variant='ALLOW_DEBUG'),
+        'all_output_hook': lambda d: d[0][0]['native']['safety'].update(all_output_hook_available=True),
+        'source_revision': lambda d: d[0][0]['source'].update(revision='foreign'),
+        'build_head': lambda d: d[0][0]['source']['build_identity']['inputs'].update(revision='foreign'),
+        'build_tree': lambda d: d[0][0]['source']['build_identity']['inputs'].update(tree='foreign'),
+        'tracked_input': lambda d: d[0][0]['source']['build_identity']['inputs']['tracked_inputs']['source.c'].update(sha256='f' * 64),
+        'build_recipe': lambda d: d[0][0]['source']['build_identity']['inputs']['recipes']['SConstruct'].update(sha256='f' * 64),
+        'source_after': lambda d: d[0][0]['source_after'].update(inputs='changed'),
+        'native_output_after': lambda d: d[0][0]['source_after']['build_identity']['declared_outputs'].clear(),
+        'can_origin': lambda d: d[0][0]['native']['can_sources'].clear(),
+        'capnp_origin': lambda d: d[0][0]['native']['pycapnp'].update(sha256='f' * 64),
+      }
+      for name, mutate in mutations.items():
+        bad = json.loads(json.dumps(docs))
+        mutate(bad)
+        with self.subTest(mutation=name), self.assertRaises(ValueError):
+          self.check(bad, count=count)
+      with self.assertRaises(ValueError):
+        self.check(copy.deepcopy(docs), count=4)
+      for remove in (False, True):
+        def alter_native(root, remove=remove):
+          artifact = root / '0/libsafety-debug.so'
+          if remove:
+            artifact.unlink()
+          else:
+            artifact.write_bytes(b'foreign native bytes')
+        with self.subTest(count=count, remove_native=remove), self.assertRaises(ValueError):
+          self.check(copy.deepcopy(docs), alter_native, count=count)
+      skipped = copy.deepcopy(docs)
+      skipped[0][1]['results'][0].update(status='skipped', detail='Existing RELEASE skip')
+      verdict = self.check(skipped, count=count)
+      self.assertEqual(verdict['cases_passed'], len(full) - 1)
+      self.assertEqual(verdict['cases_skipped'], 1)
+      self.assertEqual(verdict['cases_total'], len(full))
+
+  def test_release_workflow_binds_sixteen_workers_to_runner_and_aggregate(self):
+    workflow = Path(__file__).parents[3].joinpath('.github/workflows/tests.yaml').read_text()
+    jobs = dict(re.findall(r'^  ([a-z_]+):\n(.*?)(?=^  [a-z_]+:|\Z)', workflow, re.M | re.S))
+    workers = jobs['safety_release_contracts']
+    aggregate = jobs['safety_release_complete']
+    self.assertIn('shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]', workers)
+    self.assertRegex(workers, r'--suite safety-release .*--shard-count 16\b')
+    self.assertRegex(aggregate, r'--suite safety-release --count 16\b')
+    self.assertIn('    timeout-minutes: 35\n', workers)
+    self.assertIn('      timeout-minutes: 20\n', workers)
+    self.assertIn('      fail-fast: false\n', workers)
+    self.assertIn('    needs: safety_release_contracts\n', aggregate)
+
   def test_complete_catalog_passes(self):
     self.assertEqual(self.check(self.documents())['cases_passed'], 12)