Connect replay demo engagement and signals to native UI and music

roadscore/prototype/app.py

@@ -11,7 +11,7 @@ from phrase import pulse,cadence_runway,mix_cadence
 from driving_music import DrivingDSP
 from event_music import EventDSP
 from input_clock import InputClock
-from demo_engagement import DemoEngagement,presentation_active,annotate as annotate_demo_engagement
+from demo_engagement import DemoEngagement,presentation_active,presentation_signal,controls_snapshot,annotate as annotate_demo_engagement
 from presentation_policy import effective_config,selected as presentation_policy_selected
 from engagement_presentation import EngagementPresentation,PresentationConfig,engagement_active
 from motion_presentation import MotionPresentation
@@ -209,9 +209,10 @@ def callback(out,n,ti,status):
  rendered=render_audio(render_mode,chunk,dsp,amount,ending,gestures,mix_cadence,
   (ending_audio,frames-n,ending_start,rate) if musical_mode and ending_start is not None else None)
  active,fresh=engagement_active(engagement[1],engagement[0],engagement[2],engagement[4],engagement[3],callback_wall)
- recorded_active=active;demo_mode=demo_engagement.mode
+ recorded_active=active;demo_mode,demo_signal_mode=demo_engagement.selection
  active=presentation_active(demo_mode,recorded_active)
  signal_on,signal_fresh=engagement_active(signal_state[1],signal_state[0],signal_state[2],signal_state[4],signal_state[3],callback_wall)
+ recorded_signal=signal_on;signal_on=presentation_signal(demo_signal_mode,recorded_signal)
  if rhythm_timeline is not None:
   audible_grid=rhythm_timeline.at(frames-n)
   if shaker is not None:shaker.set_grid(audible_grid,frames-n)
@@ -242,6 +243,7 @@ def callback(out,n,ti,status):
  if presentation is not None:rendered=presentation.process(rendered,active,presentation_config)
  out[:]=0 if a.mute else rendered
  cue={**presentation_narrative,**{key:event_state[key] for key in PRESENTATION_FIELDS if key in event_state}}
+ cue['replay_demo']=controls_snapshot(demo_mode,demo_signal_mode)
  if presentation is not None:cue.update(annotate_demo_engagement(presentation.snapshot(presentation_config,active,fresh),demo_mode,recorded_active))
  if shaker is not None:cue['signal_shaker']=shaker.snapshot()
  if apex is not None:cue['core_apex']=apex.snapshot()
@@ -250,7 +252,7 @@ def callback(out,n,ti,status):
  if curve_reaction is not None:cue.update(curve_reaction.snapshot())
  rendered_presentation=dict(sequence=frames-n,callback_wall=callback_wall,
                             dac_wall=callback_wall+float(ti.outputBufferDacTime-ti.currentTime),cues=cue)
- try:capture.put_nowait((rendered,chunk,{'motion_presentation':motion.snapshot() if motion is not None else None,'signal_shaker_enabled':shaker_enabled,'signal_on':signal_on,'signal_fresh':signal_fresh,'engagement_presentation_enabled':presentation_config.enabled,'engagement_active':active,'engagement_recorded_active':recorded_active,'demo_engagement_mode':demo_mode,'engagement_fresh':fresh,'audio_s':(frames-n)/rate,'callback_wall':callback_wall,'command_received_wall':command_wall,'replay_origin_wall':replay_origin_wall,'dac_delay':float(ti.outputBufferDacTime-ti.currentTime),'route_t':source_time,'amount':amount,'phase':event_state['phase'],'strength':event_state.get('strength',0),'predicted_peak':event_state.get('predicted_peak'),'activation':event_state.get('activation'),'kind':event_state.get('kind','curve'),'cadence_entry_audio_s':None if ending_start is None else ending_start/rate,'runway_active':ending_start is not None and frames-n<ending_start,'muted':a.mute,'portaudio_status':str(status),'callback_processing_seconds':time.monotonic()-callback_wall}))
+ try:capture.put_nowait((rendered,chunk,{'motion_presentation':motion.snapshot() if motion is not None else None,'signal_shaker_enabled':shaker_enabled,'signal_on':signal_on,'signal_fresh':signal_fresh,'engagement_presentation_enabled':presentation_config.enabled,'engagement_active':active,'engagement_recorded_active':recorded_active,'demo_engagement_mode':demo_mode,'demo_signal_mode':demo_signal_mode,'signal_recorded_on':recorded_signal,'engagement_fresh':fresh,'audio_s':(frames-n)/rate,'callback_wall':callback_wall,'command_received_wall':command_wall,'replay_origin_wall':replay_origin_wall,'dac_delay':float(ti.outputBufferDacTime-ti.currentTime),'route_t':source_time,'amount':amount,'phase':event_state['phase'],'strength':event_state.get('strength',0),'predicted_peak':event_state.get('predicted_peak'),'activation':event_state.get('activation'),'kind':event_state.get('kind','curve'),'cadence_entry_audio_s':None if ending_start is None else ending_start/rate,'runway_active':ending_start is not None and frames-n<ending_start,'muted':a.mute,'portaudio_status':str(status),'callback_processing_seconds':time.monotonic()-callback_wall}))
  except queue.Full:underflows+=1
 # Optional prewarmed continuation gives ~52 seconds before replay starts.
 warm=root/'generated/job_-1.wav'
@@ -442,6 +444,8 @@ try:
     snapshot['presentation_session_id']=presentation_session_id
     snapshot['input_mode']=a.input
     snapshot['demo_engagement_mode']=demo_engagement.mode
+    snapshot['demo_signal_mode']=demo_engagement.signal_mode
+    snapshot['replay_demo']=controls_snapshot(*demo_engagement.selection)
     presentation_narrative={key:snapshot[key] for key in ('section','next_section','gesture_active','gesture_queued','turn_signal_music') if key in snapshot}
     trace.write(json.dumps(snapshot)+'\n');f=run/'status.tmp';f.write_text(json.dumps(presentation_delay.apply(snapshot,rendered_presentation)));f.replace(run/'status.json')
    if time.monotonic()-last_progress>15:raise RuntimeError('Replay model input stalled')

roadscore/prototype/cue_timing.py

@@ -3,7 +3,7 @@ import math
 
 FIELDS = ('phase','kind','amount','activation','strength','section','next_section',
           'gesture_active','gesture_queued','turn_signal_music','lead','predicted_peak','scheduled',
-          'signal_shaker','core_apex','alert_accent','engagement_presentation','motion_presentation','curve_reaction')
+          'signal_shaker','core_apex','alert_accent','engagement_presentation','motion_presentation','curve_reaction','replay_demo')
 REFERENCE = 'portaudio-dac-plus-residual-v1'
 
 

roadscore/prototype/demo_engagement.py

@@ -5,34 +5,50 @@ import time
 from pathlib import Path
 
 MODES = ('recorded', 'engaged', 'disengaged')
+SIGNAL_MODES = ('recorded', 'left', 'right', 'off')
 
 
 class DemoEngagement:
  def __init__(self, run, session_id, input_mode, *, clock=time.monotonic):
   self.path=Path(run)/'demo_engagement.json';self.session_id=session_id
-  self.input_mode=input_mode;self.clock=clock;self.started=clock();self.mode='recorded'
+  self.input_mode=input_mode;self.clock=clock;self.started=clock();self.selection=('recorded','recorded')
+
+ @property
+ def mode(self):return self.selection[0]
+
+ @property
+ def signal_mode(self):return self.selection[1]
 
  def poll(self):
-  mode='recorded'
+  selection=('recorded','recorded')
   if self.input_mode!='replay':
-   self.mode=mode
-   return mode
+   self.selection=selection
+   return self.mode
   try:
    value=json.loads(self.path.read_text())
    stamp=value.get('created_wall')
    if (value.get('version')==1 and value.get('session_id')==self.session_id
        and value.get('mode') in MODES and type(stamp) in (int,float)
+       and value.get('signal_mode','recorded') in SIGNAL_MODES
        and math.isfinite(stamp) and self.started<=stamp<=self.clock()):
-    mode=value['mode']
+    selection=(value['mode'],value.get('signal_mode','recorded'))
   except (OSError,ValueError,AttributeError):pass
-  self.mode=mode
-  return mode
+  self.selection=selection
+  return self.mode
 
 
 def presentation_active(mode, recorded_active):
  return recorded_active if mode=='recorded' else mode=='engaged'
 
 
+def presentation_signal(mode, recorded_signal):
+ return recorded_signal if mode=='recorded' else mode in ('left','right')
+
+
+def controls_snapshot(mode, signal_mode):
+ return {'mode':mode,'signal_mode':signal_mode,'simulated':mode!='recorded' or signal_mode!='recorded'}
+
+
 def annotate(snapshot, mode, recorded_active):
  result=dict(snapshot)
  result['engagement_presentation']={**result.get('engagement_presentation',{}),

roadscore/prototype/normal_onroad.py

@@ -71,6 +71,7 @@ if a.replay:initial_display.update(style='Stored score',section='ARCHIVED SCORE'
 (out/'roadscore_status.json').write_text(json.dumps(initial_display))
 env['ROADSCORE_OVERLAY']='1' if settings.overlay else '0'
 env['ROADSCORE_CLEAN_DEMO_UI']='1'
+env['ROADSCORE_REPLAY_UI_CONTROLS']='1' if native and not a.replay else '0'
 env['ROADSCORE_FORCE_MUTE']='0' if a.audible else '1'
 env['ROADSCORE_ORIGIN_FILE']=str(out/'replay_origin.json')
 env['ROADSCORE_UI_AUDIT']=str(out/'ui_audit.jsonl');env['PWD']=str(rt)

roadscore/prototype/normal_ui_audit.py

@@ -4,11 +4,14 @@ from pathlib import Path
 from openpilot.selfdrive.ui.ui_state import UIState, device
 from preparing_awake import PreparationWake
 from replay_display_hold import ReplayDisplayHold
+from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay
+replay_controls=ReplayUIControls(os.environ['ROADSCORE_STATUS_FILE'],enabled=isolated_replay(os.environ))
 display_hold=ReplayDisplayHold(os.environ.get("ROADSCORE_AUDIO_DRAIN_FILE"), enabled=os.environ.get("OPENPILOT_PREFIX")=="roadscore_replay")
 original_state=UIState._update_state
 def replay_state(self,*args,**kwargs):
  result=original_state(self,*args,**kwargs)
  self.started=display_hold.apply(self.started)
+ if isinstance(self.sm,ReplayStateView):self.sm.apply_native_mode(self)
  return result
 UIState._update_state=replay_state
 preparation_wake=PreparationWake(os.environ.get("ROADSCORE_STATUS_FILE"),Path("/TICI").exists())
@@ -42,10 +45,12 @@ def textures(self,*args,**kw):
  return original_textures(self,*args,**kw)
 def update(self,*args,**kw):
  global last
+ if replay_controls.enabled and not isinstance(self.sm,ReplayStateView):
+  self.sm=ReplayStateView(self.sm,replay_controls)
  preparation_wake.update(self,device)
  result=original_update(self,*args,**kw);now=time.monotonic()
  if now-last>=1:
-  out.write(json.dumps({'wall':now,'started':bool(self.started),'speed':float(self.sm['carState'].vEgo),'model_mono_ns':self.sm.logMonoTime['modelV2'],'model_points':len(self.sm['modelV2'].position.x),**counts})+'\n');last=now
+  out.write(json.dumps({'wall':now,'started':bool(self.started),'speed':float(self.sm['carState'].vEgo),'model_mono_ns':self.sm.logMonoTime['modelV2'],'model_points':len(self.sm['modelV2'].position.x),'replay_demo':self.sm.snapshot() if isinstance(self.sm,ReplayStateView) else None,'engaged':bool(self.engaged),'aol':bool(self.always_on_lateral_active),'ui_status':str(self.status),**counts})+'\n');last=now
  return result
 UIState.update=update;CameraView._accept_frame=accept;CameraView._render_textures=textures;ModelRenderer._draw_path=path;ModelRenderer._draw_lane_lines=lanes
 from overlay import install

roadscore/prototype/operator_output.py

@@ -439,7 +439,7 @@ class OutputOwner:
 
 PRESENTATION_FIELDS = ('phase', 'kind', 'amount', 'activation', 'strength', 'section', 'next_section',
                        'gesture_active', 'gesture_queued', 'turn_signal_music', 'lead', 'predicted_peak', 'scheduled',
-                       'signal_shaker', 'core_apex', 'alert_accent', 'engagement_presentation', 'motion_presentation', 'curve_reaction')
+                       'signal_shaker', 'core_apex', 'alert_accent', 'engagement_presentation', 'motion_presentation', 'curve_reaction', 'replay_demo')
 
 
 class PresentationDelay:

roadscore/prototype/overlay_view.py

@@ -36,6 +36,12 @@ def gesture_view(state):
   alert = state.get('alert_accent') or {}
   if alert.get('enabled') and alert.get('rendered_active'):
     return 'Road alert / Percussion', 'active', 'native_alert'
+  demo = state.get('replay_demo') or {}
+  signal = demo.get('signal_mode')
+  if signal in ('left', 'right', 'off'):
+    shaker = state.get('signal_shaker') or {}
+    description = 'Shaker' if shaker.get('rendered_active') else 'Replay signal' if signal != 'off' else 'Recorded signals muted'
+    return f'Simulated {signal} / {description}', 'active', 'simulated_signal'
   engagement = state.get('engagement_presentation') or {}
   if engagement.get('enabled') and not engagement.get('simulated') and engagement.get('input_fresh') and engagement.get('rendered_state') == 'transition':
     return ('Engaged / Opening music' if engagement.get('active') else 'Disengaged / Contained music'), 'active', 'engagement'

roadscore/prototype/replay_ui_controls.py

@@ -0,0 +1,112 @@
+"""Native replay UI view of acknowledged demo controls; never publishes messages."""
+import json
+import math
+import time
+from pathlib import Path
+
+from cue_timing import audible_state, REFERENCE
+from demo_engagement import MODES, SIGNAL_MODES
+
+
+def isolated_replay(environ):
+  return (environ.get('ROADSCORE_REPLAY_UI_CONTROLS') == '1'
+          and environ.get('SIMULATION') == '1'
+          and not environ.get('ZMQ')
+          and environ.get('OPENPILOT_PREFIX') == 'roadscore_replay')
+
+
+class ReplayUIControls:
+  def __init__(self, status_path, *, enabled=False, clock=time.monotonic):
+    self.path = Path(status_path)
+    self.enabled, self.clock = enabled, clock
+    self.session = None
+
+  def read(self):
+    if not self.enabled:
+      return 'recorded', 'recorded'
+    try:
+      state = json.loads(self.path.read_text())
+      now = self.clock()
+      stamp = state.get('command_wall')
+      session = state.get('presentation_session_id')
+      if (state.get('input_mode') != 'replay' or state.get('compute') == 'none'
+          or state.get('presentation_timing_reference') != REFERENCE
+          or not isinstance(session, str) or not session
+          or type(stamp) not in (int, float) or not math.isfinite(stamp)
+          or not 0 <= now - stamp <= 2):
+        return 'recorded', 'recorded'
+      if self.session is not None and session != self.session:
+        return 'recorded', 'recorded'
+      audible = audible_state(state, now)
+      age = audible.get('presentation_display_lateness_ms')
+      if type(age) not in (int, float) or not math.isfinite(age) or not 0 <= age <= 2000:
+        return 'recorded', 'recorded'
+      demo = audible.get('replay_demo', {})
+      mode, signal = demo.get('mode'), demo.get('signal_mode')
+      if mode in MODES and signal in SIGNAL_MODES:
+        self.session = session
+        return mode, signal
+    except (OSError, ValueError, AttributeError):
+      pass
+    return 'recorded', 'recorded'
+
+
+class ReplayStateView:
+  """Copy only UI-facing fields; original readers and freshness stay intact."""
+  def __init__(self, subscriber, controls):
+    self.subscriber, self.controls = subscriber, controls
+    self.mode = self.signal_mode = 'recorded'
+    self.changed = False
+    self.copies = {}
+
+  def __getattr__(self, name):
+    return getattr(self.subscriber, name)
+
+  @property
+  def updated(self):
+    if self.changed:
+      return {**self.subscriber.updated, 'selfdriveState': True}
+    return self.subscriber.updated
+
+  def update(self, *args, **kwargs):
+    result = self.subscriber.update(*args, **kwargs)
+    selection = self.controls.read()
+    # An operator override must not hide missing replay data.
+    if not all(self.subscriber.valid.get(s, False) and self.subscriber.alive.get(s, False)
+               for s in ('selfdriveState', 'carState')):
+      selection = ('recorded', 'recorded')
+    self.changed = selection[0] != self.mode
+    self.mode, self.signal_mode = selection
+    self.copies.clear()
+    return result
+
+  def __getitem__(self, service):
+    original = self.subscriber[service]
+    engagement = self.mode != 'recorded' and service in ('selfdriveState', 'starpilotCarState')
+    signals = self.signal_mode != 'recorded' and service == 'carState'
+    if not (engagement or signals):
+      return original
+    if service not in self.copies:
+      message = original.as_builder()
+      if service == 'selfdriveState':
+        message.enabled = message.active = self.mode == 'engaged'
+        message.state = 'enabled' if self.mode == 'engaged' else 'disabled'
+      elif service == 'starpilotCarState':
+        message.alwaysOnLateralEnabled = self.mode == 'disengaged'
+        message.pauseLateral = False
+      else:
+        message.leftBlinker = self.signal_mode == 'left'
+        message.rightBlinker = self.signal_mode == 'right'
+      self.copies[service] = message
+    return self.copies[service].as_reader()
+
+  def apply_native_mode(self, state):
+    if self.mode != 'recorded' and state.started:
+      # This demo deliberately shows native AOL mode when full engagement is off.
+      # These are UI instance fields, not real driving Params.
+      state.always_on_lateral_active = self.mode == 'disengaged'
+      state.switchback_mode_enabled = False
+
+  def snapshot(self):
+    return {'mode': self.mode, 'signal_mode': self.signal_mode,
+            'session_id': self.controls.session, 'scope': 'isolated-replay-ui'}

roadscore/prototype/test_demo_engagement.py

@@ -1,6 +1,6 @@
 import json,tempfile,unittest
 from pathlib import Path
-from demo_engagement import DemoEngagement,presentation_active,annotate
+from demo_engagement import DemoEngagement,presentation_active,presentation_signal,annotate
 
 class Tests(unittest.TestCase):
  def test_modes_are_session_scoped_and_resettable(self):
@@ -32,5 +32,21 @@ class Tests(unittest.TestCase):
   self.assertTrue(out['engagement_presentation']['simulated'])
   self.assertFalse(out['engagement_presentation']['recorded_active'])
   self.assertNotIn('simulated',original['engagement_presentation'])
+ def test_combined_selection_is_atomic_and_legacy_resets_signals(self):
+  with tempfile.TemporaryDirectory() as d:
+   c=DemoEngagement(d,'new','replay',clock=lambda:10)
+   value=dict(version=1,session_id='new',mode='disengaged',signal_mode='left',created_wall=10)
+   c.path.write_text(json.dumps(value));c.poll()
+   self.assertEqual(c.selection,('disengaged','left'))
+   value.pop('signal_mode');c.path.write_text(json.dumps(value));c.poll()
+   self.assertEqual(c.selection,('disengaged','recorded'))
+   value['signal_mode']='both';c.path.write_text(json.dumps(value));c.poll()
+   self.assertEqual(c.selection,('recorded','recorded'))
+ def test_signal_choice_preserves_recorded_and_suppresses_when_off(self):
+  self.assertTrue(presentation_signal('left',False))
+  self.assertTrue(presentation_signal('right',False))
+  self.assertFalse(presentation_signal('off',True))
+  self.assertTrue(presentation_signal('recorded',True))
+  self.assertFalse(presentation_signal('recorded',False))
 
 if __name__=='__main__':unittest.main()

roadscore/prototype/test_replay_ui_controls.py

@@ -0,0 +1,108 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+from cereal import log, car, custom
+from cue_timing import REFERENCE
+from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay
+
+
+class Subscriber:
+  def __init__(self):
+    self.builders = {
+      'selfdriveState': log.SelfdriveState.new_message(enabled=True, active=True, state='enabled', alertText1='Recorded alert'),
+      'starpilotCarState': custom.StarPilotCarState.new_message(alwaysOnLateralEnabled=False, pauseLateral=True),
+      'carState': car.CarState.new_message(leftBlinker=False, rightBlinker=True, vEgo=12),
+    }
+    self.updated = {key: False for key in self.builders}
+    self.valid = {key: True for key in self.builders}
+    self.alive = dict(self.valid)
+    self.logMonoTime = {key: 123 for key in self.builders}
+  def __getitem__(self, key):return self.builders[key].as_reader()
+  def update(self, *args, **kwargs):return 17
+
+
+class Tests(unittest.TestCase):
+  def setUp(self):
+    self.tmp = tempfile.TemporaryDirectory()
+    self.addCleanup(self.tmp.cleanup)
+    self.path = Path(self.tmp.name)/'status.json'
+    self.controls = ReplayUIControls(self.path, enabled=True, clock=lambda: 100)
+    self.sm = Subscriber()
+    self.view = ReplayStateView(self.sm, self.controls)
+  def state(self, mode='disengaged', signal='left', **patch):
+    state = dict(command_wall=99.5, input_mode='replay', presentation_session_id='session',
+                 replay_demo=dict(mode=mode, signal_mode=signal),presentation_timing_reference=REFERENCE,
+                 presentation_timeline=[dict(audible_wall=99.8,cues=dict(replay_demo=dict(mode=mode,signal_mode=signal)))])
+    state.update(patch)
+    self.path.write_text(json.dumps(state))
+  def test_full_native_state_without_mutating_recording(self):
+    before = {key:value.to_bytes() for key,value in self.sm.builders.items()}
+    for value in self.sm.builders.values():value.clear_write_flag()
+    self.state();self.assertEqual(self.view.update(0),17)
+    ss = self.view['selfdriveState']
+    self.assertFalse(ss.enabled);self.assertFalse(ss.active);self.assertEqual(str(ss.state),'disabled')
+    self.assertEqual(ss.alertText1,'Recorded alert')
+    self.assertTrue(self.view['starpilotCarState'].alwaysOnLateralEnabled)
+    self.assertFalse(self.view['starpilotCarState'].pauseLateral)
+    self.assertTrue(self.view['carState'].leftBlinker);self.assertFalse(self.view['carState'].rightBlinker)
+    self.assertEqual(self.view['carState'].vEgo,12)
+    self.assertTrue(self.view.updated['selfdriveState']);self.assertFalse(self.sm.updated['selfdriveState'])
+    self.assertIs(self.view.valid,self.sm.valid);self.assertIs(self.view.logMonoTime,self.sm.logMonoTime)
+    for key,value in self.sm.builders.items():self.assertEqual(before[key],value.to_bytes())
+    ui = SimpleNamespace(started=True, always_on_lateral_active=False, switchback_mode_enabled=True)
+    self.view.apply_native_mode(ui);self.assertTrue(ui.always_on_lateral_active);self.assertFalse(ui.switchback_mode_enabled)
+  def test_reset_restores_latest_source_and_signal_independence(self):
+    self.state();self.view.update()
+    self.state('engaged','right');self.view.update()
+    self.assertTrue(self.view['selfdriveState'].enabled);self.assertFalse(self.view['starpilotCarState'].alwaysOnLateralEnabled)
+    self.assertFalse(self.view['carState'].leftBlinker);self.assertTrue(self.view['carState'].rightBlinker)
+    self.state('recorded','off');self.view.update()
+    self.assertFalse(self.view['carState'].leftBlinker);self.assertFalse(self.view['carState'].rightBlinker)
+    self.sm.builders['selfdriveState'].enabled=False
+    self.state('recorded','recorded');self.view.update()
+    self.assertFalse(self.view['selfdriveState'].enabled);self.assertTrue(self.view['carState'].rightBlinker)
+  def test_no_override_without_fresh_same_replay_session(self):
+    self.state();self.view.update()
+    for patch in ({'command_wall':97},{'command_wall':101},{'command_wall':float('nan')},
+                  {'presentation_session_id':'different'}, {'input_mode':'live'}, {'compute':'none'}):
+      self.state(**patch);self.view.update()
+      self.assertEqual((self.view.mode,self.view.signal_mode),('recorded','recorded'))
+    self.state();self.sm.alive['carState']=False;self.view.update()
+    self.assertEqual(self.view.mode,'recorded')
+  def test_waits_for_audible_timeline(self):
+    self.state(presentation_timing_reference=REFERENCE,presentation_timeline=[
+      dict(audible_wall=101,cues=dict(replay_demo=dict(mode='disengaged',signal_mode='left'))) ])
+    self.view.update();self.assertEqual(self.view.mode,'recorded')
+    self.state(presentation_timing_reference=REFERENCE,presentation_timeline=[
+      dict(audible_wall=99.9,cues=dict(replay_demo=dict(mode='disengaged',signal_mode='right'))) ])
+    self.view.update();self.assertEqual((self.view.mode,self.view.signal_mode),('disengaged','right'))
+  def test_transport_gate_and_disabled_reader(self):
+    env=dict(ROADSCORE_REPLAY_UI_CONTROLS='1',SIMULATION='1',OPENPILOT_PREFIX='roadscore_replay')
+    self.assertTrue(isolated_replay(env))
+    for patch in ({'ZMQ':'1'}, {'OPENPILOT_PREFIX':''}, {'SIMULATION':'0'}, {'ROADSCORE_REPLAY_UI_CONTROLS':'0'}):
+      self.assertFalse(isolated_replay(env|patch))
+    self.state();self.controls.enabled=False;self.view.update();self.assertEqual(self.view.mode,'recorded')
+  def test_unverified_or_stale_audio_never_binds_or_overrides(self):
+    self.state(presentation_timing_reference='unknown');self.view.update()
+    self.assertIsNone(self.controls.session);self.assertEqual(self.view.mode,'recorded')
+    self.state(presentation_timeline=[]);self.view.update();self.assertIsNone(self.controls.session)
+    self.state(presentation_timeline=[dict(audible_wall=90,cues=dict(replay_demo=dict(mode='disengaged',signal_mode='left')))])
+    self.view.update();self.assertIsNone(self.controls.session);self.assertEqual(self.view.mode,'recorded')
+  def test_malformed_and_missing_status_revert(self):
+    self.state();self.view.update()
+    for text in ('[]','broken','null'):
+      self.path.write_text(text);self.view.update();self.assertEqual(self.view.mode,'recorded')
+    self.path.unlink();self.view.update();self.assertEqual(self.view.mode,'recorded')
+  def test_callback_controls_survive_output_delay_and_status_serialization(self):
+    from operator_output import PresentationDelay
+    delay=PresentationDelay(Path(self.tmp.name),output_provider=lambda:None,clock=lambda:100)
+    snapshot=dict(command_wall=100,input_mode='replay',presentation_session_id='new')
+    cues=dict(replay_demo=dict(mode='disengaged',signal_mode='right'))
+    state=delay.apply(snapshot,dict(sequence=1,callback_wall=99.5,dac_wall=99.8,cues=cues))
+    self.path.write_text(json.dumps(state));self.view.update()
+    self.assertEqual((self.view.mode,self.view.signal_mode),('disengaged','right'))
+
+if __name__=='__main__':unittest.main()