Keep Traffic border stable during snapshot settings reads

openpilot/starpilot/ui/runtime_snapshot.py

@@ -399,8 +399,11 @@ class RuntimeSnapshotAdapter:
                       mono_clock=self._mono_clock, boot_clock=self._boot_clock)
         self._replay_epoch = replay.epoch
       now_ns = replay.now_ns if replay is not None and replay.valid else 0
+      traffic_clock = (replay.now_ns, replay.boot_ns) if replay is not None and replay.valid and replay.boot_ns is not None else None
     else:
       now_ns = time.monotonic_ns() if now_ns is None else now_ns
+      # Pair the Traffic clocks before settings reads and snapshot work can delay them.
+      traffic_clock = paired_clocks_ns()
     if mode != ShellMode.ONROAD or self._last_mode != ShellMode.ONROAD:
       self._onroad_ancillary = None
       self._onroad_view_ancillary = None
@@ -546,14 +549,10 @@ class RuntimeSnapshotAdapter:
         except (AttributeError, OSError, TypeError, ValueError):
           self._traffic_profile_valid = False
         self._traffic_context_ns = now_ns
-      if getattr(ui, "replay_clock", None) is not None:
-        pair = (replay.now_ns, replay.boot_ns) if replay is not None and replay.valid and replay.boot_ns is not None else None
-      else:
-        pair = paired_clocks_ns()
-      if pair is not None and abs(pair[0] - now_ns) <= 10_000_000:
+      if traffic_clock is not None and abs(traffic_clock[0] - now_ns) <= 10_000_000:
         fingerprint = settings_fingerprint(self._traffic_settings.refresh(now_ns))
         traffic = self._traffic.project(
-          traffic_envelope, now_mono_ns=now_ns, now_boot_ns=pair[1],
+          traffic_envelope, now_mono_ns=now_ns, now_boot_ns=traffic_clock[1],
           drive_id=drive_id,
           settings_fingerprint=fingerprint, map_fingerprint=self._traffic_map_fingerprint,
           map_assigned=self._traffic_map_assigned,

openpilot/starpilot/ui/tests/test_onroad_axes.py

@@ -281,8 +281,17 @@ class TestOnroadAxes(unittest.TestCase):
       (OnroadAlert(size=AlertSize.SMALL, text1='Steering Paused'), (10, 186, 181, 255)),
       (OnroadAlert(size=AlertSize.MID), (10, 186, 181, 255)),
       (OnroadAlert(size=AlertSize.FULL), blue), (OnroadAlert(size=AlertSize.SMALL, critical=True), blue))]
-    for lateral, longitudinal, override, alert, compact_color, latched in cases:
-      state = replace(self.state(lateral, longitudinal), driver_override_active=override, alert=alert, aol_latched=latched)
+    cases = [(*case, False) for case in cases]
+    cases += [(True, True, override, alert, color, None, True) for override, alert, color in (
+      (False, OnroadAlert(), (201, 34, 49, 255)),
+      (False, OnroadAlert(size=AlertSize.SMALL, critical=True), blue),
+      (False, OnroadAlert(), (201, 34, 49, 255)),
+      (True, OnroadAlert(), gray),
+      (False, OnroadAlert(size=AlertSize.FULL), blue),
+      (False, OnroadAlert(), (201, 34, 49, 255)))]
+    for lateral, longitudinal, override, alert, compact_color, latched, traffic in cases:
+      state = replace(self.state(lateral, longitudinal), driver_override_active=override, alert=alert,
+                      aol_latched=latched, traffic_mode=traffic)
       for layout in state.customization['layouts'].values():
         layout[MODE_WIDGET]['enabled'] = True
       for profile in (Profile.LARGE, Profile.COMPACT):

openpilot/starpilot/ui/tests/test_traffic_status.py

@@ -16,7 +16,7 @@ from openpilot.starpilot.conditional_mode.manual import read_button_map
 from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
 from openpilot.starpilot.conditional_mode.status import settings_fingerprint
 from openpilot.starpilot.ui.onroad_compact_widgets import MiciSidebarWidgets
-from openpilot.starpilot.ui.onroad import OnroadView
+from openpilot.starpilot.ui.onroad import OnroadView, axis_status_color
 from openpilot.starpilot.ui.presentation import FontRole, Profile
 from openpilot.starpilot.ui.runtime_snapshot import RuntimeSnapshotAdapter
 from openpilot.starpilot.ui.shell import ShellMode
@@ -110,6 +110,55 @@ class TestTrafficStatus(unittest.TestCase):
     self.assertIsNone(self.project(first))
     self.assertIsNone(self.project(next_frame, drive_id=DRIVE + 1))
 
+  def test_snapshot_work_does_not_blink_fresh_traffic_border(self):
+    ui = ui_fake()
+    ui.params = self.params
+    ui.CP = interfaces[CAR.HYUNDAI_IONIQ_6].get_non_essential_params(CAR.HYUNDAI_IONIQ_6)
+    ui.CP.flags = int(HyundaiFlags.CANFD_LKA_STEER_MSG)
+    ui.CP.openpilotLongitudinalControl = True
+    ui.sm['deviceState'].startedMonoTime = DRIVE
+    ui.sm['carState'].canValid = True
+    ui.sm['carState'].canTimeout = False
+    ui.sm['carControl'].longActive = True
+    ui.sm['selfdriveState'].enabled = True
+    ui.sm.put('controlsState', NS(longControlState='pid'))
+    adapter = RuntimeSnapshotAdapter(ui, mono_clock=lambda: NOW, boot_clock=lambda: NOW + BOOT_OFFSET_NS)
+    clock = [NOW]
+    # A delayed settings read must not move clock acquisition away from the snapshot.
+    from openpilot.starpilot.ui import runtime_snapshot
+    read_settings = runtime_snapshot.read_traffic_settings
+
+    def delayed_settings(params):
+      clock[0] += 30_000_000
+      return read_settings(params)
+
+    with mock.patch.object(runtime_snapshot, 'paired_clocks_ns', side_effect=lambda: (clock[0], clock[0] + BOOT - NOW, 0)), \
+         mock.patch.object(runtime_snapshot, 'read_traffic_settings', side_effect=delayed_settings):
+      for frame, delay_read in enumerate((True, False, True, False, True)):
+        now = NOW + frame * 10_000_000
+        clock[0] = now
+        if delay_read:
+          adapter._traffic_context_ns = None
+        event = receipt(self.settings_hash, self.map_hash, observed=now, sequence=frame + 1)
+        ui.sm.put('slcState', event.slcState)
+        for _ in range(2):
+          clock[0] = now
+          shown = adapter.build(ShellMode.ONROAD, now_ns=now).onroad
+          self.assertTrue(shown.traffic_mode)
+          color = axis_status_color(shown)
+          self.assertEqual((color.r, color.g, color.b, color.a), (201, 34, 49, 255))
+      # Actual expiry still clears status; no cached red border or grace period.
+      clock[0] = now + 100_000_001
+      shown = adapter.build(ShellMode.ONROAD, now_ns=clock[0]).onroad
+      self.assertFalse(shown.traffic_mode)
+      self.assertIsNone(shown.traffic_display)
+      # A mismatched or unavailable clock pair must still fail closed.
+      clock[0] = now
+      self.assertTrue(adapter.build(ShellMode.ONROAD, now_ns=now).onroad.traffic_mode)
+      for pair in (None, (now + 10_000_001, BOOT, 0)):
+        with mock.patch.object(runtime_snapshot, 'paired_clocks_ns', return_value=pair):
+          self.assertFalse(adapter.build(ShellMode.ONROAD, now_ns=now).onroad.traffic_mode)
+
   def test_snapshot_and_compact_rail_use_only_live_effective_status(self):
     ui = ui_fake()
     ui.params = self.params