Repair CI contracts and preview rendering

openpilot/starpilot/flm/event_mean_evidence.py

@@ -1,7 +1,7 @@
 """Explicit event-only paired-mean evidence model; never native authority.
 
 Coverage is conditional on independent, approximately normal ORIGINAL whole-
-block gains. Source disjointness is checked; it does not establish independence.
+block gains. Source non-overlap is checked; it does not establish independence.
 """
 from dataclasses import asdict, dataclass
 import hashlib

openpilot/starpilot/flm/live_tracking_actor.py

@@ -884,7 +884,7 @@ class LiveTrackingActor:
         offer = optimizer.offer(now_ns, publication=False)
         if offer is None:
           raise ValueError('current trial publication unavailable')
-        self.ingress.latest_offer = offer
+        cast(LiveTrackingIngress, self.ingress).latest_offer = offer
       validate_response_witness(offer.response_witness, self.certificate.binding, now_ns)
       return offer
     except (AttributeError, TypeError, ValueError):

openpilot/starpilot/flm/live_tracking_bootstrap_producer.py

@@ -3,6 +3,8 @@
 Slot counts are availability, never an identification/admission declaration.
 The child still verifies the complete authentic prefix/native/source experiment.
 """
+from typing import cast
+
 
 import base64
 from dataclasses import asdict
@@ -368,7 +370,7 @@ class ResponseSnapshotProcess:
     eof = False
     for _ in range(16):
       try:
-        part = os.read(self.fd, 65536)
+        part = os.read(cast(int, self.fd), 65536)
       except BlockingIOError:
         break
       if not part:
@@ -384,7 +386,7 @@ class ResponseSnapshotProcess:
     # Drain the bounded remainder after exit; never accept partial output.
     while not eof and len(self.output) <= 8 * 1024 * 1024:
       try:
-        part = os.read(self.fd, 65536)
+        part = os.read(cast(int, self.fd), 65536)
       except BlockingIOError:
         break
       if not part:
@@ -405,7 +407,7 @@ class ResponseSnapshotProcess:
         result = json.loads(self.output)
       except (ValueError, UnicodeError):
         pass
-    os.close(self.fd)
+    os.close(cast(int, self.fd))
     self.pid = self.fd = None
     self.output.clear()
     return result
@@ -416,7 +418,7 @@ class ResponseSnapshotProcess:
     self.cancel(reason)
     if self.pid is not None:
       os.waitpid(self.pid, 0)
-      os.close(self.fd)
+      os.close(cast(int, self.fd))
       self.pid = self.fd = None
       self.output.clear()
     return True

openpilot/starpilot/flm/live_tracking_episodes.py

@@ -823,7 +823,7 @@ class TrackingEpisodeBuilder:
     Optional counterfactual probes are not actual actuation permission and
     cannot delay an uncensored baseline or a later measured keep/revert score.
     """
-    if len(self.ready) + len(self.completed) > self.ready.maxlen:
+    if len(self.ready) + len(self.completed) > cast(int, self.ready.maxlen):
       self.interrupt('measured_episode_backlog')
       return False
     for key, row in tuple(self.completed.items()):

openpilot/starpilot/flm/tests/test_live_tracking.py

@@ -3,6 +3,8 @@
 Real controller probes, route joins, owner application and bootstrap remain
 independent qualification. Synthetic certificate contents test the protocol.
 """
+from typing import Any
+
 
 from dataclasses import replace
 import hashlib
@@ -327,7 +329,7 @@ def test_actual_probe_coordinator_only_suspends_work_for_measurement_only_discov
 
   calls = []
   probe = SimpleNamespace(child=None, poll_probe=lambda **kwargs: None, cancel=lambda reason: calls.append(('cancel_probe', reason)))
-  actor = LiveTrackingActor(probe_process=probe)
+  actor: Any = LiveTrackingActor(probe_process=probe)
   owner = optimizer()
   owner.discovery = {'recovery': recovery}
   actor.ingress = SimpleNamespace(optimizer=owner)
@@ -346,7 +348,7 @@ def test_actual_ingress_retains_required_native_endpoint_work_during_recovery(re
   from openpilot.starpilot.flm.live_tracking_ingress import LiveTrackingIngress
 
   # Scheduling seam only: no source/native/controller admission is asserted.
-  obj = object.__new__(LiveTrackingIngress)
+  obj: Any = object.__new__(LiveTrackingIngress)
   obj.last_pose_ns = 0
   obj.binding = fixture_certificate().binding
   obj.replay = None

openpilot/starpilot/flm/tests/test_live_tracking_actor.py

@@ -62,14 +62,41 @@ def test_changed_source_waits_for_actual_old_probe_reap_before_bootstrap(monkeyp
   new = replace(old, source_epoch='9'*64)
   bootstrap, probe = PollProcess(), PollProcess(reap_after=3)
   actor = LiveTrackingActor(bootstrap, probe_process=probe)
+  # Lifecycle seam: preserve the current asynchronous preparation/poll contract.
+  # Response/controller/native admission is covered by the separate real collector tests.
+  class SnapshotPreparation:
+    pid = epoch = token = pending = None
+    reason = 'snapshot_submitted'
+    submissions = []
+
+    def cancel(self, reason):
+      self.pid = self.pending = None
+
+    def poll(self, *, token, now_ns):
+      result, self.pending = self.pending, None
+      self.pid = None
+      return result if token == self.token else None
+
+    def submit(self, collector, seed, *, token, now_ns):
+      assert probe.child is None  # Never start new snapshot work before the old probe is reaped.
+      self.epoch, self.token, self.pid = collector.original_source_capture.epoch, token, 1
+      self.submissions.append((token, now_ns))
+      self.pending = {'snapshot': live_tracking_bootstrap_producer.response_snapshot(collector, seed, now_ns)}
+      return True
+
+  snapshot = SnapshotPreparation()
+  from typing import Any, cast
+  actor.snapshot_process = cast(Any, snapshot)
+  monkeypatch.setattr(actor, '_same_bootstrap_source', lambda source, now_ns: True)
   actor.seed = old
   monkeypatch.setattr(actor, 'binding_for', lambda collector, source: new)
   collector = SimpleNamespace(ready=[None], original_source_capture=SimpleNamespace(epoch=1), live_tracking=None)
   monkeypatch.setattr(live_tracking_bootstrap_producer, 'response_snapshot', lambda *args: {'lifecycle_test_only': True})
   assert actor.tick(collector, object(), now_ns=10, current_native=True) is None
-  assert probe.child is not None and not bootstrap.submissions
+  assert probe.child is not None and not bootstrap.submissions and not snapshot.submissions
   assert actor.tick(collector, object(), now_ns=20, current_native=True) is None
-  assert probe.child is None
+  assert probe.child is None and len(snapshot.submissions) == 1 and not bootstrap.submissions
+  assert actor.tick(collector, object(), now_ns=30, current_native=True) is None
   assert len(bootstrap.submissions) == 1
   assert all(token is None for token, _ in probe.polls)
 

openpilot/starpilot/flm/tests/test_live_tracking_bootstrap_cohorts.py

@@ -1,4 +1,6 @@
 """Outcome-blind availability tests, not original/native calibration proof."""
+from typing import Any
+
 
 from dataclasses import replace
 from types import SimpleNamespace
@@ -158,7 +160,7 @@ def test_one_real_submission_per_epoch_even_after_response_failure(monkeypatch):
 
   b = certificate().binding
   process = PollProcess()
-  actor = LiveTrackingActor(process)
+  actor: Any = LiveTrackingActor(process)
   monkeypatch.setattr(actor, 'binding_for', lambda collector, source: b)
   collector = SimpleNamespace(ready=[None], original_source_capture=SimpleNamespace(epoch=1), live_tracking=None)
   calls = []

openpilot/starpilot/flm/tests/test_live_tracking_daemon.py

@@ -235,6 +235,8 @@ def test_actual_nonlive_run_does_not_read_tracking_poll_metadata(monkeypatch):
   now = [BASE-50_000_000]
 
   class SM:
+    # Ordinary SubMaster delivery metadata is required; no tracking-service metadata exists.
+    updated = {'deviceMotion': False}
     def update(self, timeout=None):
       now[0] += 50_000_000
       if now[0] > BASE+100_000_000:

openpilot/starpilot/flm/tests/test_live_tracking_history_recovery.py

@@ -1,4 +1,6 @@
 """Actual historical CAN maintenance never grants a live learning permission."""
+from typing import Any
+
 from dataclasses import replace
 from types import SimpleNamespace
 from unittest.mock import Mock
@@ -21,7 +23,7 @@ def test_stalled_live_collection_still_decodes_original_native_history(current_n
   old_size = len(replay.records)
   cutoff = replay.records[-1].control_ns - 4_000_000_000
   stamp, _ = replay.checkpoints[0]
-  actor = LiveTrackingActor(source_tracking=True)
+  actor: Any = LiveTrackingActor(source_tracking=True)
   actor.ingress = SimpleNamespace(interrupt=Mock())
   actor.certificate = SimpleNamespace(binding=SimpleNamespace(identity=replay.identity, baseline_revision=replay.baseline_revision))
   collector = SimpleNamespace(adapter=SimpleNamespace(replay=replay), ready=[])
@@ -41,7 +43,7 @@ def test_stalled_legacy_bootstrap_keeps_original_history_policy():
   replay = fixture('starpilot', two_windows=True).engine
   history = replay.emissions
   history.decode_queue.extend(history.events)
-  actor = LiveTrackingActor(source_tracking=False)
+  actor: Any = LiveTrackingActor(source_tracking=False)
   collector = SimpleNamespace(adapter=SimpleNamespace(replay=replay), ready=[])
   before = tuple(history.decode_queue), tuple(replay.records)
   assert actor.tick(collector, None, now_ns=replay.records[-1].control_ns, current_native=False) is None

openpilot/starpilot/flm/tests/test_live_tracking_requalification.py

@@ -1,4 +1,6 @@
 """Synthetic numerical/protocol tests; never actual native or road proof."""
+from typing import Any
+
 
 from dataclasses import replace
 from openpilot.starpilot.flm.runtime_adapter_registry import digest
@@ -564,7 +566,7 @@ def test_unfinished_restore_retries_only_input_scope_with_no_monitor_or_permissi
   local, _ = owner.qualified(10_350_000_000)
   assert not feed(owner, 8, command=0.020)
   assert owner.reason == 'response_command_outside_identified_range'
-  h = _LiveHistoryRestore(())
+  h: Any = _LiveHistoryRestore(())
   h.prior = SimpleNamespace(certificate=old, response_basis=basis(old))
   h.requalifier = owner
   h.offer = object()
@@ -607,7 +609,7 @@ def test_restore_scope_retry_never_reclassifies_faults_as_input_departure(reason
   from types import SimpleNamespace
   from openpilot.starpilot.flm.steerlearnerd import _LiveHistoryRestore
 
-  h = _LiveHistoryRestore(())
+  h: Any = _LiveHistoryRestore(())
   h.prior = object()
   h.requalifier = object()
   actor = SimpleNamespace(

openpilot/starpilot/flm/tests/test_live_tracking_retained_drift.py

@@ -1,4 +1,6 @@
 """Actual typed retainer/history path on synthetic qualified physical evidence."""
+from typing import Any
+
 from dataclasses import replace
 from types import SimpleNamespace
 import pytest
@@ -20,7 +22,7 @@ def setup(monkeypatch):
     '7'*64, 1, old, response_basis=basis(old))
   history = _LiveHistoryRestore((prior,))
   process = PollProcess()
-  actor = LiveTrackingActor(process)
+  actor: Any = LiveTrackingActor(process)
   class SnapshotPreparation:
     pid=epoch=token=None
     reason='snapshot_submitted'

openpilot/starpilot/galaxy/tests/test_flm_tracking_results.mjs

@@ -183,7 +183,7 @@ assert.match(FlmPage.template, /tracking\?\.label \|\| summary/)
 assert.match(FlmPage.template, /Future trial episodes/)
 assert.match(FlmPage.template, /Passing future checks does not itself confirm/)
 assert.doesNotMatch(FlmPage.template, /v-html/)
-console.log('Optional versioned FLM tracking report, paired observed metrics, honest trial acknowledgements and old report compatibility passed')
+console.log('Optional versioned FLM tracking report, paired observed metrics, honest trial acknowledgments and old report compatibility passed')
 
 const render = compile(FlmPage.template, { onError: e => { throw e }, decodeEntities: v => v.replaceAll('&', '&') })
 const ctx = { ...FlmPage.data(), ...FlmPage.methods, mode:'local', embedded:true, unauthorized() {},

openpilot/starpilot/soloist/bugsink_report.py

@@ -80,6 +80,7 @@ def windows(lines: list[str]) -> list[dict]:
       found.append(current)
     current['end'] = when
     if kind == 'rtp':
+      assert lost is not None
       current['rtp_warnings'] += 1
       current['lost_packets'] += int(lost[1])
     elif kind == 'underrun':

openpilot/starpilot/speed_limits/acceptance.py

@@ -1,4 +1,6 @@
 """Immutable acceptance decisions with explicit authority and no runtime I/O."""
+from typing import cast
+
 
 import math
 from dataclasses import dataclass, replace
@@ -201,7 +203,7 @@ def _identity_valid(identity: object) -> bool:
 
 def _speed_valid(value: object) -> bool:
   try:
-    return type(value) in (int, float) and math.isfinite(value) and value > 0
+    return type(value) in (int, float) and math.isfinite(cast(float, value)) and cast(float, value) > 0
   except OverflowError:
     return False
 

openpilot/starpilot/speed_limits/overrides.py

@@ -1,4 +1,6 @@
 """Pure SLC override contribution from qualified driver intent and current pedal evidence."""
+from typing import cast
+
 
 import math
 from dataclasses import dataclass, replace
@@ -259,7 +261,7 @@ def step(state: State, acceptance: acc.Decision, authority: acc.Authority, polic
     state = replace(state, persistent_selected_mps=None, applied_selected_mps=None)
   comparison_target = state.applied_selected_mps + selected_delta if state.applied_selected_mps is not None else effective_target
   if (not domain_unavailable and state.persistent_selected_mps is not None and effective_target is not None and
-      selected.speed_mps is not None and math.isclose(comparison_target, selected.speed_mps + selected_delta, abs_tol=0.001, rel_tol=0)):
+      selected.speed_mps is not None and math.isclose(cast(float, comparison_target), selected.speed_mps + selected_delta, abs_tol=0.001, rel_tol=0)):
     state = replace(state, persistent_selected_mps=None)
   state = replace(state, accepted_mps=accepted_speed, effective_target_mps=effective_target,
                   mode=authority.mode, owner=authority.owner)

openpilot/starpilot/speed_limits/tests/test_toyota_pcm_cruise.py

@@ -2,17 +2,21 @@
 from unittest import mock
 from types import SimpleNamespace
 from collections import deque
+from typing import cast
 
 import pytest
 
 from opendbc.can import CANPacker
 from opendbc.car import Bus, structs
+from opendbc.car.interfaces import CarInterfaceBase
 from opendbc.car.toyota.interface import CarInterface
 from opendbc.car.toyota.values import CAR, DBC
 from openpilot.selfdrive.car.cruise import VCruiseHelper
 from openpilot.selfdrive.car.card import Car
 from openpilot.cereal import messaging
+from openpilot.common.realtime import Ratekeeper
 from openpilot.selfdrive.controls.plannerd import queue_cruise_event, current_cruise_event
+from openpilot.starpilot.speed_limits.runtime import Action, Runtime
 from openpilot.starpilot.speed_limits.runtime_settings import parse
 from openpilot.starpilot.speed_limits.tests import test_runtime_replay as replay_fixture
 
@@ -61,9 +65,9 @@ def publish_change(parsed, state):
   host = Car.__new__(Car)
   host.CP = parsed.cp
   host.car_params_published = True
-  host.sm = SimpleNamespace(frame=1, all_checks=lambda services: True)
+  host.sm = cast(messaging.SubMaster, SimpleNamespace(frame=1, all_checks=lambda services: True))
   host.last_actuators_output = structs.CarControl.Actuators()
-  host.rk = SimpleNamespace(remaining=0.)
+  host.rk = cast(Ratekeeper, SimpleNamespace(remaining=0.))
   host.can_rcv_cum_timeout_counter = 0
   host.aol_replay = host.curve_replay = False
   host.slc_replay = True
@@ -73,7 +77,7 @@ def publish_change(parsed, state):
   host.slc_receipts = []
   host.pm = mock.Mock()
   host.car_gps_publisher = mock.Mock()
-  host.CI = SimpleNamespace(CS=SimpleNamespace())
+  host.CI = cast(CarInterfaceBase, SimpleNamespace(CS=SimpleNamespace()))
   host.radar_publish = mock.Mock()
   # This is the same selected-speed copy normally made by Card.state_update.
   state.vCruise = parsed.helper.v_cruise_kph
@@ -107,9 +111,15 @@ def test_pcm_resume_edge_and_release_frame_set_speed_feed_override():
   fixture.setUp()
   sm = fixture.sm
   sm['carState'].vCruise = sm['carState'].vCruiseCluster = 90.
-  fixture.cp = parsed.cp
-  runtime = fixture.accepted_runtime(parse({'SpeedLimitController': True, 'SLCPriority1': 'Dashboard'}), session_id='pcm-drive')
-  first = runtime.step(sm, parsed.cp, now_ns=2_000_000_000)
+  runtime = Runtime(parse({'SpeedLimitController': True, 'SLCPriority1': 'Dashboard', 'SLCConfirmation': True}),
+                    session_id='pcm-drive')
+  pending = runtime.step(sm, parsed.cp, now_ns=2_000_000_000)
+  assert pending.message.slcState.hasPending
+  sm.advance(2_010_000_000)
+  first = runtime.step(sm, parsed.cp, now_ns=2_010_000_000,
+                       request=Action(runtime.session_id, 1, pending.message.slcState.decisionId,
+                                      pending.message.slcState.presentationId, 'accept'))
+  assert first.result.acceptance is not None and first.result.acceptance.basis == 'driver_accept'
   assert first.result.override.context is not None
   receipt, published = publish_change(parsed, state)
   sm.advance(parsed.now)

openpilot/starpilot/system/android_auto/projection_map_lease.py

@@ -33,7 +33,8 @@ class MapLeaseWriter:
 
 def projection_map_active(path=DEFAULT_PATH, now=None):
   try:
-    age = (time.time() if now is None else now) - os.stat(path).st_mtime
+    # Filesystem mtimes use the realtime clock; monotonic time has a different epoch.
+    age = (time.time() if now is None else now) - os.stat(path).st_mtime  # noqa: TID251
   except OSError:
     return False
   return 0 <= age <= MAX_AGE_S

openpilot/starpilot/system/android_auto/tests/test_car_interface.py

@@ -1,3 +1,4 @@
+from typing import Any
 import os
 import subprocess
 import sys
@@ -216,9 +217,9 @@ class FakeNM:
 
 
 def nm_lease(nm, dedicated=True, monkeypatch=None):
-  lease = NetworkLease(Log(), "aa0", dedicated=dedicated)
+  lease: Any = NetworkLease(Log(), "aa0", dedicated=dedicated)
   lease._call, lease._get, lease._set = nm.call, nm.get, nm.set_property
-  lease._delete_stale_profiles = lambda: None  # ty: ignore[invalid-assignment]  # test double
+  lease._delete_stale_profiles = lambda: None  # test double
   lease.router = type("R", (), {"close": lambda self: None})()
   return lease
 

openpilot/starpilot/system/android_auto/tests/test_projection_onroad_boundary.py

@@ -1,5 +1,7 @@
 """CPU-only ownership and cleanup checks for the projection-only road view."""
 
+from typing import Any, cast
+
 import ast
 from dataclasses import dataclass
 import importlib.util
@@ -373,7 +375,7 @@ class TestProjectionOnroad(unittest.TestCase):
     real = {'alert': Alert()}
     native.alert, native.alert_size = Alert, SimpleNamespace(NONE='none', SMALL='small')
     native.adapter = lambda *_: SimpleNamespace(build=lambda *_a, **_k: SimpleNamespace(
-      onroad=OnroadState(False, False, None, None, SpeedLimitObservation(), alert=real['alert'])))
+      onroad=OnroadState(False, False, None, None, SpeedLimitObservation(), alert=cast(Any, real['alert']))))
     view = projection.ProjectionOnroad(dependencies=native, certificate_days=3)
     shown = []
     view.onroad.render = lambda state: shown.append(state.alert.text1)

openpilot/starpilot/tests/test_gm_pitch_preferences.py

@@ -1,3 +1,4 @@
+from typing import cast
 import tempfile
 from pathlib import Path
 from types import SimpleNamespace
@@ -108,7 +109,7 @@ class TestGMPitchStartupAndSettings(unittest.TestCase):
           self.assertTrue(owner.apply(row_default(row())))
           self.assertFalse(self.saved.get_bool(key))
           controller = SimpleNamespace(volt_sng=False, gm_auto_hold=False, volt_one_pedal=False, long_pitch=True)
-          VehicleStartupPreferences(**{preference: True}).configure_controller(SimpleNamespace(CP=cp, CC=controller))
+          VehicleStartupPreferences(**cast(dict, {preference: True})).configure_controller(SimpleNamespace(CP=cp, CC=controller))
           self.assertFalse(getattr(controller, preference))
           fresh = row()
           pending = row_change(fresh)

openpilot/starpilot/ui/layout_preview_renderer.py

@@ -1,5 +1,7 @@
 """Parked onroad layout preview using the same native view as the driving UI."""
 
+from typing import cast
+
 from pathlib import Path
 import json
 import math
@@ -20,6 +22,7 @@ from openpilot.starpilot.ui.onroad_customization import (CAMERA_WIDGETS, PROFILE
 from openpilot.starpilot.ui.onroad_widget_style import draw_widget_frame
 from openpilot.starpilot.curve_speed.status import CurveObservation
 from openpilot.starpilot.ui.onroad_state import ConditionalLead
+from openpilot.starpilot.ui.unified_speed_presentation import resolve_unified_speed
 from openpilot.starpilot.ui.onroad_state import AlertSize, ObservationKind, OnroadAlert, OnroadState, SourceReading, SpeedLimitObservation
 from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice, OnroadAppearance
 from openpilot.starpilot.ui.onroad_dm import draw_monitor_art, monitor_bounds, monitor_visible
@@ -225,7 +228,8 @@ class _Canvas:
         self.map = MapOverlay(reader=SampleTileReader(), fonts=self.fonts)
         self.view.map_layer = self._render_map
         from openpilot.starpilot.models.jetlink_status import JetlinkStatus
-        self.view.jetlink_status.status = JetlinkStatus('iPhone', 'active', 24.6, 'connected')
+        from openpilot.starpilot.ui.onroad_jetlink import JetlinkStatusWidget
+        cast(JetlinkStatusWidget, self.view.jetlink_status).status = JetlinkStatus('iPhone', 'active', 24.6, 'connected')
       self.monitor = _DriverMonitorArt(profile)
       self.view.driver_monitor_layer = self._render_driver_monitor
       self.view.preview_pip_layer = self._render_preview_pip
@@ -335,6 +339,10 @@ class _Canvas:
     captures = []
     for key in RUNTIME_PREVIEW_WIDGETS[str(self.profile)]:
       logical_width, logical_height = widget_size(state.customization, self.profile, key)
+      # The restored compact US sign adds an offset row beyond its saved base size.
+      if (self.profile == Profile.COMPACT and key == 'speed_limit' and
+          not state.appearance.use_vienna_sign and resolve_unified_speed(state).offset_text):
+        logical_height += 44
       position = positions[key]
       widget = PROFILES[str(self.profile)]['widgets'][key]
       crop_x = max(0, math.floor(position['x'] - widget.get('visualInsetLeft', 0)))

openpilot/starpilot/ui/onroad.py

@@ -9,6 +9,7 @@ LICENSE, while the application owns all state and action transport.
 from collections.abc import Callable
 from datetime import datetime
 from pathlib import Path
+from typing import Protocol
 import hashlib
 import json
 
@@ -107,8 +108,13 @@ def axis_status_color(state: OnroadState) -> rl.Color:
   return rl.Color(18, 40, 57, 255)
 
 
+class JetlinkRenderLayer(Protocol):
+  def render(self, state: OnroadState) -> None: ...
+
+
 class OnroadView:
   preview_pip_layer: PipLayer | None
+  jetlink_status: JetlinkRenderLayer
 
   def __init__(self, fonts: BitmapFonts, asset_directory: Path, *, camera_layer: CameraLayer | None = None,
                extra_overlays: OverlayLayer | None = None, pip_layer: PipLayer | None = None,

openpilot/starpilot/ui/runtime_app.py

@@ -67,6 +67,7 @@ from openpilot.starpilot.ui.startup import slc_action_transport_enabled, validat
 from openpilot.starpilot.speed_limits.vision_gate import diagnostic_choice_enabled
 from openpilot.starpilot.ui.bluetooth_status import BluetoothStatusSource
 from openpilot.starpilot.ui.runtime_snapshot import RuntimeSnapshotAdapter, current_message, display_message
+from openpilot.starpilot.ui.runtime_software import deliver_software
 from openpilot.starpilot.ui.network_panel import NetworkPanelBridge
 from openpilot.starpilot.ui.slc_action_dispatch import SlcActionDispatcher
 from openpilot.starpilot.ui.onroad_state import AlertSize, BorderSignals, OnroadRequest, OnroadState, SlcUiRequest, is_personality_notice, is_lane_change_notice
@@ -1992,31 +1993,8 @@ class StarMainLayout(MainLayout):
     return True
 
   def _deliver_software(self, request: SoftwareAction) -> bool:
-    from openpilot.selfdrive.ui.layouts.settings.settings import PanelType
-    fresh = self.star.adapter.build(ShellMode.SETTINGS, Destination.SOFTWARE, now_ns=time.monotonic_ns()).software
-    if request.request not in fresh.available_actions:
-      return False
-    owner = self._panel_owner(PanelType.SOFTWARE)
-    if request.request in (SoftwareRequest.CHECK_FOR_UPDATES, SoftwareRequest.DOWNLOAD_UPDATE):
-      expected = DownloadLabel.DOWNLOAD if request.request == SoftwareRequest.DOWNLOAD_UPDATE else DownloadLabel.CHECK
-      owner._update_state()
-      if (fresh.download_label != expected or owner._waiting_for_updater or
-          not owner._download_btn.action_item.enabled):
-        return False
-      from openpilot.system.ui.lib.multilang import tr
-      if owner._download_btn.action_item.text != tr(expected.value):
-        return False
-      owner._on_download_update()
-    elif request.request in (SoftwareRequest.FAST_UPDATE, SoftwareRequest.ROLLBACK):
-      owner._update_state()
-      owner._confirm_direct_update("rollback" if request.request == SoftwareRequest.ROLLBACK else "fast")
-    elif request.request == SoftwareRequest.OPEN_BRANCH_CHOOSER:
-      owner._on_select_branch(self._confirmed_offroad)
-    elif request.request == SoftwareRequest.OPEN_UNINSTALL_CONFIRMATION:
-      owner._on_uninstall(self._confirmed_offroad)
-    else:
-      return False
-    return True
+    return deliver_software(self, request, shell_mode=ShellMode, destination=Destination, clock=time,
+                            software_request=SoftwareRequest, download_label=DownloadLabel)
 
   def _render_main_content(self) -> None:
     if self._current_mode == MainState.ARABESQUE or self._current_mode == MainState.HOME and ui_state.is_body:

openpilot/starpilot/ui/runtime_software.py

@@ -0,0 +1,34 @@
+"""Native software request routing for the StarPilot runtime shell."""
+
+from typing import Any
+
+from openpilot.starpilot.ui.software_state import SoftwareAction
+
+
+def deliver_software(self: Any, request: SoftwareAction, *, shell_mode, destination, clock, software_request, download_label) -> bool:
+  from openpilot.selfdrive.ui.layouts.settings.settings import PanelType
+  fresh = self.star.adapter.build(shell_mode.SETTINGS, destination.SOFTWARE, now_ns=clock.monotonic_ns()).software
+  if request.request not in fresh.available_actions:
+    return False
+  owner = self._panel_owner(PanelType.SOFTWARE)
+  if request.request in (software_request.CHECK_FOR_UPDATES, software_request.DOWNLOAD_UPDATE):
+    expected = download_label.DOWNLOAD if request.request == software_request.DOWNLOAD_UPDATE else download_label.CHECK
+    owner._update_state()
+    if (fresh.download_label != expected or owner._waiting_for_updater or
+        not owner._download_btn.action_item.enabled):
+      return False
+    from openpilot.system.ui.lib.multilang import tr
+    if owner._download_btn.action_item.text != tr(expected.value):
+      return False
+    owner._on_download_update()
+  elif request.request in (software_request.FAST_UPDATE, software_request.ROLLBACK):
+    owner._update_state()
+    owner._confirm_direct_update("rollback" if request.request == software_request.ROLLBACK else "fast")
+  elif request.request == software_request.OPEN_BRANCH_CHOOSER:
+    owner._on_select_branch(self._confirmed_offroad)
+  elif request.request == software_request.OPEN_UNINSTALL_CONFIRMATION:
+    owner._on_uninstall(self._confirmed_offroad)
+  else:
+    return False
+  return True
+

openpilot/starpilot/ui/speed_limit_widget.py

@@ -139,6 +139,8 @@ class SpeedLimitWidget:
                                               (shown.posted_text, 50 if len(shown.posted_text) <= 2 else 42, 66, FontRole.BOLD)):
           measured = self.fonts.measure(label, role, size)
           origin = rect.x + (rect.width - measured.width) / 2
+          if arrow_direction and role == FontRole.SEMI_BOLD:
+            origin -= 6  # Keep four pixels between the measured header ink and the pending arrow.
           self.fonts.draw(label, role, size, origin, rect.y + label_top, white)
           if label == shown.posted_text:
             number = origin, rect.y + label_top, origin + measured.width, rect.y + label_top + measured.height

openpilot/starpilot/ui/tests/test_compact_sign_height.py

@@ -1,5 +1,5 @@
 from dataclasses import replace
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -18,17 +18,18 @@ def test_only_compact_rectangle_grows_for_visible_offset(profile, offset):
   widget.profile = profile
   widget.scale = 1
   widget.text_scale = 1
-  widget.pulse = type('Pulse', (), {'color': staticmethod(lambda color, now: color)})()
+  widget.pulse = Mock(color=lambda color, now: color)
   bounds = speed_limit_bounds(profile, current)
+  assert bounds is not None
   with patch('openpilot.starpilot.ui.speed_limit_widget.rl.draw_rectangle_rounded_lines_ex') as outline, \
        patch('openpilot.starpilot.ui.speed_limit_widget.rl.draw_rectangle_rounded'), \
        patch('openpilot.starpilot.ui.speed_limit_widget.SpeedLimitShapes.arrow'):
-    from unittest.mock import Mock
     widget.shapes = Mock()
     widget._sign(current, bounds, 10.)
   rectangle = outline.call_args.args[0]
   assert rectangle.height == ((160 if offset else 116) if profile == Profile.COMPACT else 128)
   other = speed_limit_bounds(profile, replace(current, show_slc_offset=not offset))
+  assert other is not None
   assert bounds[:3] == other[:3]
   assert abs(bounds[3] - other[3]) == (44 if profile == Profile.COMPACT else 0)
   texts = [call.args[0] for call in widget.fonts.draw.call_args_list]
@@ -36,16 +37,17 @@ def test_only_compact_rectangle_grows_for_visible_offset(profile, offset):
 
 
 @pytest.mark.parametrize('value', ['55', '125'])
-def test_compact_rectangle_uses_original_geometry_and_measured_typography(value):
-  from unittest.mock import Mock
+@pytest.mark.parametrize('pending_arrow', [False, True])
+def test_compact_rectangle_uses_original_geometry_and_measured_typography(value, pending_arrow):
   from openpilot.starpilot.ui.presentation import FontRole
   from openpilot.starpilot.ui.onroad_state import SpeedLimitObservation, ObservationKind
   current = state(show_slc_offset=False)
-  current = replace(current, speed_limit=SpeedLimitObservation(ObservationKind.VALID, 'map', int(value) / 3.6))
+  current = replace(current, speed_limit=SpeedLimitObservation(ObservationKind.VALID, 'map', int(value) / 3.6,
+    pending_speed_limit_mps=int(value) / 3.6 if pending_arrow else None))
   widget = object.__new__(SpeedLimitWidget)
   widget.fonts, widget.profile = fonts(Profile.COMPACT), Profile.COMPACT
   widget.scale, widget.text_scale = 1, 1
-  widget.pulse = type('Pulse', (), {'color': staticmethod(lambda color, now: color)})()
+  widget.pulse = Mock(color=lambda color, now: color)
   widget.shapes = Mock()
   with patch('openpilot.starpilot.ui.speed_limit_widget.rl.draw_rectangle_rounded_lines_ex') as outline:
     widget._sign(current, speed_limit_bounds(Profile.COMPACT, current), 10.)
@@ -56,4 +58,7 @@ def test_compact_rectangle_uses_original_geometry_and_measured_typography(value)
   for call, text, size, y, role in zip(calls, ['SPEED', 'LIMIT', value], [20, 20, 50 if len(value) <= 2 else 42],
                                         [38, 56, 86], [FontRole.SEMI_BOLD, FontRole.SEMI_BOLD, FontRole.BOLD], strict=True):
     measured = widget.fonts.measure(text, role, size)
-    assert call.args[:5] == (text, role, size, 332 + (116 - measured.width) / 2, y)
+    header_shift = 6 if pending_arrow and role == FontRole.SEMI_BOLD else 0
+    # Without an actual pending arrow, the original screenshot's centered origin is exact.
+    assert call.args[:5] == (text, role, size, 332 + (116 - measured.width) / 2 - header_shift, y)
+  assert widget.shapes.arrow.called is pending_arrow

openpilot/starpilot/ui/tests/test_dinish_speed.py

@@ -256,7 +256,7 @@ def test_compact_preview_sprite_keeps_the_sign_offset_arrow_and_override_cross()
         rl.unload_image(image)
       assert np.array_equal(direct[..., 3], np.asarray(rebuilt)[..., 3])
       assert slot['offsetX'] == slot['offsetY'] == 0
-      assert (slot['width'], slot['height']) == (118, 132)
+      assert (slot['width'], slot['height']) == (118, 176)
   finally:
     canvas.close()
     if owns_context:
@@ -309,6 +309,7 @@ def test_both_sign_styles_contain_text_and_arrows_at_each_profile_scale():
     for size in ('large', 'compact'):
       sx, sy, sw, sh = SPRITES[f'arrow_{size}']
       bbox = atlas.crop((sx, sy, sx+sw, sy+sh)).getchannel('A').point(lambda alpha: 255 if alpha > 40 else 0).getbbox()
+      assert bbox is not None
       glyphs[size] = (bbox[0]-sw/2, bbox[1]-sh/2, bbox[2]-sw/2, bbox[3]-sh/2)
   owns_context = not rl.is_window_ready()
   if owns_context:
@@ -324,6 +325,7 @@ def test_both_sign_styles_contain_text_and_arrows_at_each_profile_scale():
             (0, 350_000_000, 2_100_000_000), positions, (-1, 1)):
           if not vienna and posted > 145:
             continue
+          case = (profile, vienna, posted, offset, elapsed, saved_x, y, direction)
           current = state(observed_ns=10_000_000_000+elapsed)
           key = 'cruise_limits' if profile == Profile.LARGE else 'speed_limit'
           current.customization['layouts'][profile][key].update(x=saved_x, y=y)
@@ -344,36 +346,50 @@ def test_both_sign_styles_contain_text_and_arrows_at_each_profile_scale():
             if vienna:
               cx, cy = x+59*scale, y+67*scale
               assert all((px-cx)**2+(py-cy)**2 <= (47*scale)**2
-                         for px, py in product((box[0], box[2]), (box[1], box[3])))
-              assert (color.r, color.g, color.b, color.a) == (0, 0, 0, 255)
+                         for px, py in product((box[0], box[2]), (box[1], box[3]))), case
+              # The original compact no-offset numeral carries the manual confirmation pulse.
+              # Other Vienna layouts keep black numerals while pulsing the ring.
+              expected_color = ((170, 115, 235, 255)
+                                if profile == Profile.COMPACT and not offset and elapsed == 350_000_000
+                                else (0, 0, 0, 255))
+              assert (color.r, color.g, color.b, color.a) == expected_color, case
             else:
-              assert x+7*scale <= box[0] < box[2] <= x+111*scale
-              assert y+7*scale <= box[1] < box[3] <= y+125*scale
+              assert x+7*scale <= box[0] < box[2] <= x+111*scale, case
+              sign_height = 176 if profile == Profile.COMPACT and offset else 132
+              assert y+7*scale <= box[1] < box[3] <= y+(sign_height-7)*scale, case
             ink.append((text, box))
             nominal_size = size/widget.text_scale
             if role == FontRole.BOLD:
-              assert 31.99 <= nominal_size <= (48.01 if len(text)>2 else 54.01) if vienna else nominal_size == 58
+              if profile == Profile.COMPACT and not vienna:
+                assert nominal_size == (50 if len(text) <= 2 else 42), case
+              elif profile == Profile.COMPACT and vienna and not offset:
+                assert nominal_size == (58 if len(text) <= 2 else 48), case
+              else:
+                assert 31.99 <= nominal_size <= (48.01 if len(text)>2 else 54.01) if vienna else nominal_size == 58, case
             elif role == FontRole.MEDIUM:
-              assert 23.99 <= nominal_size <= 36.01 if vienna else nominal_size == 36
+              if profile == Profile.COMPACT and not vienna:
+                assert nominal_size == 30, case
+              else:
+                assert 23.99 <= nominal_size <= 36.01 if vienna else nominal_size == 36, case
           arrow_draw.assert_called_once()
           size, ax, ay, arrow_direction, _ = arrow_draw.call_args.args
-          assert size == ('compact' if vienna else 'large')
-          assert arrow_direction == direction
+          assert size == ('compact' if vienna else 'large'), case
+          assert arrow_direction == direction, case
           glyph = glyphs[size]
           left, top, right, bottom = glyph if direction>0 else (-glyph[2], -glyph[3], -glyph[0], -glyph[1])
           arrow = ax+left*scale, ay+top*scale, ax+right*scale, ay+bottom*scale
           if vienna:
             assert all((px-cx)**2+(py-cy)**2 <= (47*scale)**2
-                       for px, py in product((arrow[0], arrow[2]), (arrow[1], arrow[3])))
+                       for px, py in product((arrow[0], arrow[2]), (arrow[1], arrow[3]))), case
           else:
-            assert x+7*scale <= arrow[0] < arrow[2] <= x+111*scale
-            assert y+7*scale <= arrow[1] < arrow[3] <= y+125*scale
-            assert all(box[2]+4*scale <= arrow[0] for text, box in ink if text in ('SPEED', 'LIMIT'))
+            assert x+7*scale <= arrow[0] < arrow[2] <= x+111*scale, case
+            assert y+7*scale <= arrow[1] < arrow[3] <= y+125*scale, case
+            assert all(box[2]+4*scale <= arrow[0] for text, box in ink if text in ('SPEED', 'LIMIT')), (case, ink, arrow)
             assert arrow[3]+6*scale <= next(box[1] for text, box in ink if text==str(posted))
           ink.append(('arrow', arrow))
           for index, (_, a) in enumerate(ink):
             for _, b in ink[index+1:]:
-              assert not (a[0]<b[2] and a[2]>b[0] and a[1]<b[3] and a[3]>b[1])
+              assert not (a[0]<b[2] and a[2]>b[0] and a[1]<b[3] and a[3]>b[1]), (case, a, b)
         widget.close()
   finally:
     if owns_context:

openpilot/starpilot/vehicle_preferences.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from opendbc.car.gm.startup_preferences import prepare_disable_longitudinal
 from opendbc.car.toyota.interface import apply_toyota_auto_hold
@@ -19,18 +19,18 @@ def bolt_disable_supported(cp) -> bool:
 @dataclass(frozen=True)
 class VehicleStartupPreferences:
   toyota_auto_hold: bool = False
-  jeep_brake_hold: bool = False
+  jeep_brake_hold: bool = field(default=False, kw_only=True)
   turn_assist: bool = False
   gm_long_pitch: bool = True
   disable_bolt_long: bool = False
   honda_bosch_a_radar: bool = True
   tesla_preap_stock: bool = True
-  tesla_preap_radar: RadarSettings | None = None
-  tesla_preap_pedal: PedalSettings | None = None
-  subaru_avh_startup: bool = False
-  subaru_stop_start_off: bool = False
-  subaru_sng: bool = False
-  subaru_software_cruise: bool = False
+  tesla_preap_radar: RadarSettings | None = field(default=None, kw_only=True)
+  tesla_preap_pedal: PedalSettings | None = field(default=None, kw_only=True)
+  subaru_avh_startup: bool = field(default=False, kw_only=True)
+  subaru_stop_start_off: bool = field(default=False, kw_only=True)
+  subaru_sng: bool = field(default=False, kw_only=True)
+  subaru_software_cruise: bool = field(default=False, kw_only=True)
   volt_sng: bool = False
   gm_auto_hold: bool = False
   volt_one_pedal: bool = False

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "07517473d131cd3da0d81eb95a813b584150d0f6fb68a80a513a6c1cda5a312d", "schema_version": 1}
+{"protected_inventory_sha256": "25f884a9a6495f49cc4c52130811797f029a4f4ff31dd43626d0dc0c7fdae940", "schema_version": 1}

tools/ci/run_safety_qualification.py

@@ -92,6 +92,9 @@ def source_identity():
 
 def prerequisite_artifact_paths(prerequisites, name):
   paths = list(prerequisites)
+  if name in ("coverage", "mutation-full"):
+    if "msgq_repo/msgq/visionipc/visionipc_pyx.so" not in paths:
+      paths.append("msgq_repo/msgq/visionipc/visionipc_pyx.so")
   if name in ("coverage", "mutation-full"):
     generated = "openpilot/selfdrive/controls/lib/longitudinal_mpc_lib/c_generated_code/"
     extension = ".dylib" if sys.platform == "darwin" else ".so"
@@ -234,7 +237,7 @@ def complete_mutation_artifact(path, summary, *, digest_out=None):
     return False
 
 
-def check_import_origins(python, env, output):
+def check_import_origins(python, env, output, *, full_catalog=False):
   script = """
 import json
 from pathlib import Path
@@ -252,13 +255,20 @@ print(json.dumps({name: str(Path(module.__file__).resolve()) for name, module in
                    ('can_dbc', opendbc.can.dbc), ('cereal_log', cereal_log),
                    ('cereal_messaging', openpilot.cereal.messaging), ('msgq_ipc', msgq.ipc_pyx)]} |
                   {'panda': importlib.util.find_spec('panda').origin}))
+"""
+  if full_catalog:
+    script += """
+import msgq.visionipc.visionipc_pyx
+print(json.dumps({'msgq_visionipc': str(Path(msgq.visionipc.visionipc_pyx.__file__).resolve())}))
 """
   result = subprocess.run([python, "-c", script], cwd=output.parent, env=env, capture_output=True, text=True)
   output.write_text(result.stdout + result.stderr)
   if result.returncode:
     return False
   try:
-    origins = json.loads(result.stdout)
+    origins = {}
+    for line in result.stdout.splitlines():
+      origins.update(json.loads(line))
   except json.JSONDecodeError:
     return False
   expected = {"opendbc": ROOT / "opendbc_repo", "panda": ROOT / "panda", "msgq": ROOT / "msgq_repo",
@@ -266,12 +276,16 @@ print(json.dumps({name: str(Path(module.__file__).resolve()) for name, module in
               "can_parser": ROOT / "opendbc_repo", "can_packer": ROOT / "opendbc_repo",
               "can_dbc": ROOT / "opendbc_repo", "cereal_log": ROOT / "openpilot/cereal",
               "cereal_messaging": ROOT / "openpilot/cereal", "msgq_ipc": ROOT / "msgq_repo"}
+  if full_catalog:
+    expected["msgq_visionipc"] = ROOT / "msgq_repo"
   return all(isinstance(origins.get(name), str) and Path(origins[name]).is_relative_to(path)
              for name, path in expected.items())
 
 
 def imported_module_hashes(output):
-  origins = json.loads(output.read_text().splitlines()[0])
+  origins = {}
+  for line in output.read_text().splitlines():
+    origins.update(json.loads(line))
   return {name: {"path": path, "sha256": hashlib.sha256(Path(path).read_bytes()).hexdigest()}
           for name, path in origins.items()}
 
@@ -550,7 +564,8 @@ def main():
     prerequisite = f"openpilot/common/libparams_c{extension}"
     prerequisites = [prerequisite, "msgq_repo/msgq/ipc_pyx.so"]
     if args.gate in ("coverage", "mutation-full"):
-      # The full catalog includes actual Card consumers importing the longitudinal planner.
+      # The full catalog includes actual Card and SelfdriveD consumers.
+      prerequisites.append("msgq_repo/msgq/visionipc/visionipc_pyx.so")
       prerequisites.append("openpilot/selfdrive/controls/lib/longitudinal_mpc_lib/c_generated_code/acados_ocp_solver_pyx.so")
     built, _ = gate.run("build-imports", ["scons", "--minimal", "-j4", *prerequisites], timeout=3600)
     if built:
@@ -571,7 +586,7 @@ def main():
     for artifact in bound_artifacts:
       gate.hash_artifact(ROOT / artifact)
     prerequisite_artifacts = prerequisite_artifact_identity(bound_artifacts)
-    origins_ok = check_import_origins(gate.python, gate.env, output / "imports.log")
+    origins_ok = check_import_origins(gate.python, gate.env, output / "imports.log", full_catalog=args.gate in ("coverage", "mutation-full"))
     if origins_ok:
       imported_before = imported_module_hashes(output / "imports.log")
     version_codes = [gate.run("python-version", [gate.python, "--version"])[0],

tools/ci/tests/test_safety_qualification.py

@@ -44,6 +44,20 @@ class TestSafetyQualificationRunner(TestCase):
         else:
           self.assertNotIn("SAFETY_COVERAGE", gate.env)
 
+  def test_catalog_prerequisites_bind_visionipc_and_hash_all_import_lines(self):
+    for name in ("coverage", "mutation-full"):
+      paths = qualification.prerequisite_artifact_paths(["fixture.so"], name)
+      self.assertIn("msgq_repo/msgq/visionipc/visionipc_pyx.so", paths)
+      self.assertEqual(len(paths), len(set(paths)))
+    with tempfile.TemporaryDirectory() as td:
+      root = Path(td)
+      modules = {name: str(root / name) for name in ("ipc.so", "visionipc.so")}
+      for path in modules.values():
+        Path(path).write_bytes(b"native")
+      log = root / "imports.log"
+      log.write_text("\n".join(json.dumps({name: path}) for name, path in modules.items()) + "\n")
+      self.assertEqual(set(qualification.imported_module_hashes(log)), set(modules))
+
   def test_source_identity_includes_gate_and_workflow_inputs(self):
     identity = qualification.source_identity()
     self.assertFalse(any("/obj/" in path or "/gen/" in path for path in identity))
@@ -127,11 +141,12 @@ class TestSafetyQualificationRunner(TestCase):
           artifact.parent.mkdir(parents=True, exist_ok=True)
           artifact.write_bytes(b"built-native")
 
-        def execute_gate(case_name=name, case_root=root, case_artifacts=artifacts):
+        def execute_gate(case_name=name, case_root=root, case_artifacts=artifacts, solver_name="libacados_ocp_solver_long" + extension):
           if case_name == "native prerequisite changed during gate":
             (case_root / case_artifacts[2]).write_bytes(b"unexpected replacement")
           if case_name == "solver companion changed during gate":
-            (case_root / case_artifacts[3]).write_bytes(b"unexpected solver replacement")
+            solver_path = next(path for path in case_artifacts if path.endswith(solver_name))
+            (case_root / solver_path).write_bytes(b"unexpected solver replacement")
           return True
 
         with patch.object(sys, "argv", ["qualification", "--gate", "coverage", "--output", str(output)]), \
@@ -302,7 +317,7 @@ class TestSafetyQualificationRunner(TestCase):
       fake = Path(td) / "foreign"
       (fake / "panda").mkdir(parents=True)
       (fake / "panda/__init__.py").write_text("# second checkout shadow\n")
-      gate = qualification.Gate("mutation-list", Path(td), qualification.ROOT / ".venv/bin/python")
+      gate = qualification.Gate("mutation-list", Path(td), sys.executable)
       self.assertTrue(qualification.check_import_origins(gate.python, gate.env, Path(td) / "imports-baseline.log"))
       gate.env["PYTHONPATH"] = str(fake) + os.pathsep + gate.env["PYTHONPATH"]
       self.assertFalse(qualification.check_import_origins(gate.python, gate.env, Path(td) / "imports.log"))