UI: curve-speed glow, gas-override blink and Theme maker preview

- Steady the curve-speed glow and the gas-override blink on the on-road screen
- Stop the Theme maker preview flapping to "unavailable" and allow for Galaxy tunnel latency
- Carry the widget sprite sheet in a header so it survives the tunnel

openpilot/starpilot/curve_speed/status.py

@@ -1,12 +1,15 @@
 """Independent curve diagnostics in the existing single-owner planner event."""
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
+import math
 import secrets
 
 from openpilot.cereal import messaging
 from openpilot.starpilot.curve_speed.learning import finite_number
 
 STATUS_LIFETIME_NS = 100_000_000
+CURVATURE_DISPLAY_TAU_S = 0.6
+CURVATURE_DISPLAY_HOLD_NS = 1_000_000_000
 
 
 @dataclass(frozen=True)
@@ -90,3 +93,30 @@ def observation(state, now_ns: int) -> CurveObservation | None:
                             str(value.persistenceStatus), bool(value.glow), bool(value.curveOnly), curvature)
   except (AttributeError, TypeError, ValueError, OverflowError):
     return None
+
+
+class CurvatureDisplay:
+  """Smooth published curvature for the glow, label, and gauge colors.
+
+  roadCurvature is a raw per-model-frame max over the horizon and drops out
+  whenever there is no candidate; the green-to-red ramp turned that jitter
+  into a strobing border. Short dropouts hold the last value.
+  """
+
+  def __init__(self):
+    self._value: float | None = None
+    self._stamp_ns = 0
+
+  def update(self, curve: CurveObservation | None, now_ns: int) -> CurveObservation | None:
+    if curve is None:
+      return None
+    raw, gap = curve.road_curvature, now_ns - self._stamp_ns
+    if self._value is None or not 0 <= gap <= CURVATURE_DISPLAY_HOLD_NS:
+      self._value = raw
+      self._stamp_ns = now_ns
+      return curve
+    if raw is None:
+      return replace(curve, road_curvature=self._value)
+    self._value += (raw - self._value) * (1.0 - math.exp(-gap / 1e9 / CURVATURE_DISPLAY_TAU_S))
+    self._stamp_ns = now_ns
+    return replace(curve, road_curvature=self._value)

openpilot/starpilot/galaxy/batch_routes.py

@@ -53,6 +53,21 @@ def delete_records(self, path, payload, sentry_events_lock, crash_reports_lock,
       self.json(401, {'error': 'Sign in to Galaxy'})
 
 
+def widget_art_header(image):
+  """The renderer's widget-art tEXt chunk, base64-encoded for an HTTP header ('' when absent or oversized)."""
+  import base64
+  import struct
+  signature = b'starpilot-widget-art\0'
+  offset = 8
+  while offset + 12 <= len(image):
+    size, kind = struct.unpack('!I4s', image[offset:offset + 8])
+    if kind == b'tEXt' and image[offset + 8:offset + 8 + len(signature)] == signature:
+      text = image[offset + 8 + len(signature):offset + 8 + size]
+      return base64.b64encode(text).decode() if len(text) <= 8192 else ''
+    offset += size + 12
+  return ''
+
+
 def preview_layout(self, payload, projection_layout_owner, layout_owner, layout_preview_socket):
   identity = self.settings_session()
   if identity is None:
@@ -91,7 +106,14 @@ def preview_layout(self, payload, projection_layout_owner, layout_owner, layout_
     elif not layout_owner().parked():
       self.json(403, {'error': 'Turn off the vehicle to preview this layout'})
     else:
-      self.respond(200, image, 'image/png')
+      headers, cache = {}, None
+      if payload.get('art') is True:
+        # Proxies that re-encode PNGs drop the tEXt chunk, so the sprite sheet also rides in a header.
+        header = widget_art_header(image)
+        if header:
+          headers['X-Widget-Art'] = header
+        cache = 'no-store, no-transform'
+      self.respond(200, image, 'image/png', asset_headers=headers or None, cache_control=cache)
   except PreviewDenied:
     self.json(403, {'error': 'Turn off the vehicle to preview this layout'})
   except PreviewBusy:

openpilot/starpilot/galaxy/tests/test_widget_art.mjs

@@ -10,13 +10,19 @@ function chunk(kind, bytes) {
   return Buffer.concat([size,body,sum])
 }
 const meta={version:1,width:2,height:1,sprites:{conditional_mode:{x:0,y:0,width:2,height:1}}}
-function image(metadata=meta){
+function image(metadata=meta,text=true){
   const header=Buffer.alloc(13);header.writeUInt32BE(2);header.writeUInt32BE(1,4);header[8]=8;header[9]=6
   return new Blob([Buffer.concat([Buffer.from([137,80,78,71,13,10,26,10]),chunk('IHDR',header),
-    chunk('tEXt',Buffer.from('starpilot-widget-art\0'+JSON.stringify(metadata))),
+    ...(text?[chunk('tEXt',Buffer.from('starpilot-widget-art\0'+JSON.stringify(metadata)))]:[]),
     chunk('IDAT',deflateSync(Buffer.from([0,255,255,255,255,0,0,0,0]))),chunk('IEND',Buffer.alloc(0))])],{type:'image/png'})
 }
 assert.deepEqual(await widgetArtMetadata(image()),meta)
+// A proxy that re-encodes the PNG drops the tEXt chunk; the header still carries the sheet.
+const stripped=image(meta,false),header=Buffer.from(JSON.stringify(meta)).toString('base64')
+assert.deepEqual(await widgetArtMetadata(stripped,header),meta,'Sprite sheet survives a proxy that strips PNG text')
+await assert.rejects(widgetArtMetadata(stripped,null),/metadata/)
+assert.deepEqual(await widgetArtMetadata(image(),'image/png'),meta,'A malformed header falls back to the PNG chunk')
+await assert.rejects(widgetArtMetadata(stripped,Buffer.from(JSON.stringify({...meta,width:3})).toString('base64')))
 const withHeader={...meta,sprites:{speed_limit_actions:{x:0,y:0,width:2,height:1,offsetY:-32.5}}}
 assert.deepEqual(await widgetArtMetadata(image(withHeader)),withHeader,'Native action sprites include the heading above the buttons')
 for(const bad of [{...meta,width:3},{...meta,sprites:{conditional_mode:{x:1,y:0,width:2,height:1}}},
@@ -125,6 +131,7 @@ stalled.feed.stop()
   run()
   await flush();assert.equal(requests.length,1,'The second feed waits instead of being turned away as busy')
   assert.equal(JSON.parse(requests[0].options.body).art,'backdrop')
+  assert.equal([...timers.values()].filter(t=>t.delay>=7000).length,1,'Only the feed holding the renderer runs a response clock; waiting its turn costs no budget')
   backdrop.update(document,'compact','aol');run()
   await flush();assert.equal(requests.length,1,'Cancelling a request does not release the renderer before it answers')
   requests[0].resolve({status:200,ok:true,headers:{get:()=>'image/png'},blob:async()=>image()});await flush()

openpilot/starpilot/galaxy/tests/test_widget_art_header.py

@@ -0,0 +1,29 @@
+import base64
+import json
+import struct
+import unittest
+import zlib
+
+from openpilot.starpilot.galaxy.batch_routes import widget_art_header
+
+
+def chunk(kind, body):
+  return struct.pack('!I', len(body)) + kind + body + struct.pack('!I', zlib.crc32(kind + body))
+
+
+class WidgetArtHeaderTest(unittest.TestCase):
+  def png(self, text):
+    return b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('!IIBBBBB', 1, 1, 8, 6, 0, 0, 0)) + text + chunk(b'IEND', b'')
+
+  def test_sheet_is_copied_into_header(self):
+    meta = {'version': 1, 'width': 1, 'height': 1, 'sprites': {}}
+    image = self.png(chunk(b'tEXt', b'starpilot-widget-art\0' + json.dumps(meta).encode()))
+    self.assertEqual(json.loads(base64.b64decode(widget_art_header(image))), meta)
+
+  def test_missing_or_oversized_sheet_gives_no_header(self):
+    self.assertEqual(widget_art_header(self.png(b'')), '')
+    self.assertEqual(widget_art_header(self.png(chunk(b'tEXt', b'starpilot-widget-art\0' + b'x' * 9000))), '')
+
+
+if __name__ == '__main__':
+  unittest.main()

openpilot/starpilot/galaxy/web/js/layout-preview.js

@@ -59,6 +59,9 @@ export function sceneHidesWidget(scene, profile, id, layout = {}, widgets = {})
 
 // The device renders one preview at a time and answers a second request with 429.
 // Feeds sharing a queue take turns instead, widget art before the backdrop.
+// Room for a rendered PNG to cross the Galaxy tunnel, not just the local network.
+const PREVIEW_TIMEOUT_MS = 20000
+
 export class PreviewQueue {
   constructor() { this.running = false; this.waiting = [] }
 
@@ -181,7 +184,9 @@ export class LayoutPreviewFeed {
     this.last = item
     const request = new AbortController()
     this.request = request
-    const timeout = this.later(() => request.abort(), 7000)
+    // The clock starts once this feed holds the renderer: waiting behind the other feed (slow over a tunnel)
+    // used to eat the whole budget and flip the backdrop to "unavailable" before it was even sent.
+    let timeout = null
     let rejectAborted
     const aborted = new Promise((_, reject) => { rejectAborted = () => reject(new Error("Preview request cancelled")) })
     request.signal.addEventListener("abort", rejectAborted, { once: true })
@@ -190,6 +195,7 @@ export class LayoutPreviewFeed {
     const work = (async () => {
       const held = !!this.queue && await this.queue.acquire(this.art ? 0 : 1, request.signal).then(() => true)
       try {
+        timeout = this.later(() => request.abort(), PREVIEW_TIMEOUT_MS)
         const response = await this.fetcher("./api/ui/layout/preview", { method: "POST", credentials: "same-origin",
           cache: "no-store", headers: { "Content-Type": "application/json" },
           body: JSON.stringify({ document: item.document, profile: item.profile, scene: item.scene, ...(this.art ? {art: true} : this.backdrop ? {art: "backdrop"} : {}) }), signal: request.signal })
@@ -202,7 +208,7 @@ export class LayoutPreviewFeed {
         const blob = await response.blob()
         if (!this.active || item.version !== this.version || request.signal.aborted) return
         if (!blob.size || blob.size > 4 * 1024 * 1024) throw new Error("Preview is unavailable. Reconnecting automatically…")
-        const sprites = this.art ? await widgetArtMetadata(blob) : null
+        const sprites = this.art ? await widgetArtMetadata(blob, response.headers.get("X-Widget-Art")) : null
         if (!this.active || item.version !== this.version || request.signal.aborted) return
         const cacheKey = item.profile + "\n" + item.scene
         const previousUrl = this.art ? this.artCache.get(cacheKey)?.url : this.currentUrl
@@ -227,13 +233,13 @@ export class LayoutPreviewFeed {
         this.inputKey = null
         this.failures = denied ? 2 : this.failures + 1
         if (this.art && denied) this.clearImage()
-        // Art retries promptly, as does any feed turned away by a busy renderer; real outages back off.
-        const recovering = (this.art || busy) && this.failures < 2
+        // Art and backdrop retry promptly, as does any feed turned away by a busy renderer; real outages back off.
+        const recovering = (this.art || this.backdrop || busy) && this.failures < 2
         this.publish({ status: recovering ? "updating" : "unavailable", error: recovering ? "" : failure })
         this.timer = this.later(() => { this.timer = null; this.retry() }, recovering ? 600 : 5000)
       }
     } finally {
-      this.cancelTimer(timeout)
+      if (timeout !== null) this.cancelTimer(timeout)
       request.signal.removeEventListener("abort", rejectAborted)
       if (this.request === request) this.request = null
       if (this.active && this.pending && this.timer === null) this.dispatch()

openpilot/starpilot/galaxy/web/js/widget-art.js

@@ -1,17 +1,28 @@
 // Pixels and sprite locations come from the actual onroad renderer, not browser widget copies.
-export async function widgetArtMetadata(blob) {
+// The sheet also travels in an X-Widget-Art header: image-optimising proxies (such as the one in front of the
+// Galaxy tunnel) re-encode PNGs and drop the tEXt chunk, which used to reject the art over the tunnel.
+function headerMetadata(header) {
+  try {
+    const bytes = Uint8Array.from(atob(header), c => c.charCodeAt(0))
+    return JSON.parse(new TextDecoder().decode(bytes))
+  } catch { return null }
+}
+
+export async function widgetArtMetadata(blob, header = null) {
   const bytes = new Uint8Array(await blob.arrayBuffer())
   const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
   if (bytes.length < 45 || bytes.length > 4 * 1024 * 1024 ||
       ![137,80,78,71,13,10,26,10].every((value, index) => bytes[index] === value)) throw new Error("Invalid widget art")
   const text = new TextDecoder(), signature = "starpilot-widget-art\0"
-  let metadata = null
+  let metadata = typeof header === "string" && header.length <= 12288 ? headerMetadata(header) : null
+  const fromHeader = metadata !== null
   for (let offset = 8; offset + 12 <= bytes.length;) {
     const size = view.getUint32(offset), type = text.decode(bytes.subarray(offset + 4, offset + 8))
     if (size > bytes.length - offset - 12) throw new Error("Invalid widget art chunk")
     if (type === "tEXt") {
       const value = text.decode(bytes.subarray(offset + 8, offset + 8 + size))
       if (value.startsWith(signature)) {
+        if (fromHeader) { offset += size + 12; continue }
         if (metadata || size > 8192) throw new Error("Invalid widget art metadata")
         metadata = JSON.parse(value.slice(signature.length))
       }

openpilot/starpilot/ui/onroad.py

@@ -25,7 +25,7 @@ from openpilot.starpilot.ui.onroad_alerts import AlertRenderer
 from openpilot.starpilot.ui.onroad_border import render_compact_half_borders, render_large_half_borders
 from openpilot.starpilot.ui.onroad_compact_widgets import CompactHudRenderer, MiciSidebarWidgets
 from openpilot.starpilot.ui.onroad_personality import PersonalityWidget
-from openpilot.starpilot.ui.onroad_curve import controlling as curve_controlling, glow_color, intensity, render_glow, status_label
+from openpilot.starpilot.ui.onroad_curve import GlowPhase, controlling as curve_controlling, glow_color, intensity, render_glow, status_label
 from openpilot.starpilot.ui.onroad_conditional import stop_active
 from openpilot.starpilot.ui.onroad_corner import CornerHintCache, render_corner_hint
 from openpilot.starpilot.ui.onroad_large_widgets import CurrentSpeedHud, UnifiedSpeedWidget, SteeringWheelWidget
@@ -126,6 +126,7 @@ class OnroadView:
     self.wall_clock = wall_clock or (lambda: datetime.now().astimezone())
     self._corner_cache = CornerHintCache()
     self.quick_select_layer = None
+    self._curve_glow_phase = GlowPhase()
     gui_app.add_render_prepare(self._corner_cache.prepare, self._corner_cache.close)
     self.fonts = fonts
     self.camera_layer = camera_layer
@@ -457,7 +458,8 @@ class OnroadView:
       self.fonts.draw(label, FontRole.SEMI_BOLD, 29, 1390 + right_shift, 56,
                       curve_color)
     self._traffic_badge(state, 56 + 45 * int(label is not None))
-    render_glow(content, state, border_width=30, time_s=state.observed_ns / 1e9 if state.observed_ns else rl.get_time())
+    render_glow(content, state, border_width=30, time_s=state.observed_ns / 1e9 if state.observed_ns else rl.get_time(),
+                phase=getattr(self, "_curve_glow_phase", None))
     rl.draw_rectangle_lines_ex(frame, 30, rl.BLACK)
     rl.draw_rectangle_rounded_lines_ex(content, 0.12, 10, 30, edge)
     render_large_half_borders(state, state.observed_ns / 1e9 if state.observed_ns else rl.get_time(), content, int(height))

openpilot/starpilot/ui/onroad_curve.py

@@ -91,7 +91,22 @@ def status_label(state: OnroadState) -> str | None:
   return None
 
 
-def render_glow(rect: rl.Rectangle, state: OnroadState, *, border_width: float, time_s: float) -> None:
+class GlowPhase:
+  """Advance the breath by dt/period; (t % period) / period jumps whenever the period changes."""
+
+  def __init__(self):
+    self._phase = 0.0
+    self._time_s: float | None = None
+
+  def advance(self, time_s: float, cycle: float) -> float:
+    if self._time_s is not None and 0.0 <= time_s - self._time_s <= 1.0:
+      self._phase = (self._phase + (time_s - self._time_s) / cycle) % 1.0
+    self._time_s = time_s
+    return self._phase
+
+
+def render_glow(rect: rl.Rectangle, state: OnroadState, *, border_width: float, time_s: float,
+                phase: GlowPhase | None = None) -> None:
   """Draw behind the axis-status border; absent/stale/off draws nothing."""
   preview = state.visual_preview
   preview_curve = preview.curve_curvature if preview is not None and state.alert.size != AlertSize.FULL else None
@@ -109,7 +124,8 @@ def render_glow(rect: rl.Rectangle, state: OnroadState, *, border_width: float,
   cycle = period(strength)
   fraction = max(0.0, min(1.0, (strength - BASE_INTENSITY) / (1.0 - BASE_INTENSITY)))
   amplitude = 0.3 + 0.7 * fraction
-  breath = max(0.0, min(1.0, 0.5 + amplitude * math.sin((time_s % cycle) / cycle * 2 * math.pi)))
+  position = phase.advance(time_s, cycle) if phase is not None else (time_s % cycle) / cycle
+  breath = max(0.0, min(1.0, 0.5 + amplitude * math.sin(position * 2 * math.pi)))
   alpha = int(200 * strength * (0.6 + 0.4 * breath))
   span = max(1, int(border_width * SPAN))
   edge = rl.Color(color.r, color.g, color.b, alpha)

openpilot/starpilot/ui/runtime_snapshot.py

@@ -17,11 +17,10 @@ from typing import Any
 from openpilot.starpilot.galaxy.access import AccessStatus, GalaxyAccessOwner
 from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice, OnroadAppearance, onroad_appearance
 from openpilot.starpilot.ui.brand import DISPLAY_VERSION, home_description
-from openpilot.starpilot.curve_speed.status import observation as curve_observation
+from openpilot.starpilot.curve_speed.status import CurvatureDisplay, observation as curve_observation
 from openpilot.starpilot.conditional_mode.policy import ModeChoice
 from openpilot.starpilot.conditional_mode.manual import TRAFFIC_MODE_ACTION, read_button_map
 from openpilot.starpilot.feature_capabilities import lateral_supported, longitudinal_supported
-from openpilot.starpilot.aol.runtime import current_axis
 from openpilot.starpilot.aol.wire import INTENT_SERVICE, decode_intent
 from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns
 from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
@@ -57,6 +56,7 @@ from openpilot.starpilot.ui.toggles_state import Personality, ToggleKey, Toggles
 
 
 SLC_SIGN_HOLD_NS = 150_000_000
+EVENTS_DISPLAY_MAX_AGE_NS = 1_500_000_000  # selfdrived sends onroadEvents on every change and at least once a second
 
 
 def _boot_time_ns() -> int:
@@ -324,6 +324,7 @@ class RuntimeSnapshotAdapter:
     self._conditional_read_ns: int | None = None
     self._traffic = TrafficDisplayProjector()
     self._switchback = SwitchbackStatusOwner()
+    self._curve_curvature = CurvatureDisplay()
     self._traffic_settings = ConditionalSettingsOwner(ui_state.params)
     self._traffic_context_ns: int | None = None
     self._traffic_map_fingerprint: str | None = None
@@ -451,7 +452,7 @@ class RuntimeSnapshotAdapter:
     display_model = display_message(sm, "modelV2", now_ns, after_frame=after) if started else None
     display_radar = display_message(sm, "radarState", now_ns, after_frame=after) if started else None
     display_plan = display_message(sm, 'longitudinalPlan', now_ns, after_frame=after) if started else None
-    display_events = display_message(sm, 'onroadEvents', now_ns, after_frame=after) if started else None
+    display_events = _message_at_age(sm, 'onroadEvents', now_ns, after, EVENTS_DISPLAY_MAX_AGE_NS) if started else None
     display_monitor = display_message(sm, 'driverMonitoringState', now_ns, after_frame=after) if started else None
     if display_monitor is not None:
       self._display_rhd = bool(display_monitor.isRHD)
@@ -463,7 +464,7 @@ class RuntimeSnapshotAdapter:
       now_ns=now_ns, native_alert=native_alert, hide=self._appearance_value.hide_personality_alerts)
     slc = display_message(sm, "slcState", now_ns, after_frame=after) if started else None
     curve_envelope = current_curve_message(sm, now_ns, after_frame=after) if started else None
-    curve = curve_observation(curve_envelope, now_ns) if curve_envelope is not None else None
+    curve = self._curve_curvature.update(curve_observation(curve_envelope, now_ns) if curve_envelope is not None else None, now_ns)
     conditional_envelope = display_message(sm, 'starpilotSelfdriveState', now_ns, after_frame=after) if started else None
 
     from openpilot.starpilot.navigation.wire import navigation_state
@@ -660,8 +661,10 @@ class RuntimeSnapshotAdapter:
     display_long_active = bool(display_control.longActive) if display_control is not None else False
     long_engaged = bool(display_selfdrive is not None and getattr(display_selfdrive, 'enabled', False))
     if getattr(ui, 'aol_runtime', False):
-      axis = current_axis(sm, now_ns=now_ns) if started and sm.recv_frame.get('aolAxisState', 0) > after else None
-      long_engaged = long_engaged and bool(axis is not None and axis.nativeAcknowledged and axis.desiredLongitudinal)
+      # Display only: the control-side 30 ms proof window would blink the override edge whenever a frame runs late.
+      axis = display_message(sm, 'aolAxisState', now_ns, after_frame=after) if started else None
+      long_engaged = long_engaged and bool(axis is not None and axis.qualified and axis.sessionId and
+                                           axis.nativeAcknowledged and axis.desiredLongitudinal)
     longitudinal_overridden = bool(
       display_car_valid and getattr(display_car, 'gasPressed', False) and long_engaged and
       str(getattr(display_selfdrive, 'state', '')) == 'overriding' and

openpilot/starpilot/ui/tests/test_curve_glow_flicker.py

@@ -0,0 +1,80 @@
+"""Noisy model curvature must not strobe the curve glow between green and red."""
+
+import random
+import unittest
+from unittest.mock import patch
+
+import pyray as rl
+
+from openpilot.starpilot.curve_speed.status import CurvatureDisplay, CurveObservation
+from openpilot.starpilot.ui.developer_preview import OnroadVisualPreview
+from openpilot.starpilot.ui.onroad_curve import GlowPhase, render_glow
+from openpilot.starpilot.ui.onroad_state import OnroadAlert, OnroadState, SpeedLimitObservation
+
+FRAME_NS = 50_000_000
+START_NS = 5_000 * 1_000_000_000  # observed time is monotonic uptime, not seconds since the glow appeared
+
+
+def curve(curvature: float | None) -> CurveObservation:
+  return CurveObservation(True, 20.0, 20.0, True, True, False, 100.0, 2.0, 50.0, 'curve', 'selected', 'ok', True, False,
+                          curvature)
+
+
+def jitter(frames: int) -> list[float | None]:
+  rng = random.Random(7)
+  return [None if rng.random() < .1 else rng.uniform(.004, .016) for _ in range(frames)]
+
+
+def edges(curvatures, phase):
+  rect = rl.Rectangle(30, 30, 1800, 1020)
+  colors = []
+  with patch.object(rl, 'draw_rectangle_gradient_v') as vertical, patch.object(rl, 'draw_rectangle_gradient_h'):
+    for frame, curvature in enumerate(curvatures):
+      state = OnroadState(engaged=False, camera_available=False, speed_mps=0.0, cruise_kph=None,
+                          speed_limit=SpeedLimitObservation(), alert=OnroadAlert(),
+                          visual_preview=OnroadVisualPreview(None, curvature))
+      vertical.reset_mock()
+      render_glow(rect, state, border_width=30, time_s=(START_NS + frame * FRAME_NS) / 1e9, phase=phase)
+      colors.append(vertical.call_args_list[0].args[4])
+  return colors
+
+
+def worst_step(colors) -> int:
+  return max(max(abs(a.r - b.r), abs(a.g - b.g), abs(a.b - b.b), abs(a.a - b.a)) for a, b in zip(colors, colors[1:], strict=False))
+
+
+class CurveGlowFlickerTests(unittest.TestCase):
+  def test_smoothed_curvature_holds_dropouts_and_resets_after_gap(self):
+    display = CurvatureDisplay()
+    self.assertEqual(display.update(curve(.01), START_NS).road_curvature, .01)
+    held = display.update(curve(None), START_NS + FRAME_NS)
+    self.assertEqual(held.road_curvature, .01)
+    stepped = display.update(curve(.02), START_NS + 2 * FRAME_NS).road_curvature
+    self.assertTrue(.01 < stepped < .013, stepped)
+    self.assertIsNone(display.update(None, START_NS + 3 * FRAME_NS))
+    self.assertEqual(display.update(curve(.003), START_NS + 3 * FRAME_NS + 2_000_000_000).road_curvature, .003)
+
+  def test_glow_changes_gradually_under_jittery_curvature(self):
+    display = CurvatureDisplay()
+    smoothed = [display.update(curve(value), START_NS + frame * FRAME_NS).road_curvature
+                for frame, value in enumerate(jitter(200))]
+    self.assertLessEqual(worst_step(edges(smoothed, GlowPhase())), 40)
+
+  def test_raw_jitter_without_phase_state_strobes(self):
+    raw = [value if value is not None else .0001 for value in jitter(200)]
+    self.assertGreater(worst_step(edges(raw, None)), 150)
+
+  def test_tiny_period_changes_do_not_jump_the_breath(self):
+    # 1% curvature noise barely moves the color, but re-deriving the phase from uptime strobed the alpha.
+    curvatures = [.01 if frame % 2 else .0101 for frame in range(200)]
+    self.assertGreater(worst_step(edges(curvatures, None)), 30)
+    self.assertLessEqual(worst_step(edges(curvatures, GlowPhase())), 8)
+
+  def test_constant_period_matches_stateless_breath(self):
+    curvatures = [.01] * 40
+    stateless, phased = edges(curvatures, None), edges(curvatures, GlowPhase())
+    self.assertLessEqual(worst_step(phased), worst_step(stateless) + 1)
+
+
+if __name__ == '__main__':
+  unittest.main()

openpilot/starpilot/ui/tests/test_onroad_axes.py

@@ -219,22 +219,24 @@ class TestOnroadAxes(unittest.TestCase):
               nativeAcknowledged=True, desiredLongitudinal=True)
     self.ui.sm.put('aolAxisState', axis)
     self.assertTrue(self.state(True, False).longitudinal_overridden)
-    for field, value in (('qualified', False), ('sessionId', ''), ('observedMonoTime', NOW + 1),
-                         ('validUntilMonoTime', NOW - 1), ('nativeAcknowledged', False)):
+    for field, value in (('qualified', False), ('sessionId', ''), ('nativeAcknowledged', False), ('desiredLongitudinal', False)):
       with self.subTest(field=field):
         original = getattr(axis, field)
         setattr(axis, field, value)
         state = self.state(True, False)
         self.assertFalse(state.longitudinal_overridden or state.driver_override_active or state.cruise_active)
         setattr(axis, field, original)
-    for stamp in (NOW + 1, NOW - 31_000_000):
+    self.ui.sm.logMonoTime['aolAxisState'] = NOW - 60_000_000
+    self.assertTrue(self.state(True, False).longitudinal_overridden,
+                    'a late UI frame past the 30 ms control window does not blink the override edge')
+    for stamp in (NOW + 1, NOW - 201_000_000):
       self.ui.sm.logMonoTime['aolAxisState'] = stamp
       self.assertFalse(self.state(True, False).longitudinal_overridden)
     self.ui.sm.logMonoTime['aolAxisState'] = NOW
     self.ui.sm.recv_frame['aolAxisState'] = self.ui.started_frame
     self.assertFalse(self.state(True, False).longitudinal_overridden)
     self.ui.sm.recv_frame['aolAxisState'] = self.ui.started_frame + 1
-    for flags in (self.ui.sm.valid, self.ui.sm.alive, self.ui.sm.seen):
+    for flags in (self.ui.sm.valid, self.ui.sm.alive):
       flags['aolAxisState'] = False
       self.assertFalse(self.state(True, False).longitudinal_overridden)
       flags['aolAxisState'] = True

openpilot/starpilot/ui/tests/test_runtime_snapshot.py

@@ -175,6 +175,8 @@ class TestRuntimeSnapshot(unittest.TestCase):
     adapter = RuntimeSnapshotAdapter(ui)
     state = adapter.build(ShellMode.ONROAD, now_ns=NOW).onroad
     self.assertTrue(state.lead_detected and state.gas_pressed_override and state.right_hand_drive)
+    state = adapter.build(ShellMode.ONROAD, now_ns=NOW + 1_000_000_000).onroad
+    self.assertTrue(state.gas_pressed_override, 'events last until the next once-a-second send, so the blink stays steady')
     ui.sm.recv_frame['longitudinalPlan'] = ui.sm.recv_frame['onroadEvents'] = 0
     state = adapter.build(ShellMode.ONROAD, now_ns=NOW + 1).onroad
     self.assertFalse(state.lead_detected or state.gas_pressed_override)