diff --git a/.gitattributes b/.gitattributes
index a7a632ca1b..1c53d34964 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -41,3 +41,5 @@ openpilot/system/loggerd/loggerd -text
 openpilot/starpilot/galaxy/bin/frpc_linux_arm64 -text
 openpilot/starpilot/maps/provider/mapd -text
 openpilot/selfdrive/test/process_replay/refs/starpilot/*.zst -text
+
+openpilot/tools/replay/replay -text
diff --git a/openpilot/selfdrive/selfdrived/helpers.py b/openpilot/selfdrive/selfdrived/helpers.py
index e7dca0f9d5..567d53aef1 100644
--- a/openpilot/selfdrive/selfdrived/helpers.py
+++ b/openpilot/selfdrive/selfdrived/helpers.py
@@ -1,7 +1,8 @@
 import math
 from enum import StrEnum, auto
 
-from openpilot.cereal import messaging
+from openpilot.cereal import log, messaging
+from openpilot.starpilot.selfdrive_loop_timing import timed_params_read
 from opendbc.car.structs import car
 from openpilot.common.realtime import DT_CTRL
 from openpilot.selfdrive.locationd.helpers import Pose
@@ -53,3 +54,28 @@ class ExcessiveActuationCheck:
         excessive_type = ExcessiveActuationType.LATERAL
 
     return excessive_type
+
+
+def update_big_model_status(owner) -> bool:
+  big_active = timed_params_read(owner, "ChestnutActive")
+  fresh = owner.sm.seen['modelV2'] and owner.sm.alive['modelV2'] and owner.sm.valid['modelV2']
+  published_big = fresh and owner.sm['modelV2'].big
+  if published_big and (big_active is True or owner.sm['deviceState'].chestnutPresent):
+    owner.big_model_chestnut = True
+  model_unavailable = owner.big_model_active and owner.sm.seen['modelV2'] and not fresh
+  demoted = owner.big_model_active and fresh and not published_big
+  chestnut_lost = owner.big_model_active and owner.big_model_chestnut and not owner.sm['deviceState'].chestnutPresent
+  load_mismatch = big_active is True and fresh and not published_big
+  big_failed = (big_active is False and not published_big) or load_mismatch or model_unavailable or demoted or chestnut_lost
+  if big_failed and not owner.big_model_failed:
+    owner.events.add(log.OnroadEvent.EventName.bigModelFailed)
+  owner.big_model_failed = big_failed
+  # A successful load is not yet an executed frame; warmup output can be invalid.
+  if published_big:
+    owner.big_model_active = True
+  control_current = owner.sm.seen['carControl'] and owner.sm.alive['carControl'] and owner.sm.valid['carControl']
+  axis_active = control_current and (owner.sm['carControl'].latActive or owner.sm['carControl'].longActive)
+  if not owner.enabled and not axis_active and not model_unavailable:
+    owner.big_model_active = False
+    owner.big_model_chestnut = False
+  return big_failed
diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py
index 467a4a707c..31235a105d 100755
--- a/openpilot/selfdrive/selfdrived/selfdrived.py
+++ b/openpilot/selfdrive/selfdrived/selfdrived.py
@@ -29,7 +29,7 @@ from openpilot.selfdrive.car.car_events import CarEvents
 from openpilot.selfdrive.locationd.helpers import PoseCalibrator, Pose
 from openpilot.selfdrive.selfdrived.events import Events, ET, Alert, Priority as AlertPriority
 from openpilot.starpilot.longitudinal.force_stop_alert import HoldAlertState, EVENT_TYPE as FORCE_STOP_HOLD
-from openpilot.selfdrive.selfdrived.helpers import ExcessiveActuationCheck
+from openpilot.selfdrive.selfdrived.helpers import ExcessiveActuationCheck, update_big_model_status
 from openpilot.system.manager.process_health import driving_process_failures
 from openpilot.selfdrive.selfdrived.state import StateMachine
 from opendbc.car.gm.aol import native_bootstrap_supported
@@ -268,28 +268,7 @@ class SelfdriveD:
     return False
 
   def update_big_model_status(self) -> bool:
-    big_active = timed_params_read(self, "ChestnutActive")
-    fresh = self.sm.seen['modelV2'] and self.sm.alive['modelV2'] and self.sm.valid['modelV2']
-    published_big = fresh and self.sm['modelV2'].big
-    if published_big and (big_active is True or self.sm['deviceState'].chestnutPresent):
-      self.big_model_chestnut = True
-    model_unavailable = self.big_model_active and self.sm.seen['modelV2'] and not fresh
-    demoted = self.big_model_active and fresh and not published_big
-    chestnut_lost = self.big_model_active and self.big_model_chestnut and not self.sm['deviceState'].chestnutPresent
-    load_mismatch = big_active is True and fresh and not published_big
-    big_failed = (big_active is False and not published_big) or load_mismatch or model_unavailable or demoted or chestnut_lost
-    if big_failed and not self.big_model_failed:
-      self.events.add(EventName.bigModelFailed)
-    self.big_model_failed = big_failed
-    # A successful load is not yet an executed frame; warmup output can be invalid.
-    if published_big:
-      self.big_model_active = True
-    control_current = self.sm.seen['carControl'] and self.sm.alive['carControl'] and self.sm.valid['carControl']
-    axis_active = control_current and (self.sm['carControl'].latActive or self.sm['carControl'].longActive)
-    if not self.enabled and not axis_active and not model_unavailable:
-      self.big_model_active = False
-      self.big_model_chestnut = False
-    return big_failed
+    return update_big_model_status(self)
 
   def _update_steering_authority(self, CS):
     if not monitored_profile(self.CP):
diff --git a/openpilot/starpilot/galaxy/batch_routes.py b/openpilot/starpilot/galaxy/batch_routes.py
index 52a57bb45b..67dae4eb7f 100644
--- a/openpilot/starpilot/galaxy/batch_routes.py
+++ b/openpilot/starpilot/galaxy/batch_routes.py
@@ -147,3 +147,12 @@ def non_default_settings(self, settings_recovery):
       self.json(200, result)
     else:
       self.json(401, {'error': 'Sign in to Galaxy'})
+
+
+def preview_settings_recovery(payload, settings_recovery, identity):
+  if (not isinstance(payload, dict) or set(payload) != {'view', 'selected'} or
+      not isinstance(payload['view'], str) or len(payload['view']) > 64 or
+      not isinstance(payload['selected'], list) or not 1 <= len(payload['selected']) <= 128 or
+      any(type(index) is not int for index in payload['selected'])):
+    raise ValueError('Invalid settings recovery selection')
+  return settings_recovery().preview(payload['view'], payload['selected'], *identity)
diff --git a/openpilot/starpilot/galaxy/server.py b/openpilot/starpilot/galaxy/server.py
index bc7076293e..28629d6ea2 100644
--- a/openpilot/starpilot/galaxy/server.py
+++ b/openpilot/starpilot/galaxy/server.py
@@ -14,7 +14,7 @@ import threading
 import time
 from urllib.parse import parse_qs, unquote, urlsplit
 
-from openpilot.starpilot.galaxy.batch_routes import change_drive_state, delete_records, non_default_settings, preview_layout
+from openpilot.starpilot.galaxy.batch_routes import change_drive_state, delete_records, non_default_settings, preview_layout, preview_settings_recovery
 from openpilot.starpilot.galaxy.access import AccessStatus, default_owner
 from openpilot.starpilot.galaxy.auth import LocalSessions
 from openpilot.starpilot.galaxy.http_server import (_LocalHTTPHandler, _LocalHTTPServer, allowed_authority,
@@ -2349,12 +2349,7 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
               result = ToggleBackup(feature_settings(), layout_owner()).restore(
                 payload, identity, authorized=lambda: self.settings_session() == identity)
           elif path == '/api/settings/recovery/preview':
-            if (not isinstance(payload, dict) or set(payload) != {'view', 'selected'} or
-                not isinstance(payload['view'], str) or len(payload['view']) > 64 or
-                not isinstance(payload['selected'], list) or not 1 <= len(payload['selected']) <= 128 or
-                any(type(index) is not int for index in payload['selected'])):
-              raise ValueError('Invalid settings recovery selection')
-            result = settings_recovery().preview(payload['view'], payload['selected'], *identity)
+            result = preview_settings_recovery(payload, settings_recovery, identity)
           elif path == '/api/settings/recovery/confirm':
             if (not isinstance(payload, dict) or set(payload) != {'intent', 'confirmed'} or
                 not isinstance(payload['intent'], str) or len(payload['intent']) > 64 or payload['confirmed'] is not True):
diff --git a/openpilot/starpilot/galaxy/tests/test_http_server.py b/openpilot/starpilot/galaxy/tests/test_http_server.py
index 7b398ad891..aaf6cba3c8 100644
--- a/openpilot/starpilot/galaxy/tests/test_http_server.py
+++ b/openpilot/starpilot/galaxy/tests/test_http_server.py
@@ -1,4 +1,5 @@
 """Persistent read connections must never reuse an unread body or partial response."""
+from typing import Any
 
 import http.client
 from contextlib import closing
@@ -22,7 +23,7 @@ def server(tmp_path):
   directory.mkdir(parents=True)
   media = directory / 'init.mp4'
   media.write_bytes(b'x' * (65536 + 5))
-  state = {'requests': [], 'chunks': 0, 'revoke_after': None, 'expire_session': False, 'expired': False,
+  state: dict[str, Any] = {'requests': [], 'chunks': 0, 'revoke_after': None, 'expire_session': False, 'expired': False,
            'recording_checks': 0, 'change_recording': False}
 
   def owner(payload):
diff --git a/openpilot/starpilot/galaxy/tests/test_settings_defaults.mjs b/openpilot/starpilot/galaxy/tests/test_settings_defaults.mjs
index c94deb4d45..65f9d14015 100644
--- a/openpilot/starpilot/galaxy/tests/test_settings_defaults.mjs
+++ b/openpilot/starpilot/galaxy/tests/test_settings_defaults.mjs
@@ -29,7 +29,9 @@ assert.equal(requests.length, 4, "an unavailable default cannot be submitted")
 feed.stop()
 assert.ok(GalaxySettingRow.emits.includes("reset-default"))
 
-const render = compile(GalaxySettingRow.template, { decodeEntities: raw => raw })
+const renderTemplate = compile(GalaxySettingRow.template, { decodeEntities: raw => raw })
+const rowContext = context => ({ ...context, presentedRow: GalaxySettingRow.computed.presentedRow.call(context) })
+const render = (context, cache) => renderTemplate(rowContext(context), cache)
 function buttons(node) {
   if (!node || typeof node !== "object") return []
   return [...(node.type === "button" ? [node] : []), ...(Array.isArray(node.children) ? node.children.flatMap(buttons) : [])]
@@ -68,7 +70,7 @@ assert.equal(dependent.at(-1).data.rows[0].value, "On")
 source.stop()
 consumer.stop()
 
-assert.equal(GalaxySettingRow.computed.showDefault.call({ row }), true)
-assert.equal(GalaxySettingRow.computed.showDefault.call({ row: { ...row, value: 'On' } }), false)
-assert.equal(GalaxySettingRow.computed.showDefault.call({ row: { ...row, value: '0.9', defaultValue: '0.90' } }), false)
+assert.equal(GalaxySettingRow.computed.showDefault.call(rowContext({ row })), true)
+assert.equal(GalaxySettingRow.computed.showDefault.call(rowContext({ row: { ...row, value: 'On' } })), false)
+assert.equal(GalaxySettingRow.computed.showDefault.call(rowContext({ row: { ...row, value: '0.9', defaultValue: '0.90' } })), false)
 assert.equal(buttons(render({ ...context, showDefault: false }, [])).some(node => node.children === 'Default'), false)
diff --git a/openpilot/starpilot/replay/frames.py b/openpilot/starpilot/replay/frames.py
index 2fee61cb93..0f42b2c686 100644
--- a/openpilot/starpilot/replay/frames.py
+++ b/openpilot/starpilot/replay/frames.py
@@ -90,8 +90,8 @@ class ReplayFrameReader:
     if SESSION.fullmatch(session_id) is None or profile is not None and profile not in PROFILES:
       raise ValueError('Invalid replay frame identity')
     self.session_id, self.profile, self.last_key = session_id, profile, None
-    self.mm = None
-    self.fd = None
+    self.mm: mmap.mmap | None = None
+    self.fd: int | None = None
     fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | os.O_CLOEXEC)
     try:
       info = os.fstat(fd)
@@ -105,7 +105,7 @@ class ReplayFrameReader:
     self.fd = fd
 
   def latest(self) -> ReplayFrame | None:
-    if self.mm is None:
+    if self.mm is None or self.fd is None:
       return None
     try:
       fcntl.flock(self.fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
@@ -117,6 +117,8 @@ class ReplayFrameReader:
       fcntl.flock(self.fd, fcntl.LOCK_UN)
 
   def _latest(self):
+    if self.mm is None:
+      return None
     for _ in range(3):
       seq = SEQ.unpack_from(self.mm, 8)[0]
       if not seq or seq & 1:
diff --git a/openpilot/starpilot/replay/media.py b/openpilot/starpilot/replay/media.py
index 65d2c8be78..1a0ce893da 100644
--- a/openpilot/starpilot/replay/media.py
+++ b/openpilot/starpilot/replay/media.py
@@ -6,6 +6,7 @@ import math
 import os
 from pathlib import Path
 import re
+from typing import Any
 
 MAX_FRAGMENT_BYTES = 1024 * 1024
 MAX_MEDIA_BYTES = 8 * 1024 * 1024
@@ -108,8 +109,8 @@ class MediaStore:
     self.session_id, self.width, self.height = session_id, width, height
     self.epoch = 0
     self.codec = ''
-    self.init = None
-    self.fragments = deque()
+    self.init: dict[str, Any] | None = None
+    self.fragments: deque[dict[str, Any]] = deque()
     self.sequence = 0
 
   def _write(self, name: str, data: bytes):
@@ -141,6 +142,8 @@ class MediaStore:
   def publish(self, data: bytes, *, position: float, keyframe: bool):
     if not self.fragments and not keyframe:
       raise ValueError('Replay media must begin with a keyframe')
+    if self.init is None:
+      raise RuntimeError("Replay media epoch has not begun")
     self.sequence += 1
     item = {'sequence': self.sequence, 'mediaPath': self._write(f'{self.sequence}.m4s', data), 'length': len(data),
             'firstPosition': position, 'lastPosition': position, 'keyframe': bool(keyframe)}
diff --git a/openpilot/starpilot/replay/owner.py b/openpilot/starpilot/replay/owner.py
index d18ea58dfe..155b4cff03 100644
--- a/openpilot/starpilot/replay/owner.py
+++ b/openpilot/starpilot/replay/owner.py
@@ -22,6 +22,7 @@ import sys
 import threading
 import time
 import uuid
+from typing import cast
 
 from openpilot.starpilot.replay.client import MAX_REQUEST, MAX_RESPONSE, encode, parse, receive, replay_root
 from openpilot.tools.replay.display_clock import RECORD
@@ -338,6 +339,8 @@ class ReplayOwner:
     if self.status.get('target') != 'web' or request['epoch'] != self.status['epoch'] or not self.status.get('mediaReady'):
       raise Rejected('epoch_changed')
     from openpilot.starpilot.replay.media import media_descriptor
+    if self.session_dir is None:
+      raise Rejected("epoch_changed")
     return media_descriptor(self.session_dir, request['epoch'], request['sequence'], count)
 
   def _stop(self, reason):
@@ -389,7 +392,7 @@ class ReplayOwner:
       if (type(ranges) is not list or len(ranges) > 512 or
           any(type(item) is not list or len(item) != 2 or not all(_finite(number) for number in item) or
               not value['start'] <= item[0] <= item[1] <= value['end'] for item in ranges) or
-          any(before[1] > after[0] for before, after in zip(ranges, ranges[1:], strict=False))):
+          any(before[1] > after[0] for before, after in zip(cast(list[list[float]], ranges), cast(list[list[float]], ranges)[1:], strict=False))):
         raise Rejected('worker_failed')
       self.status['seekRanges'] = ranges
       warnings = value.get('sourceWarnings', [])
@@ -524,6 +527,7 @@ class ReplayOwner:
       with self.lock:
         self.child, self.child_birth = child, _process_birth(child.pid)
       self._record_session(directory, child)
+      assert child.stdout is not None and child.stderr is not None and child.stdin is not None
       for file in (child.stdout, child.stderr, child.stdin):
         os.set_blocking(file.fileno(), False)
       stdout = bytearray()
diff --git a/openpilot/starpilot/replay/render.py b/openpilot/starpilot/replay/render.py
index 80e968e013..d39dbbf7c2 100644
--- a/openpilot/starpilot/replay/render.py
+++ b/openpilot/starpilot/replay/render.py
@@ -13,6 +13,7 @@ import subprocess
 import sys
 import threading
 import time
+from typing import cast
 
 from openpilot.starpilot.replay.frames import ReplayFramePublisher, SESSION
 from openpilot.starpilot.replay.media import MediaMux, MediaStore
@@ -299,7 +300,7 @@ class Compositor:
         self.resources.callback(rl.unload_render_texture, self.output)
       self.readback = FrameReadback(width * height * (3 if nv12 else 8) // 2, asynchronous=True)
       self.resources.callback(self.readback.close)
-      self.meta = None
+      self.meta: tuple[int, float, int] | None = None
       self.epoch = None
       self.last_route_frame = None
       self.primary_ready = False
@@ -342,14 +343,15 @@ class Compositor:
     updates = original | {name: sm.seen[name] and receipt != self.render_receipts.get(name)
                           for name, receipt in receipts.items()}
     sample = self.ui.replay_sample
-    updates['extrinsicsCalibration'] = bool(updates['extrinsicsCalibration'] and sample and sample.valid and
+    updates['extrinsicsCalibration'] = bool(updates['extrinsicsCalibration'] and sample and sample.valid and sample.now_ns is not None and
                                          sm.valid['extrinsicsCalibration'] and
                                          0 < sm.logMonoTime['extrinsicsCalibration'] <= sample.now_ns)
     flags = DrawUpdates(updates)
     sm.updated = flags
     try:
       if self.profile == 'compact':
-        self.camera.render_camera_model_layer(rect, camera_view=state.appearance.camera_view,
+        from openpilot.selfdrive.ui.mici.onroad.augmented_road_view import AugmentedRoadView as CompactRoadView
+        cast(CompactRoadView, self.camera).render_camera_model_layer(rect, camera_view=state.appearance.camera_view,
                                              reverse_driver=state.reverse_driver_camera,
                                              lead_indicator=state.appearance.show_lead_indicator and state.lead_indicator_source_fresh,
                                              lead_info_mode=state.appearance.lead_info_mode, lead_info_metric=state.appearance.lead_info_metric,
@@ -382,6 +384,8 @@ class Compositor:
              VisionStreamType.VISION_STREAM_WIDE_ROAD: 'wideRoadEncodeIdx',
              VisionStreamType.VISION_STREAM_CABIN: 'cabinEncodeIdx'}
     name = names.get(self.camera.stream_type)
+    if name is None:
+      return False
     sm, sample = self.ui.sm, self.ui.replay_sample
     state = sm[name] if name is not None and sm.seen[name] and sm.valid[name] else None
     return bool(sample and sample.valid and state is not None and frame_matches(
@@ -413,6 +417,7 @@ class Compositor:
       if position['enabled']:
         width, height = widget_size(state.customization, self.profile, key)
         positions[side] = Rect(position['x'], position['y'], width, height)
+    assert saved.on_blinker is not None and saved.on_bsm is not None and saved.invert is not None
     status = self.pip.render(rect, saved.mask, signals, enabled=self.cabin, on_blinker=saved.on_blinker,
                              on_bsm=saved.on_bsm, invert=saved.invert, now=now / 1e9, placements=positions, submit=submit)
     if not self.cabin or status not in ('rendered', 'inactive'):
@@ -507,6 +512,8 @@ class Compositor:
     from openpilot.starpilot.system.android_auto import gpu_nv12
     from openpilot.starpilot.ui.shell import ShellMode
     rl = self.rl
+    if sample.now_ns is None:
+      return False
     self.context.begin_frame(sample.now_ns / 1e9)
     self.placement.maintain(now_ns / 1e9)
     snapshot = self.adapter.build(ShellMode.ONROAD)
@@ -547,7 +554,7 @@ class Compositor:
     try:
       pixels = self.readback.finish()
       sample = self.ui.replay_clock.sample()
-      yield (pixels, self.meta) if sample.valid and sample.epoch == self.meta[0] else None
+      yield (pixels, self.meta) if self.meta is not None and sample.valid and sample.epoch == self.meta[0] else None
     finally:
       self.readback.release()
 
@@ -679,12 +686,12 @@ def run(session_dir: Path, route_id: str, data_dir: Path, target: str, profile:
         if engine.command({'op': 'seek', 'position': status['position'], 'requestId': status.get('lastRequestId') or '0' * 32}, reprime=True):
           primed_clients.update(unprimed)
           valid = False
-      if valid and getattr(core.ui, 'replay_needs_reprime', False):
+      if valid and status is not None and getattr(core.ui, 'replay_needs_reprime', False):
         if engine.command({'op': 'seek', 'position': status['position'], 'requestId': status.get('lastRequestId') or '0' * 32}, reprime=True):
           core.ui.replay_needs_reprime = False
           valid = False
       interval = FRAME_INTERVAL
-      if valid and not core.readback.pending and now - last_draw >= interval:
+      if valid and status is not None and sample.now_ns is not None and not core.readback.pending and now - last_draw >= interval:
         origin = status.get('routeStartNs')
         if type(origin) is not int:
           raise RuntimeError('Replay engine did not expose its route clock origin')
diff --git a/openpilot/starpilot/replay/tests/test_catalog.py b/openpilot/starpilot/replay/tests/test_catalog.py
index efa1eaa0b2..a2bee48fb2 100644
--- a/openpilot/starpilot/replay/tests/test_catalog.py
+++ b/openpilot/starpilot/replay/tests/test_catalog.py
@@ -229,6 +229,7 @@ class ReplayCatalogTest(unittest.TestCase):
     uploader.listdir_by_creation = lambda root: sorted(os.listdir(root))
     path = Path(__file__).resolve().parents[3] / 'system/loggerd/deleter.py'
     spec = importlib.util.spec_from_file_location('replay_fixture_deleter', path)
+    assert spec is not None and spec.loader is not None
     deleter = importlib.util.module_from_spec(spec)
     with patch.dict(sys.modules, {'openpilot.system.loggerd.uploader': uploader}):
       spec.loader.exec_module(deleter)
diff --git a/openpilot/starpilot/replay/tests/test_frames.py b/openpilot/starpilot/replay/tests/test_frames.py
index 73df499955..a52d5b038f 100644
--- a/openpilot/starpilot/replay/tests/test_frames.py
+++ b/openpilot/starpilot/replay/tests/test_frames.py
@@ -75,6 +75,7 @@ def test_kernel_lock_prevents_copying_pixels_while_writer_is_active(tmp_path):
     assert reader.latest() is None
     fcntl.flock(publisher.fd, fcntl.LOCK_UN)
     assert reader.latest().data == b'a' * 32
+    assert reader.fd is not None
     fcntl.flock(reader.fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
     with pytest.raises(RuntimeError, match='consumer stalled'):
       publisher.publish(b'b' * 32, epoch=2, position=1, captured_ns=2)
diff --git a/openpilot/starpilot/replay/tests/test_media.py b/openpilot/starpilot/replay/tests/test_media.py
index 31a3277027..654cb783b0 100644
--- a/openpilot/starpilot/replay/tests/test_media.py
+++ b/openpilot/starpilot/replay/tests/test_media.py
@@ -93,10 +93,11 @@ def test_descriptor_rejects_hard_linked_metadata(tmp_path):
 def test_mux_copy_has_explicit_pts_and_independent_initialization(tmp_path):
   root = Path(__file__).parents[4]
   library = Path(os.environ.get('SP_REPLAY_TEST_MUX_LIBRARY', root / 'openpilot/tools/replay/libmedia_mux.so'))
-  ffmpeg = root / '.venv/lib/python3.12/site-packages/ffmpeg/install/bin/ffmpeg'
+  import ffmpeg as ffmpeg_package
+  ffmpeg = Path(ffmpeg_package.__file__).resolve().parent / 'install/bin/ffmpeg'
   ffprobe = ffmpeg.with_name('ffprobe')
   if not library.exists() or not ffmpeg.exists() or not ffprobe.exists():
-    pytest.skip('Host replay muxer and bundled FFmpeg must be built')
+    pytest.skip(reason='Host replay muxer and bundled FFmpeg must be built')
   source = tmp_path / 'fixture.ts'
   subprocess.run([str(ffmpeg), '-nostdin', '-hide_banner', '-loglevel', 'error', '-f', 'rawvideo', '-pixel_format', 'yuv420p',
                   '-video_size', '32x18', '-framerate', '20', '-i', 'pipe:0',
diff --git a/openpilot/starpilot/replay/tests/test_owner.py b/openpilot/starpilot/replay/tests/test_owner.py
index a24e15f0de..d0dd7d9f42 100644
--- a/openpilot/starpilot/replay/tests/test_owner.py
+++ b/openpilot/starpilot/replay/tests/test_owner.py
@@ -355,7 +355,7 @@ def test_real_machine_uses_owner_created_clock_and_cleans_it(tmp_path, monkeypat
 
   binary = Path(BASEDIR) / 'openpilot/tools/replay/replay'
   if not binary.is_file():
-    pytest.skip('Build the isolated replay engine first')
+    pytest.skip(reason='Build the isolated replay engine first')
   routes = tmp_path / 'routes'
   routes.mkdir()
   segment(routes, 0, [.1, .2, 10], compressed=True)
diff --git a/openpilot/starpilot/replay/tests/test_render.py b/openpilot/starpilot/replay/tests/test_render.py
index 31185f2d7a..efd268eb1a 100644
--- a/openpilot/starpilot/replay/tests/test_render.py
+++ b/openpilot/starpilot/replay/tests/test_render.py
@@ -50,7 +50,7 @@ def test_camera_matches_recorded_identity_across_distinct_clock_domains():
 
 def test_paused_same_route_clock_never_advances_filters_or_gpu():
   core = object.__new__(Compositor)
-  core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100))
+  vars(core)['ui'] = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100))
   core.last_route_frame = (2, 100)
   core.context = Mock()
   assert core.draw(2, 0, 9_000_000_000) is False
@@ -65,7 +65,7 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea
     viewport_width: int = 0
 
   core = object.__new__(Compositor)
-  core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100),
+  vars(core)['ui'] = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100),
                             sm=SimpleNamespace(seen={'extrinsicsCalibration': False}))
   core.last_route_frame = None
   core.context = core.placement = Mock()
@@ -74,10 +74,10 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea
   core._primary_matches = Mock(return_value=frame_ready)
   core._optional_placeholder_ready = Mock(return_value=False)
   core.rl = Mock()
-  core.gui = SimpleNamespace(_render_prepare=[], _frame=0)
+  vars(core)['gui'] = SimpleNamespace(_render_prepare=[], _frame=0)
   core.content = Mock()
   core.scale = 1
-  core.profile = SimpleNamespace(size=(1600, 800))
+  vars(core)['profile'] = SimpleNamespace(size=(1600, 800))
   core.view = Mock()
   core.view.render.side_effect = lambda state: setattr(core, 'primary_ready', state.camera_available)
   core.output = None
@@ -242,7 +242,7 @@ def test_epoch_reset_preserves_native_camera_stream_and_eligible_paused_frame(mo
   sm.__getitem__ = Mock(return_value=idx)
   sm.seen = sm.valid = {'narrowRoadEncodeIdx': True}
   sm.logMonoTime = {'narrowRoadEncodeIdx': 9_020_000_000}
-  core.ui = SimpleNamespace(replay_sample=SimpleNamespace(valid=True, boot_ns=9_050_000_000), sm=sm)
+  vars(core)['ui'] = SimpleNamespace(replay_sample=SimpleNamespace(valid=True, boot_ns=9_050_000_000), sm=sm)
   core.render_receipts = {'modelV2': 1}
   core.reset_epoch(3)
   assert core.render_receipts == {}
@@ -301,7 +301,7 @@ def test_early_camera_subscriber_is_polled_before_paused_reprime():
   # The first receive synchronizes with a publisher created after this client.
   client.recv.side_effect = [None, frame]
   core = object.__new__(Compositor)
-  core.ui = Mock(replay_sample=sample, sm=sm)
+  vars(core)['ui'] = Mock(replay_sample=sample, sm=sm)
   core.epoch, core.cabin = 1, False
   core.camera = Mock(client=client, frame=None, stream_type=VisionStreamType.VISION_STREAM_NARROW_ROAD)
   core.camera._ensure_connection.return_value = True
@@ -335,7 +335,7 @@ def test_camera_reprime_cannot_overwrite_an_unacknowledged_user_action():
   import threading
   engine = object.__new__(Engine)
   engine.write_lock = threading.Lock()
-  engine.process = SimpleNamespace(stdin=io.StringIO())
+  vars(engine)['process'] = SimpleNamespace(stdin=io.StringIO())
   engine.last_request_id = engine.seek_request = None
   request = {'op': 'seek', 'position': 65, 'requestId': 'a' * 32}
   assert engine.command(request)
@@ -385,8 +385,8 @@ def test_replay_draw_retains_updates_until_stock_consumers_read_them(monkeypatch
   camera.available_streams, camera.frame = {camera.stream_type}, object()
   camera.model_renderer = camera._model_renderer = renderer
   core = object.__new__(Compositor)
-  core.ui, core.camera = ui, camera
-  core.profile = 'compact' if profile == 'mici.onroad' else 'large'
+  vars(core)['ui'], core.camera = ui, camera
+  vars(core)['profile'] = 'compact' if profile == 'mici.onroad' else 'large'
   core.cabin = core.wide = False
   core.render_receipts = {}
   core._primary_matches = Mock(return_value=True)
@@ -480,9 +480,9 @@ def test_finished_frame_is_borrowed_until_delivery_finishes(valid, epoch):
   pixels = memoryview(bytearray(b'pixels'))
   core.readback = Mock(pending=True)
   core.readback.finish.return_value = pixels
-  core.ui = SimpleNamespace(replay_clock=Mock())
+  vars(core)['ui'] = SimpleNamespace(replay_clock=Mock())
   core.ui.replay_clock.sample.return_value = SimpleNamespace(valid=valid, epoch=epoch)
-  core.meta = 1, 0., 100
+  vars(core)['meta'] = 1, 0., 100
   with pytest.raises(RuntimeError, match='delivery failed'):
     with core.finish() as finished:
       core.readback.release.assert_not_called()
diff --git a/openpilot/starpilot/ui/pip_render.py b/openpilot/starpilot/ui/pip_render.py
index f03a59a57f..953ad28d07 100644
--- a/openpilot/starpilot/ui/pip_render.py
+++ b/openpilot/starpilot/ui/pip_render.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import time
+from typing import Protocol
 
 import pyray as rl
 
@@ -26,10 +27,18 @@ def _external_shader(fragment: str) -> str:
   return result.replace(yuv, "vec3 rgb = pow(texture(texture0, uv).rgb, vec3(1.0/1.28));")
 
 
+class PiPStreamSource(Protocol):
+  @property
+  def generation(self) -> int: ...
+  def set_active(self, active: bool) -> None: ...
+  def poll(self, now: float) -> object | None: ...
+  def release_retired(self) -> None: ...
+
+
 class PiPRenderer:
   """One renderer; close it on onroad exit or when the owning shell closes."""
 
-  def __init__(self, shape: str, stream: PiPStream | None = None, *, frame_availability=None):
+  def __init__(self, shape: str, stream: PiPStreamSource | None = None, *, frame_availability=None):
     if shape not in ("bubble", "curved"):
       raise ValueError("unknown PiP shape")
     self.frame_availability = frame_availability if frame_availability is not None else CameraAvailability()
diff --git a/openpilot/starpilot/ui/replay_settings.py b/openpilot/starpilot/ui/replay_settings.py
index 465173e31a..3d60c26f5d 100644
--- a/openpilot/starpilot/ui/replay_settings.py
+++ b/openpilot/starpilot/ui/replay_settings.py
@@ -5,6 +5,7 @@ from html import escape, unescape
 import math
 import time
 import uuid
+from typing import cast
 
 import pyray as rl
 
@@ -133,7 +134,10 @@ class ReplayController:
     if type(position) not in (int, float) or not math.isfinite(position):
       return False
     start, end = self.status.get('start'), self.status.get('end')
-    if type(start) not in (int, float) or type(end) not in (int, float) or end <= start:
+    if type(start) not in (int, float) or type(end) not in (int, float):
+      return False
+    start, end = cast(float, start), cast(float, end)
+    if end <= start:
       return False
     position = max(start, min(end, position))
     ranges = self.status.get('seekRanges')
@@ -283,7 +287,7 @@ class ReplayPage(Widget):
     self.last_touch = time.monotonic()
     self.stopping = False
     self._epoch = None
-    self.stack_widget = self
+    self.stack_widget: Widget = self
 
   def show_event(self):
     super().show_event()
@@ -620,7 +624,8 @@ class ReplayPage(Widget):
       start, end = self.controller.status.get('start') or 0, self.controller.status.get('end') or 0
       self.seek_preview = start + max(0, min(1, (x - left) / width)) * (end - start)
 
-  def _handle_mouse_event(self, event):
+  def _handle_mouse_event(self, mouse_event):
+    event = mouse_event
     if event.left_released and not rl.check_collision_point_rec(event.pos, self._hit_rect):
       self.press = self.seek_preview = None
       self.input.cancel()
@@ -682,7 +687,8 @@ class ReplayCompactPage(NavWidget):
   def _back_enabled(self):
     return self.content.page != 'viewer'
 
-  def _handle_mouse_event(self, event):
+  def _handle_mouse_event(self, mouse_event):
+    event = mouse_event
     was_dismissing = self._playing_dismiss_animation
     super()._handle_mouse_event(event)
     if event.left_released and not was_dismissing and self._playing_dismiss_animation and self.content.page == 'detail':
diff --git a/openpilot/starpilot/ui/tests/test_onroad_parity.py b/openpilot/starpilot/ui/tests/test_onroad_parity.py
index 82e7715e78..dcde5ed61f 100644
--- a/openpilot/starpilot/ui/tests/test_onroad_parity.py
+++ b/openpilot/starpilot/ui/tests/test_onroad_parity.py
@@ -69,7 +69,7 @@ def test_aol_model_geometry_uses_fresh_lateral_display_context(monkeypatch):
       return values[key]
   source = SubMaster()
   source.recv_frame, source.updated, source.valid = sm.recv_frame, sm.updated, sm.valid
-  ui = NS(sm=source, started_frame=1, status=UIStatus.DISENGAGED)
+  ui = NS(sm=source, started_frame=1, status=UIStatus.DISENGAGED, display_time_ns=lambda: 1_000_000_000)
   with patch.object(native_model, 'ui_state', ui):
     renderer._render(rl.Rectangle(0, 0, 476, 240))
     renderer._draw_lane_lines.assert_not_called()
@@ -94,7 +94,7 @@ def test_aol_lane_and_path_colors_do_not_use_disengaged_black(monkeypatch):
   renderer._rainbow_path.refresh_enabled.return_value = False
   renderer._experimental_mode = True
   monkeypatch.setattr(renderer, '_exp_gradient', NS(colors=[]), raising=False)
-  ui = NS(status=UIStatus.DISENGAGED, params=Mock())
+  ui = NS(status=UIStatus.DISENGAGED, params=Mock(), display_time_ns=lambda: 1_000_000_000)
   with patch.object(native_model, 'ui_state', ui), patch.object(native_model, 'draw_polygon') as polygon:
     assert renderer._get_ll_color(0.7, True, True).r == 0
     renderer._draw_path({'longitudinalPlan': NS(allowThrottle=True)})
diff --git a/openpilot/starpilot/ui/tests/test_rainbow_path.py b/openpilot/starpilot/ui/tests/test_rainbow_path.py
index e94ad30598..1b0229df48 100644
--- a/openpilot/starpilot/ui/tests/test_rainbow_path.py
+++ b/openpilot/starpilot/ui/tests/test_rainbow_path.py
@@ -65,23 +65,23 @@ class TestRainbowPath(unittest.TestCase):
           def __getitem__(self, key):
             return self.values[key]
         sm = SubMaster(messages)
-        ui = NS(status=compact.UIStatus.ENGAGED, started_frame=5)
-        with patch.object(module, "ui_state", ui), patch.object(module.time, "monotonic_ns", return_value=1_000_000_000), \
+        ui = NS(status=compact.UIStatus.ENGAGED, started_frame=5, display_time_ns=Mock())
+        with patch.object(module, "ui_state", ui), patch.object(ui, "display_time_ns", return_value=1_000_000_000), \
              patch.object(module, "draw_polygon") as draw:
           renderer._draw_path(sm)
           self.assertIs(draw.call_args.kwargs["gradient"], renderer._exp_gradient)
         renderer.road_style = {"pathMode": "rainbow"}
-        with patch.object(module, "ui_state", ui), patch.object(module.time, "monotonic_ns", return_value=2_000_000_000), \
+        with patch.object(module, "ui_state", ui), patch.object(ui, "display_time_ns", return_value=2_000_000_000), \
              patch.object(module, "draw_polygon") as draw:
           renderer._draw_path(sm)
           self.assertEqual(len(draw.call_args.kwargs["gradient"].colors), 12)
         sm.recv_frame["carState"] = 4
-        with patch.object(module, "ui_state", ui), patch.object(module.time, "monotonic_ns", return_value=2_100_000_000), \
+        with patch.object(module, "ui_state", ui), patch.object(ui, "display_time_ns", return_value=2_100_000_000), \
              patch.object(module, "draw_polygon"):
           renderer._draw_path(sm)
         self.assertEqual(renderer._rainbow_path.phase_deg, 0.0)
         sm.recv_frame["carState"] = 5
-        with patch.object(module, "ui_state", ui), patch.object(module.time, "monotonic_ns", return_value=2_200_000_000), \
+        with patch.object(module, "ui_state", ui), patch.object(ui, "display_time_ns", return_value=2_200_000_000), \
              patch.object(module, "draw_polygon"):
           renderer._draw_path(sm)
         self.assertGreater(renderer._rainbow_path.phase_deg, 0.0)
diff --git a/openpilot/starpilot/ui/tests/test_replay_settings.py b/openpilot/starpilot/ui/tests/test_replay_settings.py
index 3be5070f12..dccc498b8a 100644
--- a/openpilot/starpilot/ui/tests/test_replay_settings.py
+++ b/openpilot/starpilot/ui/tests/test_replay_settings.py
@@ -1,4 +1,5 @@
 """Fake transport exercises native ownership, cancellation and passive input."""
+from typing import Any
 
 from types import SimpleNamespace as NS
 import threading
@@ -19,7 +20,7 @@ ROUTE = {'routeId': '00000001--0123456789', 'displayName': 'Recorded route', 'se
 class Backend:
   def __init__(self):
     self.calls = []
-    self.status = {'schemaVersion': 1, 'ownerId': 'a' * 32, 'sessionId': None, 'state': 'idle',
+    self.status: dict[str, Any] = {'schemaVersion': 1, 'ownerId': 'a' * 32, 'sessionId': None, 'state': 'idle',
                    'routeId': None, 'profile': 'large', 'revision': 0, 'epoch': 0,
                    'position': 0.0, 'start': 0.0, 'end': 120.0, 'rate': 1.0, 'paused': True,
                    'frameReady': False, 'errorCode': None, 'owned': False, 'offroad': True}
diff --git a/openpilot/starpilot/ui/tests/test_runtime_panel_actions.py b/openpilot/starpilot/ui/tests/test_runtime_panel_actions.py
index f7647e524c..c0bc80e8f2 100644
--- a/openpilot/starpilot/ui/tests/test_runtime_panel_actions.py
+++ b/openpilot/starpilot/ui/tests/test_runtime_panel_actions.py
@@ -644,7 +644,8 @@ class TestRuntimePanelActions(unittest.TestCase):
 
   def test_compact_galaxy_is_on_main_menu_and_device_keeps_native_rows(self):
     from openpilot.selfdrive.ui.mici.layouts.settings.device import device_layout as compact_device
-    from openpilot.starpilot.ui import display_compact
+    from openpilot.starpilot.ui import display_compact, replay_settings
+    replay_button = Mock()
     display_entry = object()
     panel = NS(_scroller=NS(add_widgets=Mock()))
     layout = runtime_app.StarMiciMainLayout.__new__(runtime_app.StarMiciMainLayout)
@@ -653,14 +654,21 @@ class TestRuntimePanelActions(unittest.TestCase):
     object.__setattr__(layout, "star", NS(galaxy_flow=galaxy))
     with patch.object(compact_device, "DeviceLayoutMici", return_value=panel) as built, \
          patch.object(display_compact.DisplayCompact, "entry_button", return_value=display_entry), \
+         patch.object(replay_settings, "BigButton", return_value=replay_button) as replay_constructor, \
          patch.object(runtime_app.gui_app, "push_widget") as pushed:
       layout._open_compact_destination(Destination.DEVICE)
     built.assert_called_once_with()
-    panel._scroller.add_widgets.assert_called_once_with([display_entry])
+    panel._scroller.add_widgets.assert_called_once_with([display_entry, replay_button])
     pushed.assert_called_once_with(panel)
     layout._open_compact_destination(Destination.GALAXY)
     galaxy.open_compact.assert_called_once_with()
 
+    replay_constructor.assert_called_once_with("replay", "recorded driving routes on this device")
+    replay_button.set_click_callback.assert_called_once()
+    with patch.object(replay_settings, "open_replay") as open_replay:
+      replay_button.set_click_callback.call_args.args[0]()
+    open_replay.assert_called_once_with(layout.star)
+
   def test_compact_toggles_and_software_keep_only_native_rows(self):
     from openpilot.selfdrive.ui.mici.layouts.settings import toggles, software
     panels = []
@@ -682,6 +690,8 @@ class TestRuntimePanelActions(unittest.TestCase):
     self.assertEqual([call.args[0] for call in pushed.call_args_list], panels)
 
   def test_compact_model_visuals_and_pair_use_existing_owners(self):
+    from openpilot.starpilot.ui import replay_settings
+    replay_button = Mock()
     from openpilot.starpilot.ui import appearance_compact, models_compact, display_compact
     from openpilot.selfdrive.ui.mici.layouts.settings.device import device_layout as compact_device
     layout = runtime_app.StarMiciMainLayout.__new__(runtime_app.StarMiciMainLayout)
@@ -692,6 +702,7 @@ class TestRuntimePanelActions(unittest.TestCase):
          patch.object(appearance_compact.AppearanceCompact, "open") as visuals_open, \
          patch.object(compact_device, "DeviceLayoutMici", return_value=panel), \
          patch.object(display_compact.DisplayCompact, "entry_button", return_value="display entry"), \
+         patch.object(replay_settings, "BigButton", return_value=replay_button) as replay_constructor, \
          patch.object(runtime_app.gui_app, "push_widget") as pushed:
       layout._open_compact_destination(Destination.DRIVING_MODEL)
       layout._open_compact_destination(Destination.APPEARANCE)
@@ -700,7 +711,13 @@ class TestRuntimePanelActions(unittest.TestCase):
     visuals_open.assert_called_once_with()
     pushed.assert_called_once_with(panel)
     panel.scroll_to_pairing.assert_called_once_with()
-    panel._scroller.add_widgets.assert_called_once_with(["display entry"])
+    panel._scroller.add_widgets.assert_called_once_with(["display entry", replay_button])
+
+    replay_constructor.assert_called_once_with("replay", "recorded driving routes on this device")
+    replay_button.set_click_callback.assert_called_once()
+    with patch.object(replay_settings, "open_replay") as open_replay:
+      replay_button.set_click_callback.call_args.args[0]()
+    open_replay.assert_called_once_with(layout.star)
 
   def test_compact_connection_can_be_viewed_while_started_but_not_mutated(self):
     with tempfile.TemporaryDirectory() as directory:
diff --git a/openpilot/starpilot/ui/tests/test_ui_started_authority.py b/openpilot/starpilot/ui/tests/test_ui_started_authority.py
index 7e81bcae16..9340d2a0bf 100644
--- a/openpilot/starpilot/ui/tests/test_ui_started_authority.py
+++ b/openpilot/starpilot/ui/tests/test_ui_started_authority.py
@@ -120,7 +120,7 @@ def test_started_presentation_does_not_replace_existing_message_freshness():
 def replay_owner(sample):
   from openpilot.starpilot.ui.tests.test_runtime_snapshot import SubMasterFake
   ui = object.__new__(UIState)
-  ui.sm = SubMasterFake()
+  vars(ui)['sm'] = SubMasterFake()
   ui.sm.sock = {name: object() for name in ('carControl', 'carState', 'selfdriveState')}
   ui.sm.static_freq_services = set(ui.sm.sock)
   ui.sm.data = ui.sm.messages
@@ -244,6 +244,7 @@ def test_replay_boot_time_camera_primer_retention_and_freshness():
   ui.sm.sock['cabinCameraState'] = object()
   ui.sm.static_freq_services.add('cabinCameraState')
   ui.sm.put('cabinCameraState', NS(frameId=42))
+  assert sample.boot_ns is not None
   ui.sm.logMonoTime['cabinCameraState'] = sample.boot_ns - 1_000_000
   ui.sm.replay_sample = sample
   ui._update_replay_alive(sample)
diff --git a/openpilot/tools/replay/tests/test_machine.py b/openpilot/tools/replay/tests/test_machine.py
index 880b58a97b..f30b8bbe4a 100644
--- a/openpilot/tools/replay/tests/test_machine.py
+++ b/openpilot/tools/replay/tests/test_machine.py
@@ -97,6 +97,7 @@ class Machine:
     self.process = subprocess.Popen(self.command, env=self.env, stdin=subprocess.PIPE,
                                     stdout=subprocess.PIPE, stderr=self.stderr, text=True)
     def reader():
+      assert self.process.stdout is not None
       for line in self.process.stdout:
         try:
           self.statuses.put(json.loads(line))
@@ -283,6 +284,7 @@ class TestMachine(unittest.TestCase):
       for name, levels in (("a", (30, 90, 120)), ("b", (200, 220, 240))):
         raw, clip = root / (name + ".yuv"), root / (name + ".ts")
         raw.write_bytes(b"".join(bytes([level]) * (32 * 18) + bytes([128]) * (32 * 18 // 2) for level in levels))
+        assert FFMPEG is not None
         subprocess.run([FFMPEG, "-hide_banner", "-loglevel", "error", "-f", "rawvideo", "-pix_fmt", "yuv420p",
                         "-s", "32x18", "-r", "20", "-i", str(raw), "-c:v", "libx264", "-qp", "0", "-bf", "0",
                         "-g", "100", "-tune", "zerolatency", "-f", "mpegts", str(clip)], check=True, timeout=10)