Add Replay compositor trace

openpilot/starpilot/replay/render.py

@@ -507,13 +507,28 @@ class Compositor:
     self.render_receipts = {}
     self.last_route_frame = None
 
-  def draw(self, epoch: int, position: float, now_ns: int):
+  def _draw_rejection(self, epoch: int):
     sample = self.ui.replay_sample
-    if sample is None or sample.now_ns is None:
+    if sample is None or sample.now_ns is None or not sample.valid:
+      return 'invalid_clock'
+    if sample.epoch != epoch:
+      return 'wrong_epoch'
+    if (epoch, sample.now_ns) == self.last_route_frame:
+      return 'unchanged_clock'
+    # OnroadView skips the camera layer when unavailable, so this draw cannot publish.
+    if not (self._primary_matches() or self._optional_placeholder_ready()):
+      return 'camera_not_ready'
+    return None
+
+  def _prepare(self):
+    for prepare in tuple(self.gui._render_prepare):
+      prepare()
+
+  def draw(self, epoch: int, position: float, now_ns: int):
+    if self._draw_rejection(epoch) is not None:
       return False
+    sample = self.ui.replay_sample
     key = epoch, sample.now_ns
-    if key == self.last_route_frame:
-      return False
     from openpilot.starpilot.system.android_auto import gpu_nv12
     from openpilot.starpilot.ui.shell import ShellMode
     rl = self.rl
@@ -521,8 +536,7 @@ class Compositor:
     self.placement.maintain(now_ns / 1e9)
     snapshot = self.adapter.build(ShellMode.ONROAD)
     self.primary_ready = False
-    for prepare in tuple(self.gui._render_prepare):
-      prepare()
+    self._prepare()
     rl.begin_texture_mode(self.content)
     try:
       rl.clear_background(rl.BLACK)

openpilot/starpilot/replay/tests/test_render.py

@@ -1,5 +1,5 @@
 from contextlib import nullcontext
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import Mock
@@ -9,6 +9,7 @@ import pytest
 from openpilot.starpilot.replay import render
 from openpilot.starpilot.replay.render import Compositor, Engine, camera_options, frame_matches, parse_wall_offset
 from openpilot.starpilot.ui.presentation import Profile
+from openpilot.tools.replay.display_clock import ClockSample
 
 
 def test_camera_selection_uses_selected_sources_and_current_preferences():
@@ -52,7 +53,7 @@ def test_camera_matches_recorded_identity_across_distinct_clock_domains():
 
 def test_paused_same_route_clock_never_advances_filters_or_gpu():
   core: Any = object.__new__(Compositor)
-  core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100))
+  core.ui = SimpleNamespace(replay_sample=ClockSample(100, 100, 100, epoch=2, paused=True, valid=True))
   core.last_route_frame = (2, 100)
   core.context = Mock()
   assert core.draw(2, 0, 9_000_000_000) is False
@@ -76,7 +77,7 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea
     viewport_width: int = 0
 
   core: Any = object.__new__(Compositor)
-  core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100),
+  core.ui = SimpleNamespace(replay_sample=ClockSample(100, 100, 100, epoch=2, valid=True),
                             sm=SimpleNamespace(seen={'extrinsicsCalibration': False}))
   core.last_route_frame = None
   core.context = core.placement = Mock()
@@ -95,12 +96,178 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea
   core.converter = Mock()
   core.readback = Mock()
   assert core.draw(2, 0, 9_000_000_000) is frame_ready
+  if not frame_ready:
+    core.context.begin_frame.assert_not_called()
+    core.adapter.build.assert_not_called()
+    core.view.render.assert_not_called()
+    core.readback.start.assert_not_called()
+    return
   drawn = core.view.render.call_args.args[0]
   assert drawn.camera_available is frame_ready
   assert core.adapter.build.return_value.onroad.camera_available is False
   assert core.readback.start.call_count == int(frame_ready)
 
 
+@pytest.fixture(params=(Profile.LARGE, Profile.COMPACT))
+def draw_core(request):
+  from openpilot.cereal.visionipc import VisionStreamType
+  from openpilot.starpilot.ui.onroad_state import OnroadState, SpeedLimitObservation
+
+  class Messages(dict):
+    pass
+
+  idx = SimpleNamespace(frameId=5, timestampSof=9_000_000_000, timestampEof=9_010_000_000)
+  sm = Messages(narrowRoadEncodeIdx=idx, wideRoadEncodeIdx=idx)
+  sm.seen = dict.fromkeys(('narrowRoadEncodeIdx', 'wideRoadEncodeIdx', 'modelV2', 'radarState', 'carParams',
+                           'carState', 'extrinsicsCalibration'), True)
+  sm.valid = dict(sm.seen)
+  sm.logMonoTime = dict.fromkeys(sm, 9_020_000_000)
+  sm.logMonoTime['extrinsicsCalibration'] = 1_000_000_000
+  sm.recv_frame = dict.fromkeys(sm.seen, 2)
+  sm.updated = {'modelV2': True}
+  client = Mock(frame_id=5, timestamp_sof=idx.timestampSof, timestamp_eof=idx.timestampEof)
+  frame = SimpleNamespace(frame_id=5)
+  client.recv.return_value = frame
+  core: Any = object.__new__(Compositor)
+  core.profile = request.param
+  core.wide = core.cabin = False
+  core.ui = SimpleNamespace(sm=sm, replay_sample=ClockSample(1_000_000_000, 9_050_000_000, 100, epoch=2, valid=True))
+  core.camera = Mock(client=client, frame=frame, stream_type=VisionStreamType.VISION_STREAM_NARROW_ROAD)
+  core.camera._ensure_connection.return_value = True
+  core.last_route_frame = None
+  core.render_receipts = {'modelV2': 1}
+  core.context, core.placement, core.adapter = Mock(), Mock(), Mock()
+  core.adapter.build.return_value = SimpleNamespace(onroad=OnroadState(True, False, 15., 80., SpeedLimitObservation()))
+  core.rl, core.content = Mock(), Mock()
+  core.gui = SimpleNamespace(_render_prepare=[Mock()], _frame=0)
+  core.scale = 1
+  core.view, core.converter, core.readback = Mock(), Mock(), Mock()
+  core.output = None
+  core.camera.render_camera_model_layer.side_effect = lambda *args, **kwargs: core._receive_primary()
+  def render_view(state):
+    # The real OnroadView enters its camera layer only with this availability flag.
+    if state.camera_available:
+      core._camera_layer(Mock(), state)
+  core.view.render.side_effect = render_view
+  return core
+
+
+@pytest.mark.parametrize('change', ('ahead', 'behind', 'missing_frame', 'timestamp', 'stale_state', 'unseen', 'invalid'))
+def test_known_camera_rejection_skips_preparation_and_preserves_updates(draw_core, change):
+  core = draw_core
+  if change in ('ahead', 'behind'):
+    core.camera.client.frame_id = 6 if change == 'ahead' else 4
+  elif change == 'missing_frame':
+    core.camera.frame = None
+  elif change == 'timestamp':
+    core.camera.client.timestamp_sof += 1
+  elif change == 'stale_state':
+    core.ui.sm.logMonoTime['narrowRoadEncodeIdx'] = 8_000_000_000
+  else:
+    getattr(core.ui.sm, 'seen' if change == 'unseen' else 'valid')['narrowRoadEncodeIdx'] = False
+  assert core.draw(2, 0., 100) is False
+  core.context.begin_frame.assert_not_called()
+  core.adapter.build.assert_not_called()
+  core.gui._render_prepare[0].assert_not_called()
+  core.view.render.assert_not_called()
+  core.readback.start.assert_not_called()
+  assert core.last_route_frame is None and core.gui._frame == 0
+  assert core.render_receipts == {'modelV2': 1} and core.ui.sm.updated['modelV2']
+
+
+@pytest.mark.parametrize('sample', (ClockSample(100, 100, 100, epoch=2),
+                                  ClockSample(100, 100, 100, epoch=3, valid=True)))
+def test_invalid_or_other_epoch_clock_never_advances_draw(draw_core, sample):
+  core = draw_core
+  core.ui.replay_sample = sample
+  assert core.draw(2, 0., 100) is False
+  core.context.begin_frame.assert_not_called()
+  core.adapter.build.assert_not_called()
+
+
+@pytest.mark.parametrize('replacement_id', (5, 6))
+def test_later_receive_keeps_final_identity_check(draw_core, replacement_id):
+  core = draw_core
+  def receive(**kwargs):
+    core.camera.client.frame_id = replacement_id
+    return SimpleNamespace(frame_id=replacement_id)
+  core.camera.client.recv.side_effect = receive
+  assert core.draw(2, 0., 100) is (replacement_id == 5)
+  core.camera.client.recv.assert_called_once_with(timeout_ms=0)
+  assert core.readback.start.call_count == int(replacement_id == 5)
+
+
+def test_optional_placeholder_still_allows_later_camera_recovery(draw_core):
+  from openpilot.cereal.visionipc import VisionStreamType
+  core = draw_core
+  core.camera.stream_type = VisionStreamType.VISION_STREAM_WIDE_ROAD
+  core.camera.client.frame_id = 4
+  def receive(**kwargs):
+    core.camera.client.frame_id = 5
+    return SimpleNamespace(frame_id=5)
+  core.camera.client.recv.side_effect = receive
+  assert core.draw(2, 0., 100)
+  core.camera.client.recv.assert_called_once_with(timeout_ms=0)
+  core.readback.start.assert_called_once()
+
+
+def test_eligible_retained_camera_draws_fresh_hud_and_route_time(draw_core):
+  core = draw_core
+  frame = core.camera.frame
+  core.camera.client.recv.return_value = None
+  assert core.draw(2, 0., 100)
+  core.ui.replay_sample = replace(core.ui.replay_sample, now_ns=1_050_000_000)
+  core.adapter.build.return_value.onroad = replace(core.adapter.build.return_value.onroad, speed_mps=20.)
+  assert core.draw(2, .05, 150)
+  core.ui.replay_sample = replace(core.ui.replay_sample, now_ns=1_100_000_000)
+  assert core.draw(2, .1, 200)  # Animation advances even without another source update.
+  assert core.camera.frame is frame
+  assert core.view.render.call_args.args[0].speed_mps == 20.
+  assert [call.args[0] for call in core.context.begin_frame.call_args_list] == [1., 1.05, 1.1]
+  assert core.gui._frame == 3
+
+
+def test_rejected_camera_can_match_next_update_without_losing_model_receipt(draw_core):
+  core = draw_core
+  core.camera.client.frame_id = 6
+  core.camera.frame = SimpleNamespace(frame_id=6)
+  assert not core.draw(2, 0., 100)
+  core.ui.sm['narrowRoadEncodeIdx'].frameId = 6
+  def consume_model(*args, **kwargs):
+    assert core.ui.sm.updated['modelV2']
+    core._receive_primary()
+  core.camera.client.recv.return_value = None
+  core.camera.render_camera_model_layer.side_effect = consume_model
+  assert core.draw(2, 0., 100)
+  assert core.render_receipts['modelV2'] == core.ui.sm.recv_frame['modelV2']
+  assert core.context.begin_frame.call_count == 1
+
+
+def test_optional_missing_camera_still_publishes_existing_placeholder(draw_core, monkeypatch):
+  from openpilot.cereal.visionipc import VisionStreamType
+  core = draw_core
+  label = Mock()
+  monkeypatch.setattr('openpilot.system.ui.widgets.label.gui_label', label)
+  core.camera.stream_type = VisionStreamType.VISION_STREAM_WIDE_ROAD
+  core.camera.client.frame_id = 4
+  core.camera.client.recv.return_value = None
+  assert core.draw(2, 0., 100)
+  label.assert_called_once()
+  core.readback.start.assert_called_once()
+
+
+@pytest.mark.parametrize('camera_view', (1, 3, 4))
+@pytest.mark.parametrize('draw_core', (Profile.COMPACT,), indirect=True)
+def test_compact_hidden_and_unrecorded_camera_keep_acquisition(draw_core, camera_view, monkeypatch):
+  core = draw_core
+  core.adapter.build.return_value.onroad = replace(core.adapter.build.return_value.onroad, appearance=replace(
+    core.adapter.build.return_value.onroad.appearance, camera_view=camera_view))
+  monkeypatch.setattr('openpilot.system.ui.widgets.label.gui_label', Mock())
+  assert core.draw(2, 0., 100)
+  core.camera.client.recv.assert_called_once_with(timeout_ms=0)
+  core.camera.render_camera_model_layer.assert_not_called()
+
+
 @pytest.mark.parametrize('profile', ('onroad', 'mici.onroad'))
 @pytest.mark.parametrize('replay', (False, True))
 def test_camera_layers_allow_recorded_preview_and_preserve_live_start_gate(monkeypatch, profile, replay):

tools/diagnostics/README.md

@@ -60,6 +60,52 @@ Focused tests (standard library compatible):
 python -m unittest tools.diagnostics.tests.test_performance
 ```
 
+## Replay compositor trace (20 FPS)
+
+`replay_render.py` is an opt-in wrapper for an existing owner-created replay worker. The normal worker never imports it.
+Keep the existing owner, private session, decoder, encoder and control procedure. In a separately authorized diagnostic
+owner's spawn callback, replace only `-m openpilot.starpilot.replay.render` with
+`-m tools.diagnostics.replay_render --output /absolute/unused-trace.json --`, preserving all existing worker arguments
+and environment. The wrapper refuses invalid private contexts and an existing output file; it does not connect to a
+device, create an owner, spoof parked authority, or start a session on its own. Do not run a private producer beside live replay.
+
+Records stay in a 30,000-event ring until worker shutdown, when one mode-0600 JSON report is written. Overwritten records
+are counted. Stop through the owning session and wait boundedly for owner idle; a successful stop response alone is insufficient.
+Normal SIGTERM cleanup and worker errors flush the report; SIGKILL cannot. There are no pixel copies or per-frame file/console writes.
+
+The report includes source revision/hashes, worker arguments, attempt IDs, clock and camera metadata before/after acquisition,
+early/late rejection reasons, inclusive and exclusive wall/thread-CPU timings, preparation, camera/model, HUD entry points,
+alerts/PiP, conversion, readback, encoding, muxing, publication, sleep and readback-pending observations. Deadline lateness
+mirrors the existing accepted-draw schedule; the initial draw has no previous deadline. Summary distributions cover only the
+retained event window; outcome counters cover the worker lifetime. GC intervals overlap other spans and have no exclusive/CPU measurement.
+The wrapper also accepts the unchanged worker before this patch: absent eligibility/preparation hooks are declared in
+`capabilities`, and its rejected draws remain `unclassified_rejection` rather than invented early/late classifications.
+
+`connect_import` includes connection checks, possible FD exchange/import and texture setup. `primary_acquisition` includes
+receive/synchronization and matching; the camera span also includes its later receive and submission. These are nested
+boundaries, not isolated receive-only or pure GPU timings. `submit_to_observed_ready_ns` includes GPU queueing and polling
+delay. HUD methods not individually wrapped remain in composition's exclusive remainder. Buffer metadata is an observation,
+not an owned pixel snapshot. The wrapper retains no readback pixels beyond the original delivery context.
+
+Use a fresh unchanged baseline and candidate at the same source bundle, scene, rate and profile; record GPU/browser/encoder
+conditions separately. Collect real CPU/load/temperature alongside this trace with the existing performance capture when
+authorized. Compare instrumentation on/off and reverse promising A/B order. Keep picture/decoded-content comparisons separate.
+This trace does not measure native upload/presentation, browser presentation, physical scanout or complete instrumentation overhead.
+
+Local harness checks: `./dev pytest -q tools/diagnostics/tests/test_replay_render.py`.
+
+Replay compatibility checks:
+
+```sh
+./dev pytest -q openpilot/starpilot/replay/tests tools/diagnostics/tests/test_replay_render.py \
+  openpilot/starpilot/ui/tests/test_replay_settings.py openpilot/starpilot/ui/tests/test_camera_display.py \
+  openpilot/starpilot/ui/tests/test_camera_resources.py openpilot/starpilot/ui/tests/test_onroad_alerts.py \
+  openpilot/starpilot/ui/tests/test_onroad_display_holds.py openpilot/selfdrive/ui/tests/test_camera_speed_gates.py \
+  openpilot/starpilot/galaxy/tests/test_replay_http.py
+node openpilot/starpilot/galaxy/tests/test_replay.mjs
+node openpilot/starpilot/galaxy/tests/test_replay_media.mjs
+```
+
 ## Offline longitudinal trace
 
 `longitudinal.py` reads one existing local rlog or qlog segment and prints one JSON object. It never opens a device, starts a publisher, or writes a log. Example: `./dev python -m tools.diagnostics.longitudinal /absolute/path/2-rlog.zst` (the `./dev` wrapper prints its own build progress before the tool's JSON). The report records input and source hashes, limits, message counts, mode transitions, rejected pairs and distributions.

tools/diagnostics/replay_render.py

@@ -0,0 +1,330 @@
+"""Opt-in bounded timings for an owner-created replay worker; no decoder or rate changes."""
+from collections import Counter, deque
+from contextlib import ExitStack, contextmanager
+import argparse
+import gc
+import hashlib
+import itertools
+import json
+import os
+from pathlib import Path
+import sys
+import threading
+import time
+from types import SimpleNamespace
+
+from tools.diagnostics.performance import Distribution, metadata
+
+MAX_EVENTS = 30000
+STAMPS = ('narrowRoadEncodeIdx', 'wideRoadEncodeIdx', 'cabinEncodeIdx',
+          'carState', 'controlsState', 'selfdriveState', 'modelV2', 'radarState')
+
+
+class Trace:
+  def __init__(self):
+    self.events = deque(maxlen=MAX_EVENTS)
+    self.sequence = itertools.count(1)
+    self.local = threading.local()
+    self.count = self.attempt = 0
+    self.outcomes = Counter()
+    self.core = None
+    self.pending_attempt = self.submitted_ns = None
+    self.last_draw = 0.
+    self.loop_now = None
+    self.await_loop_clock = False
+    self.connection_generation = 0
+    self.capabilities = {}
+    self.configuration = {}
+
+  def record(self, event):
+    self.events.append(event)
+    self.count += 1
+
+  @contextmanager
+  def span(self, stage):
+    stack = getattr(self.local, 'stack', [])
+    self.local.stack = stack
+    event = {'id': next(self.sequence), 'stage': stage, 'thread': threading.get_ident(),
+             'attempt': getattr(self.local, 'attempt', None), 'parent': stack[-1]['id'] if stack else None,
+             'start_ns': time.monotonic_ns()}
+    cpu = time.thread_time_ns()
+    entry = {'id': event['id'], 'child_wall': 0, 'child_cpu': 0}
+    stack.append(entry)
+    try:
+      yield event
+    except BaseException as error:
+      event['error_type'] = type(error).__name__
+      raise
+    finally:
+      event['wall_ns'] = time.monotonic_ns() - event['start_ns']
+      event['cpu_ns'] = time.thread_time_ns() - cpu
+      event['exclusive_wall_ns'] = event['wall_ns'] - entry['child_wall']
+      event['exclusive_cpu_ns'] = event['cpu_ns'] - entry['child_cpu']
+      stack.pop()
+      if stack:
+        stack[-1]['child_wall'] += event['wall_ns']
+        stack[-1]['child_cpu'] += event['cpu_ns']
+      self.record(event)
+
+  def snapshot(self):
+    began = time.monotonic_ns()
+    core = self.core
+    sample, sm, camera = core.ui.replay_sample, core.ui.sm, core.camera
+    client = camera.client
+    connected = client is not None and client.is_connected()
+    frame = camera.frame if connected else None
+    return {'epoch': getattr(sample, 'epoch', None), 'route_ns': getattr(sample, 'now_ns', None),
+            'boot_ns': getattr(sample, 'boot_ns', None), 'clock_valid': getattr(sample, 'valid', False),
+            'paused': getattr(sample, 'paused', None), 'stream': int(camera.stream_type),
+            'client': id(client) if client is not None else None, 'connected': connected,
+            'switching': camera._switching, 'connection_generation': self.connection_generation,
+            'pip_generation': core.pip_stream.generation,
+            'client_frame_id': getattr(client, 'frame_id', None),
+            'sof': getattr(client, 'timestamp_sof', None), 'eof': getattr(client, 'timestamp_eof', None),
+            'buffer': {name: getattr(frame, name, None) for name in
+                       ('frame_id', 'idx', 'fd', 'width', 'height', 'stride', 'uv_offset')},
+            'sources': {name: {'seen': sm.seen[name], 'valid': sm.valid[name], 'stamp': sm.logMonoTime[name],
+                               **({'frame_id': sm[name].frameId, 'sof': sm[name].timestampSof, 'eof': sm[name].timestampEof}
+                                  if name.endswith('EncodeIdx') and sm.seen[name] else {})}
+                        for name in STAMPS},
+            'observation_ns': time.monotonic_ns() - began}
+
+  def report(self):
+    stages = {}
+    for event in self.events:
+      values = stages.setdefault(event['stage'], {key: Distribution() for key in
+                                ('wall_ms', 'cpu_ms', 'exclusive_wall_ms', 'exclusive_cpu_ms')})
+      for key, distribution in values.items():
+        value = event.get(key.replace('_ms', '_ns'))
+        distribution.add(value / 1e6 if value is not None else None)
+    return {'schemaVersion': 1, 'kind': 'replay-render-trace', 'compositor_fps': 20,
+            'capabilities': self.capabilities, 'configuration': self.configuration,
+            'max_events': MAX_EVENTS, 'overwritten_events': self.count - len(self.events),
+            'lifetime_draw_outcomes': dict(self.outcomes),
+            'retained_window_stages': {stage: {key: value.summary() for key, value in values.items()}
+                                       for stage, values in stages.items()}, 'events': list(self.events)}
+
+
+@contextmanager
+def instrument(render, trace):
+  """Patch only this diagnostic worker; restore callbacks even when a frame fails."""
+  patched = set()
+  trace.capabilities = {name: hasattr(render.Compositor, method) for name, method in
+                        (('early_rejection_reason', '_draw_rejection'), ('preparation_hooks', '_prepare'))}
+  with ExitStack() as restore:
+    def replace(target, name, value):
+      previous = vars(target).get(name)
+      owned = name in vars(target)
+      setattr(target, name, value)
+      restore.callback(setattr, target, name, previous) if owned else restore.callback(delattr, target, name)
+
+    def wrap(target, name, stage, *, camera=False, after=None):
+      key = id(target), name
+      if key in patched:
+        return
+      patched.add(key)
+      original = getattr(target, name)
+      def measured(*args, **kwargs):
+        before = trace.snapshot() if camera else None
+        with trace.span(stage) as event:
+          result = original(*args, **kwargs)
+          if after is not None:
+            after(event, result, args)
+        if camera:
+          event.update(before=before, after=trace.snapshot())
+          if stage == 'stream_switch' or stage == 'connect_import' and not before['connected'] and event['after']['connected']:
+            trace.connection_generation += 1
+            event['after']['connection_generation'] = trace.connection_generation
+        return result
+      replace(target, name, measured)
+
+    def bind_view(event, result, args):
+      view = args[0].view
+      wrap(type(view), 'render', 'composition')
+      wrap(type(view.alert), 'render', 'alerts')
+      for name in ('current_speed', 'unified_speed', 'cem_gauge', 'steering_wheel', 'compact_hud', 'compact_sidebar', 'torque_bar'):
+        widget = getattr(view, name, None)
+        if widget is not None:
+          for method in vars(type(widget)):
+            if method == 'render' or method.startswith('render_'):
+              wrap(type(widget), method, 'hud')
+
+    def bind_core(event, result, args):
+      core = trace.core = args[0]
+      trace.configuration = {'profile': getattr(getattr(core, 'profile', None), 'value', None),
+                             'width': getattr(core, 'width', None), 'height': getattr(core, 'height', None),
+                             'gui_target_fps': getattr(core.gui, 'target_fps', None)}
+      from openpilot.starpilot.system.android_auto import gpu_nv12
+      camera_base = sys.modules[type(core.camera).__module__].CameraView
+      for method, stage in (('_render', 'camera'), ('_ensure_connection', 'connect_import'),
+                            ('_handle_switch', 'target_acquisition'), ('_complete_switch', 'stream_switch')):
+        wrap(camera_base, method, stage, camera=True)
+      for target, method, stage in (
+        (core.ui, 'update', 'replay_update'), (core.ui.sm, 'update', 'messages'),
+        (type(core.ui.replay_clock), 'sample', 'clock_sample'), (type(core.adapter), 'build', 'snapshot'),
+        (core.placement, 'maintain', 'placement'), (core.context, 'begin_frame', 'frame_clock'),
+        (gpu_nv12, 'compose_rgba', 'rgba_conversion'), (gpu_nv12.Nv12Converter, 'convert', 'nv12_conversion'),
+        (type(core.pip_stream), 'poll', 'pip_acquisition'), (type(core.pip), 'render', 'pip'),
+        (type(core.monitor), 'render', 'driver_monitor'),
+        (core.readback, 'finish', 'readback_map'), (core.readback, 'release', 'readback_release'),
+        (core.rl, 'begin_texture_mode', 'texture_begin'), (core.rl, 'end_texture_mode', 'texture_end'),
+      ):
+        wrap(target, method, stage)
+      def phase(name, callback, *args, **kwargs):
+        with trace.span('model' if name == 'model' else 'camera_phase'):
+          return original_phase(name, callback, *args, **kwargs)
+      original_phase = core.gui.measure_frame_phase
+      replace(core.gui, 'measure_frame_phase', phase)
+      def submitted(event, result, args):
+        trace.pending_attempt = trace.local.attempt
+        trace.submitted_ns = time.monotonic_ns()
+      def ready(event, result, args):
+        event['ready'] = result
+        if result and trace.submitted_ns is not None:
+          event['submit_to_observed_ready_ns'] = time.monotonic_ns() - trace.submitted_ns
+          trace.submitted_ns = None
+      def released(event, result, args):
+        trace.pending_attempt = trace.submitted_ns = None
+      wrap(core.readback, 'start', 'readback_submit', after=submitted)
+      wrap(core.readback, 'ready', 'readback_poll', after=ready)
+      # Record cancellation on seek too, without retaining the borrowed pixel memory.
+      original_release = core.readback.release
+      def release():
+        try:
+          return original_release()
+        finally:
+          released(None, None, None)
+      replace(core.readback, 'release', release)
+
+    original_draw, original_finish = render.Compositor.draw, render.Compositor.finish
+    def draw(core, *args):
+      trace.attempt += 1
+      trace.local.attempt = trace.attempt
+      trace.local.rejection = None
+      before = trace.snapshot()
+      try:
+        with trace.span('draw') as event:
+          accepted = original_draw(core, *args)
+          reason = 'accepted' if accepted else trace.local.rejection or (
+            'late_camera_not_ready' if trace.capabilities['early_rejection_reason'] else 'unclassified_rejection')
+          trace.outcomes[reason] += 1
+          event.update(accepted=accepted, reason=reason, captured_ns=args[2], position=args[1], before=before)
+          if accepted and trace.loop_now is not None:
+            trace.last_draw = max(trace.last_draw + render.FRAME_INTERVAL, trace.loop_now - render.FRAME_INTERVAL)
+        event['after'] = trace.snapshot()
+        return accepted
+      except BaseException:
+        trace.outcomes['error'] += 1
+        raise
+      finally:
+        trace.local.attempt = None
+
+    @contextmanager
+    def finish(core):
+      trace.local.attempt = trace.pending_attempt
+      try:
+        with trace.span('delivery_scope') as event:
+          with original_finish(core) as value:
+            event['delivered'] = value is not None
+            yield value
+      finally:
+        trace.local.attempt = None
+
+    def eligibility(event, result, args):
+      trace.local.rejection = event['reason'] = result
+
+    def updated(event, result, args):
+      trace.await_loop_clock = True
+      event['readback_pending'] = trace.core.readback.pending
+
+    def monotonic():
+      now = time.monotonic()
+      if trace.await_loop_clock:
+        trace.await_loop_clock, trace.loop_now = False, now
+        with trace.span('scheduler') as event:
+          event.update(readback_pending=trace.core.readback.pending,
+                       deadline_lateness_ns=round(max(0., now - trace.last_draw - render.FRAME_INTERVAL) * 1e9)
+                       if trace.last_draw else None)
+      return now
+
+    def sleep(seconds):
+      with trace.span('scheduler_sleep') as event:
+        event['requested_ns'] = round(seconds * 1e9)
+        time.sleep(seconds)
+
+    pending_gc = {}
+    def gc_event(phase, info):
+      generation = info['generation']
+      if phase == 'start':
+        pending_gc[generation] = time.monotonic_ns()
+      elif generation in pending_gc:
+        start = pending_gc.pop(generation)
+        trace.record({'id': next(trace.sequence), 'stage': 'gc', 'start_ns': start,
+                      'wall_ns': time.monotonic_ns() - start, 'generation': generation,
+                      'collected': info['collected'], 'overlaps_other_stages': True})
+
+    wrap(render.Compositor, '_new_view', 'view_init', after=bind_view)
+    wrap(render.Compositor, '__init__', 'compositor_init', after=bind_core)
+    wrap(render.Compositor, 'update', 'update', after=updated)
+    wrap(render.Compositor, '_receive_primary', 'primary_acquisition', camera=True)
+    if trace.capabilities['early_rejection_reason']:
+      wrap(render.Compositor, '_draw_rejection', 'eligibility', after=eligibility)
+    if trace.capabilities['preparation_hooks']:
+      wrap(render.Compositor, '_prepare', 'preparation_hooks')
+    wrap(render.Compositor, 'reset_epoch', 'epoch_reset')
+    replace(render.Compositor, 'draw', draw)
+    replace(render.Compositor, 'finish', finish)
+    replace(render, 'time', SimpleNamespace(monotonic=monotonic, monotonic_ns=time.monotonic_ns, sleep=sleep))
+    from openpilot.starpilot.system.android_auto.hw_encoder import HardwareH264Encoder
+    for target, method, stage in ((HardwareH264Encoder, 'encode_nv12', 'encode'),
+                                  (render.MediaMux, 'frame', 'mux'), (render.MediaStore, 'publish', 'web_publish'),
+                                  (render.ReplayFramePublisher, 'publish', 'native_publish')):
+      wrap(target, method, stage)
+    gc.callbacks.append(gc_event)
+    restore.callback(gc.callbacks.remove, gc_event)
+    yield
+
+
+def main(argv=None):
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--output', type=Path, required=True)
+  parser.add_argument('worker_args', nargs=argparse.REMAINDER)
+  args = parser.parse_args(argv)
+  if args.worker_args[:1] != ['--']:
+    parser.error('pass the existing owner-created worker arguments after --')
+  from openpilot.starpilot.replay import render
+  if render.FRAME_INTERVAL != .05:
+    parser.error('this diagnostic is restricted to the existing 20 FPS compositor')
+  # Reject a non-owner context before installing any instrumentation or starting the engine.
+  worker = argparse.ArgumentParser(add_help=False)
+  worker.add_argument('--session-dir', type=Path, required=True)
+  context, _ = worker.parse_known_args(args.worker_args[1:])
+  render.validate_context(context.session_dir)
+  trace = Trace()
+  root = Path(render.__file__).resolve().parents[3]
+  source = metadata(root)
+  for name in ('openpilot/cereal/messaging/__init__.py', 'openpilot/selfdrive/ui/ui_state.py',
+               'openpilot/selfdrive/ui/onroad/cameraview.py', 'openpilot/selfdrive/ui/mici/onroad/cameraview.py',
+               'openpilot/tools/replay/display_clock.py', 'openpilot/starpilot/system/android_auto/headless_egl.py',
+               'openpilot/starpilot/system/android_auto/hw_encoder.py'):
+    source['sourceSha256'][name] = hashlib.sha256((root / name).read_bytes()).hexdigest()
+  source['replayHashes'] = {name: hashlib.sha256(path.read_bytes()).hexdigest() for name, path in
+                          (('render', Path(render.__file__)), ('owner', Path(render.__file__).with_name('owner.py')),
+                           ('media', Path(render.__file__).with_name('media.py')), ('harness', Path(__file__)))}
+  fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
+  previous = sys.argv
+  try:
+    with os.fdopen(fd, 'w') as output:
+      try:
+        sys.argv = ['replay-render-diagnostic', *args.worker_args[1:]]
+        with instrument(render, trace):
+          render.main()
+      finally:
+        json.dump({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:]}, output, allow_nan=False)
+        output.write('\n')
+  finally:
+    sys.argv = previous
+
+
+if __name__ == '__main__':
+  main()

tools/diagnostics/tests/test_replay_render.py

@@ -0,0 +1,236 @@
+from contextlib import contextmanager
+from types import SimpleNamespace as NS
+from unittest.mock import Mock
+import gc
+import json
+import stat
+import sys
+
+import pytest
+
+from tools.diagnostics import replay_render as diagnostic
+
+
+def test_nested_timings_are_exclusive_and_failures_propagate(monkeypatch):
+  trace = diagnostic.Trace()
+  ticks = iter((100, 120, 150, 200))
+  monkeypatch.setattr(diagnostic.time, 'monotonic_ns', lambda: next(ticks))
+  monkeypatch.setattr(diagnostic.time, 'thread_time_ns', lambda: 0)
+  with pytest.raises(ValueError, match='original error'), trace.span('outer'):
+    with trace.span('inner'):
+      raise ValueError('original error')
+  inner, outer = trace.events
+  assert inner['wall_ns'] == inner['exclusive_wall_ns'] == 30
+  assert outer['wall_ns'] == 100 and outer['exclusive_wall_ns'] == 70
+  assert inner['parent'] == outer['id']
+  assert inner['error_type'] == outer['error_type'] == 'ValueError'
+
+
+def test_records_are_bounded_and_missing_cpu_is_not_zero(monkeypatch):
+  monkeypatch.setattr(diagnostic, 'MAX_EVENTS', 2)
+  trace = diagnostic.Trace()
+  for number in range(3):
+    trace.record({'stage': 'gc', 'wall_ns': number})
+  report = trace.report()
+  assert len(report['events']) == 2 and report['overwritten_events'] == 1
+  assert report['retained_window_stages']['gc']['cpu_ms']['p50'] is None
+  json.dumps(report, allow_nan=False)
+
+
+class Hooks:
+  def update(self, *args): pass
+  def sample(self): return NS(valid=True, epoch=2)
+  def build(self, *args): pass
+  def render(self, *args): pass
+  def maintain(self, *args): pass
+  def begin_frame(self, *args): pass
+  def begin_texture_mode(self, *args): pass
+  def end_texture_mode(self, *args): pass
+  def measure_frame_phase(self, name, callback, *args, **kwargs): return callback(*args, **kwargs)
+
+
+class CameraView:
+  def __init__(self):
+    self.client = NS(is_connected=lambda: True, frame_id=5, timestamp_sof=10, timestamp_eof=20)
+    self.frame = NS(frame_id=5, idx=0, fd=10, width=4, height=2, stride=4, uv_offset=8)
+    self.stream_type, self._switching = 0, False
+  def _render(self): self._ensure_connection()
+  def _ensure_connection(self): return True
+  def _handle_switch(self): pass
+  def _complete_switch(self): pass
+
+
+class Pip(Hooks):
+  generation = 1
+  def poll(self, *args): pass
+
+
+class Readback:
+  def __init__(self):
+    self.pending = False
+    self.pixels = memoryview(b'pixels')
+  def start(self): self.pending = True
+  def ready(self): return self.pending
+  def finish(self): return self.pixels
+  def release(self): self.pending = False
+
+
+class View(Hooks):
+  def __init__(self): self.alert = Hooks()
+
+
+class Compositor:
+  def __init__(self):
+    self.ui = Hooks()
+    self.ui.replay_clock = Hooks()
+    self.ui.replay_sample = NS(epoch=2, now_ns=100, boot_ns=100, valid=True, paused=False)
+    self.ui.sm = Hooks()
+    self.ui.sm.seen = self.ui.sm.valid = dict.fromkeys(diagnostic.STAMPS, False)
+    self.ui.sm.logMonoTime = dict.fromkeys(diagnostic.STAMPS, 0)
+    self.camera, self.pip_stream, self.pip, self.monitor = CameraView(), Pip(), Pip(), Hooks()
+    self.adapter, self.placement, self.context, self.gui, self.rl = (Hooks() for _ in range(5))
+    self.readback = Readback()
+    self._new_view()
+    self.rejection = None
+  def _new_view(self): self.view = View()
+  def update(self): return self.ui.replay_sample
+  def _receive_primary(self): self.camera._render()
+  def _draw_rejection(self, epoch): return self.rejection
+  def _prepare(self): pass
+  def reset_epoch(self, epoch): self.readback.release()
+  def draw(self, *args):
+    if self._draw_rejection(args[0]):
+      return False
+    self._receive_primary()
+    self._prepare()
+    self.gui.measure_frame_phase('model', lambda: None)
+    self.readback.start()
+    return True
+  @contextmanager
+  def finish(self):
+    if not self.readback.ready():
+      yield None
+      return
+    try:
+      yield self.readback.finish(), (2, 0., 100)
+    finally:
+      self.readback.release()
+
+
+class Output:
+  def frame(self): pass
+  def publish(self): pass
+
+
+def test_worker_hooks_preserve_delivery_lifetime_and_restore_after_failure():
+  trace = diagnostic.Trace()
+  module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output,
+              FRAME_INTERVAL=.05, time=diagnostic.time)
+  original_draw, original_clock, callbacks = Compositor.draw, module.time, list(gc.callbacks)
+  with pytest.raises(RuntimeError, match='encoder failed'), diagnostic.instrument(module, trace):
+    core = Compositor()
+    core.update()
+    module.time.monotonic()
+    assert core.draw(2, 0., 100)
+    with core.finish() as value:
+      assert value[0].tobytes() == b'pixels' and core.readback.pending
+      raise RuntimeError('encoder failed')
+  assert not core.readback.pending
+  assert trace.pending_attempt is trace.submitted_ns is None
+  assert Compositor.draw is original_draw and module.time is original_clock
+  assert gc.callbacks == callbacks
+  delivery = next(event for event in trace.events if event['stage'] == 'delivery_scope')
+  assert delivery['attempt'] == 1 and delivery['error_type'] == 'RuntimeError'
+  ready = next(event for event in trace.events if event['stage'] == 'readback_poll')
+  assert ready['submit_to_observed_ready_ns'] >= 0
+  assert trace.outcomes == {'accepted': 1}
+  json.dumps(trace.report(), allow_nan=False)
+
+
+def test_early_reason_is_recorded_without_preparation():
+  trace = diagnostic.Trace()
+  module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output,
+              FRAME_INTERVAL=.05, time=diagnostic.time)
+  with diagnostic.instrument(module, trace):
+    core = Compositor()
+    core.rejection = 'camera_not_ready'
+    assert not core.draw(2, 0., 100)
+  assert trace.outcomes == {'camera_not_ready': 1}
+  assert not any(event['stage'] == 'preparation_hooks' for event in trace.events)
+
+
+def test_unchanged_worker_does_not_invent_rejection_reasons(monkeypatch):
+  monkeypatch.delattr(Compositor, '_draw_rejection')
+  monkeypatch.delattr(Compositor, '_prepare')
+  monkeypatch.setattr(Compositor, 'draw', lambda *args: False)
+  trace = diagnostic.Trace()
+  module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output,
+              FRAME_INTERVAL=.05, time=diagnostic.time)
+  with diagnostic.instrument(module, trace):
+    assert not Compositor().draw(2, 0., 100)
+  assert trace.capabilities == {'early_rejection_reason': False, 'preparation_hooks': False}
+  assert trace.outcomes == {'unclassified_rejection': 1}
+
+
+def test_scheduler_reports_pending_inhibition_and_seek_cancels_readback(monkeypatch):
+  now = [10.]
+  monkeypatch.setattr(diagnostic.time, 'monotonic', lambda: now[0])
+  trace = diagnostic.Trace()
+  module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output,
+              FRAME_INTERVAL=.05, time=diagnostic.time)
+  with diagnostic.instrument(module, trace):
+    core = Compositor()
+    core.update()
+    module.time.monotonic()
+    assert core.draw(2, 0., 100)
+    now[0] += .02
+    core.update()
+    module.time.monotonic()
+    scheduled = [event for event in trace.events if event['stage'] == 'scheduler']
+    assert scheduled[0]['deadline_lateness_ns'] is None
+    assert scheduled[1]['deadline_lateness_ns'] == 20_000_000 and scheduled[1]['readback_pending']
+    core.reset_epoch(3)
+    assert trace.pending_attempt is trace.submitted_ns is None
+    with core.finish() as value:
+      assert value is None
+
+
+def test_snapshot_does_not_dereference_disconnected_buffer():
+  class InvalidBuffer:
+    def __getattribute__(self, name):
+      raise AssertionError('disconnected buffer was read')
+  trace = diagnostic.Trace()
+  core = trace.core = Compositor()
+  core.camera.frame = InvalidBuffer()
+  core.camera.client.is_connected = lambda: False
+  assert trace.snapshot()['buffer']['frame_id'] is None
+
+
+def test_cli_refuses_invalid_context_and_existing_output(tmp_path, monkeypatch):
+  from openpilot.starpilot.replay import render
+  output = tmp_path / 'trace.json'
+  launch = Mock()
+  monkeypatch.setattr(render, 'main', launch)
+  with pytest.raises((OSError, ValueError)):
+    diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path / 'missing')])
+  assert not output.exists()
+  launch.assert_not_called()
+  monkeypatch.setattr(render, 'validate_context', lambda _: None)
+  output.write_text('keep')
+  with pytest.raises(FileExistsError):
+    diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)])
+  assert output.read_text() == 'keep'
+  launch.assert_not_called()
+
+
+def test_cli_flushes_partial_report_on_worker_failure(tmp_path, monkeypatch):
+  from openpilot.starpilot.replay import render
+  output = tmp_path / 'trace.json'
+  monkeypatch.setattr(render, 'validate_context', lambda _: None)
+  monkeypatch.setattr(render, 'main', Mock(side_effect=SystemExit(1)))
+  previous = sys.argv
+  with pytest.raises(SystemExit):
+    diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)])
+  assert sys.argv is previous
+  assert stat.S_IMODE(output.stat().st_mode) == 0o600
+  assert json.loads(output.read_text())['events'] == []