Replay diagnostic checkpoint

tools/diagnostics/README.md

@@ -69,9 +69,13 @@ owner's spawn callback, replace only `-m openpilot.starpilot.replay.render` with
 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.
+Records stay in a 30,000-event ring until one mode-0600 JSON report is written. Overwritten records are counted.
+After pausing through the owner and confirming readiness, send SIGUSR1 to this diagnostic worker PID to request a checkpoint.
+It writes at the next valid paused loop boundary with no borrowed readback pending; wait for a complete JSON report before stopping.
+The report declares `report_boundary=paused_checkpoint` and excludes subsequent teardown. Without a checkpoint it attempts a
+shutdown report, but the owner's cleanup deadline can kill it before that completes. Stop through the owning session and wait
+boundedly for owner idle; a successful stop response alone is insufficient. SIGKILL cannot flush. 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,

tools/diagnostics/replay_render.py

@@ -8,6 +8,7 @@ import itertools
 import json
 import os
 from pathlib import Path
+import signal
 import sys
 import threading
 import time
@@ -35,6 +36,7 @@ class Trace:
     self.connection_generation = 0
     self.capabilities = {}
     self.configuration = {}
+    self.flush_requested = False
 
   def record(self, event):
     self.events.append(event)
@@ -90,8 +92,9 @@ class Trace:
             'observation_ns': time.monotonic_ns() - began}
 
   def report(self):
+    events, count = list(self.events), self.count
     stages = {}
-    for event in self.events:
+    for event in 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():
@@ -99,14 +102,14 @@ class Trace:
         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),
+            'max_events': MAX_EVENTS, 'overwritten_events': count - len(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)}
+                                       for stage, values in stages.items()}, 'events': events}
 
 
 @contextmanager
-def instrument(render, trace):
+def instrument(render, trace, *, flush=None):
   """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
@@ -251,6 +254,11 @@ def instrument(render, trace):
       with trace.span('scheduler_sleep') as event:
         event['requested_ns'] = round(seconds * 1e9)
         time.sleep(seconds)
+      sample = trace.core.ui.replay_sample if trace.core is not None else None
+      if (flush is not None and trace.flush_requested and sample is not None and sample.valid and sample.paused and
+          not trace.core.readback.pending):
+        trace.flush_requested = False
+        flush()
 
     pending_gc = {}
     def gc_event(phase, info):
@@ -315,13 +323,23 @@ def main(argv=None):
   previous = sys.argv
   try:
     with os.fdopen(fd, 'w') as output:
+      saved = False
+      def save(boundary):
+        nonlocal saved
+        if not saved:
+          output.write(json.dumps({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:],
+                                   'report_boundary': boundary}, separators=(',', ':'), allow_nan=False))
+          output.write('\n')
+          output.flush()
+          saved = True
+      previous_signal = signal.signal(signal.SIGUSR1, lambda *_: setattr(trace, 'flush_requested', True))
       try:
         sys.argv = ['replay-render-diagnostic', *args.worker_args[1:]]
-        with instrument(render, trace):
+        with instrument(render, trace, flush=lambda: save('paused_checkpoint')):
           render.main()
       finally:
-        json.dump({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:]}, output, allow_nan=False)
-        output.write('\n')
+        signal.signal(signal.SIGUSR1, previous_signal)
+        save('worker_exit')
   finally:
     sys.argv = previous
 

tools/diagnostics/tests/test_replay_render.py

@@ -3,6 +3,7 @@ from types import SimpleNamespace as NS
 from unittest.mock import Mock
 import gc
 import json
+import signal
 import stat
 import sys
 
@@ -206,6 +207,43 @@ def test_snapshot_does_not_dereference_disconnected_buffer():
   assert trace.snapshot()['buffer']['frame_id'] is None
 
 
+def test_requested_checkpoint_waits_for_a_valid_paused_frame_without_borrowed_pixels():
+  trace = diagnostic.Trace()
+  reports = []
+  module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output,
+              FRAME_INTERVAL=.05, time=diagnostic.time)
+  with diagnostic.instrument(module, trace, flush=lambda: reports.append(trace.report())):
+    core = Compositor()
+    trace.flush_requested = True
+    module.time.sleep(0)
+    assert reports == []
+    core.ui.replay_sample.paused = True
+    core.readback.pending = True
+    module.time.sleep(0)
+    assert reports == []
+    core.readback.pending = False
+    core.ui.replay_sample.valid = False
+    module.time.sleep(0)
+    assert reports == []
+    core.ui.replay_sample.valid = True
+    module.time.sleep(0)
+    module.time.sleep(0)
+    assert len(reports) == 1 and reports[0]['events']
+
+
+def test_report_keeps_a_stable_window_when_gc_records_during_summary(monkeypatch):
+  trace = diagnostic.Trace()
+  trace.record({'stage': 'draw', 'wall_ns': 100})
+  original = diagnostic.Distribution.add
+  def add(distribution, *args):
+    trace.record({'stage': 'gc', 'wall_ns': 1})
+    return original(distribution, *args)
+  monkeypatch.setattr(diagnostic.Distribution, 'add', add)
+  report = trace.report()
+  assert len(report['events']) == 1 and report['events'][0]['stage'] == 'draw'
+  assert report['overwritten_events'] == 0
+
+
 def test_cli_refuses_invalid_context_and_existing_output(tmp_path, monkeypatch):
   from openpilot.starpilot.replay import render
   output = tmp_path / 'trace.json'
@@ -233,4 +271,28 @@ def test_cli_flushes_partial_report_on_worker_failure(tmp_path, monkeypatch):
     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'] == []
+  report = json.loads(output.read_text())
+  assert not any(event['stage'] == 'draw' for event in report['events'])
+  assert report['report_boundary'] == 'worker_exit'
+
+
+def test_cli_checkpoint_is_complete_before_teardown_and_is_not_appended_on_exit(tmp_path, monkeypatch):
+  from openpilot.starpilot.replay import render
+  output = tmp_path / 'trace.json'
+  monkeypatch.setattr(render, 'validate_context', lambda _: None)
+  monkeypatch.setattr(render, 'Compositor', Compositor)
+  saved = []
+  def worker():
+    core = render.Compositor()
+    core.ui.replay_sample.paused = True
+    signal.raise_signal(signal.SIGUSR1)
+    render.time.sleep(0)
+    saved.append(output.read_text())
+    assert json.loads(saved[0])['report_boundary'] == 'paused_checkpoint'
+    raise SystemExit(1)
+  monkeypatch.setattr(render, 'main', worker)
+  previous = signal.getsignal(signal.SIGUSR1)
+  with pytest.raises(SystemExit):
+    diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)])
+  assert signal.getsignal(signal.SIGUSR1) is previous
+  assert output.read_text() == saved[0]