Launch saved paired demo with actual screen mirror and preserve signal cues

2026-10-04 13:24:13 +08:00 · 2026-09-20 11:06:58 -07:00
parent 8824508b69
commit 71bbdb089f
14 changed files with 287 additions and 22 deletions
@@ -1,6 +1,13 @@
 #!/usr/bin/env bash
 set -euo pipefail
 for argument in "$@"; do
+  if [[ "$argument" == "--demo" ]]; then
+    if [ -f /TICI ]; then
+      export PYTHONPATH="/data/roadscore-feasibility/venv/lib/python3.12/site-packages${PYTHONPATH:+:$PYTHONPATH}"
+      exec /usr/local/venv/bin/python "$(dirname "$0")/prototype/native_prepared_showcase.py" "$@"
+    fi
+    exec python3 "$(dirname "$0")/prototype/paired_showcase.py" "$@"
+  fi
  if [[ "$argument" == "--prepared-showcase" ]]; then
    exec python3 "$(dirname "$0")/prototype/mac_showcase.py" "$@"
  fi
@@ -243,6 +243,7 @@ def callback(out,n,ti,status):
  curve_fresh=road_model_valid and 0<=callback_wall-command_wall<=.5
  motion_blocked=bool(motion is not None and motion.stopped and motion.fresh)
  curve_state=event_state.get('demo_curve_state',event_state)
+  curve_state={**curve_state,'manual_signal_priority':demo_signal_mode in ('left','right') and signal_fresh}
  curve_grid=None
  if curve_state.get('demo_build_drop') and rhythm_timeline is not None:
   target_frame=frames-n+round((curve_state['predicted_peak']-source_time)*rate)
@@ -25,7 +25,7 @@ class CurveBuildDrop:
    self.status=dict(rendered_stage='bypass',rendered_active=False,added_delay_samples=0)

  def process(self, pcm, start_frame, grid, build_start_frame, payoff_frame, *, enabled=False,
-              source_fresh=True, blocked=False):
+              source_fresh=True, blocked=False, allow_roll=True):
    """Same frames, no input mutation. Caller supplies and authorizes the schedule.

    Disabling a fully bypassed instance is bit exact; cancelling an active effect
@@ -56,8 +56,9 @@ class CurveBuildDrop:
      if not np.any(strength):
        self.tail=np.zeros((0,2),np.float32);outputs.append(x);continue
      frames=self.indices[:n]+first
+      if not allow_roll:self.tail=np.zeros((0,2),np.float32)
      roll=np.zeros_like(x);take=min(n,len(self.tail));roll[:take]+=self.tail[:take];self.tail=self.tail[take:].copy()
-      if requested and first<cut_start and end>build:
+      if allow_roll and requested and first<cut_start and end>build:
        beat=self.rate*60/grid.bpm;origin=grid.beat_phase*self.rate
        # Eighth notes establish the motif; the final four beats use sixteenths.
        boundary=max(build,payoff-4*beat)
@@ -107,7 +108,7 @@ class CurveBuildDrop:
      if self.strength==0 and not valid:self.schedule=None
      result=pcm
    else:result=np.concatenate(outputs) if len(outputs)>1 else outputs[0]
-    self.status=dict(enabled=bool(enabled),input_fresh=bool(source_fresh),blocked=bool(blocked),grid_usable=bool(grid.usable),
+    self.status=dict(enabled=bool(enabled),input_fresh=bool(source_fresh),blocked=bool(blocked),grid_usable=bool(grid.usable),roll_allowed=bool(allow_roll),
      rendered_stage=stage,rendered_active=any_active,rendered_strength=self.strength,rendered_roll_peak=max(peaks,default=0.),
      rendered_cut_gain_min=cut_min,planned_build_frame=None if self.schedule is None else self.schedule[0],
      planned_payoff_frame=None if self.schedule is None else self.schedule[1],actual_payoff_frame=self.actual_payoff,
@@ -49,7 +49,8 @@ class CurveReaction:
  result=self._filter(pcm,start_frame,filtered_state,source_fresh,blocked,self.impact_payoff if armed else None)
  if self.impact is not None:
   result=self.impact.process(result,start_frame,self.grid,self.impact_start,self.impact_payoff,
-                              enabled=armed,source_fresh=source_fresh,blocked=blocked)
+                              enabled=armed,source_fresh=source_fresh,blocked=blocked,
+                              allow_roll=not state.get('manual_signal_priority',False))
  if not valid:
   self.impact_activation=None;self.impact_payoff=None;self.impact_start=None;self.impact_grid=None
  return result
@@ -60,7 +60,8 @@ class DemoSession:
      result['ready_session_id'] = ready.get('session_id') if result['prepared'] else None
      state = read(out/'status.json')
      result['presentation_session_id'] = state.get('presentation_session_id') if running else None
-      result['failure'] = read(out/'failure.json').get('error')
+      result['failure'] = read(out/'demo_failure.json').get('error')
+      result['complete'] = read(out/'demo_complete.json').get('complete') is True
    return result

  def start(self, data):
@@ -159,8 +159,10 @@ def audio_worker(a):
  from demo_engagement import DemoEngagement
  from operator_output import PresentationDelay
  from prepared_core import load_archive, initial_frame, PreparedPresentation
+  from prepared_clock import PreparedClock
  audio, rate, meta = load_archive(a.score_archive, a.route)
  processor = PreparedPresentation(a.score_archive, rate)
+  playback_clock = PreparedClock(rate)
  session = os.environ['ROADSCORE_SHOWCASE_SESSION']
  controls = DemoEngagement(a.out, session, 'replay')
  delay = PresentationDelay(a.presentation_root or a.out,output_provider=lambda: {'id':a.output_identity} if a.output_identity else None)
@@ -192,14 +194,12 @@ def audio_worker(a):
    if anchor is None or done:return
    now=time.monotonic();dac=now+float(ti.outputBufferDacTime-ti.currentTime)
    expected=initial_frame(meta,anchor[0],dac-anchor[1],rate)
-    if position is None:position=expected;first_frame=position
-    start=position;position+=n
-    max_drift=max(max_drift,abs(start-expected)/rate)
    if status:flags+=1
-    chunk=np.zeros((n,2),np.float32)
-    lo=max(0,-start);hi=min(n,len(audio)-start)
-    if hi>lo:chunk[lo:hi]=audio[start+lo:start+hi]
    try:
+      chunk,clock_info=playback_clock.render(audio,expected,n)
+      start=clock_info['source_frame'];position=clock_info['next_source_frame']
+      if first_frame is None:first_frame=start
+      max_drift=max(max_drift,abs(clock_info['pre_error_frames'])/rate)
      wet,cues=processor.process(chunk,start,state,controls.selection)
      rendered=dict(sequence=start,callback_wall=now,dac_wall=dac,cues=cues)
      if not a.muted:out[:]=wet
@@ -241,6 +241,7 @@ def audio_worker(a):
          if anchor is not None and state['route_t'] is not None and np.isfinite(state['route_t']):
            snapshot.update(route_t=float(state['route_t']),elapsed=max(0,(position or 0)/rate),
                            source_model_ns=int(sm.logMonoTime['modelV2']))
+          snapshot['prepared_clock']=playback_clock.snapshot()
          snapshot=delay.apply(snapshot,rendered)
          # The command API must remain available during initial DAC lead-in.
          snapshot.setdefault('engagement_presentation',{'enabled':True})
@@ -257,9 +258,9 @@ def audio_worker(a):
          finally:server.server_close()
        elif forwarder is not None:forwarder.close()
      finally:trace.close()
-    write_json(a.out/'prepared_summary.json',dict(generation_invoked=False,source=str(a.score_archive),first_source_frame=first_frame,last_source_frame=position,sample_rate=rate,portaudio_flags=flags,max_clock_error_seconds=max_drift,callback_errors=errors,muted=a.muted,session_id=session,manual_scope='isolated replay display and presentation only'))
+    write_json(a.out/'prepared_summary.json',dict(generation_invoked=False,source=str(a.score_archive),first_source_frame=first_frame,last_source_frame=position,sample_rate=rate,portaudio_flags=flags,max_clock_error_seconds=max_drift,prepared_clock=playback_clock.snapshot(),callback_errors=errors,muted=a.muted,session_id=session,manual_scope='isolated replay display and presentation only'))
  if errors:raise RuntimeError(errors[0])
-  if max_drift>.05:raise RuntimeError('Prepared audio clock drift exceeded 50 ms')
+  if playback_clock.snapshot()['max_post_error_seconds']>.05:raise RuntimeError('Prepared audio clock drift remained above 50 ms after recovery')


 def apply_showcase_config(a, config):
@@ -74,6 +74,7 @@ def native_environment(project, root, out, session, inherited=None):
  env.update(PYTHONDONTWRITEBYTECODE='1',OPENPILOT_PREFIX='roadscore_replay',BASEDIR=str(project),PWD=str(project),NOBOARD='1',SIMULATION='1',SKIP_FW_QUERY='1',BIG='0',ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_SHOWCASE_SESSION=session,ROADSCORE_REPLAY_UI_CONTROLS='1',ROADSCORE_CLEAN_DEMO_UI='1',ROADSCORE_OVERLAY='1',ROADSCORE_STATUS_FILE=str(out/'status.json'),ROADSCORE_UI_AUDIT=str(out/'ui_audit.jsonl'),ROADSCORE_OVERLAY_CAPTURE=str(out/'overlay.png'),ROADSCORE_AUDIO_DRAIN_FILE=str(out/'audio_drained.json'),ROADSCORE_PRESENTATION_POLICY='conservative-v4',SP_ALLOW_DESKTOP_FAKE_WIFI='0',SP_ALLOW_DESKTOP_FAKE_BLUETOOTH='0',SP_ONROAD_NAV_DEMO='0',SP_ONROAD_CEM_DEMO='0')
  paths=[HERE,project,project/'starpilot/third_party',*project.glob('*_repo'),Path('/data/roadscore-feasibility/venv/lib/python3.12/site-packages')]
  env['PYTHONPATH']=':'.join(map(str,paths))
+  env['ROADSCORE_MIRROR_DIR']=str(out/'mirror')
  return env


@@ -112,11 +113,11 @@ def main():
  from route_favorites import resolve_favorite
  route,_=resolve_favorite(a.route,root/'routes/favorites.json')
  if a.score_archive is None:
-    registry=json.loads((root/'assets/prepared_showcase.json').read_text())
-    entry=registry.get('routes',{}).get(a.route,registry)
-    if entry.get('route')!=route:raise ValueError('No matching prepared showcase registered')
-    a.score_archive=Path(entry['archive'])
-    if a.curve_plan is None and entry.get('curve_plan'):a.curve_plan=Path(entry['curve_plan'])
+    from demo_catalog import entry
+    selected=entry(root,a.route)
+    if selected['route']!=route:raise ValueError('No matching prepared showcase registered')
+    a.score_archive=Path(selected['archive'])
+    if a.curve_plan is None and selected.get('curve_plan'):a.curve_plan=Path(selected['curve_plan'])
  archive=a.score_archive.resolve(strict=True)
  from prepared_core import load_archive
  audio,rate,meta=load_archive(archive,route)
@@ -0,0 +1,16 @@
+<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>RoadScore demo</title>
+<style>body{margin:0;background:#070b0c;color:#f4f5f6;font:17px system-ui;text-align:center}main{max-width:1440px;margin:auto}header{display:flex;align-items:center;justify-content:space-between;padding:12px 20px}h1{font-size:22px;margin:0}.screen{position:relative;background:#000;aspect-ratio:2.23}img{width:100%;height:100%;object-fit:contain}.notice{position:absolute;inset:0;display:grid;place-items:center;background:#000b;font-size:24px}.notice[hidden]{display:none}section{display:flex;flex-wrap:wrap;gap:10px;justify-content:center;padding:12px}button,a{border:1px solid #506166;border-radius:10px;padding:12px 18px;background:#19262a;color:white;font:inherit;text-decoration:none;cursor:pointer}button.on{background:#0b6d59;border-color:#69dec3}button:disabled{opacity:.4;cursor:default}small{color:#a8b8bc}#error{color:#ffcc94;min-height:24px;margin:0 10px}body:fullscreen main{max-width:none}</style>
+<main><header><h1>RoadScore</h1><small id="state">Connecting to the comma…</small><button id="fullscreen">Full screen</button></header>
+<div class="screen"><img id="view" alt="Live mirror of the comma replay"><div class="notice" id="notice">Preparing saved replay…</div></div>
+<section><button data-field="mode" data-value="engaged">Demo engage</button><button data-field="mode" data-value="disengaged">Demo disengage</button><button data-field="mode" data-value="recorded">Recorded engagement</button></section>
+<section><button data-field="signal_mode" data-value="left">Left signal</button><button data-field="signal_mode" data-value="right">Right signal</button><button data-field="signal_mode" data-value="off">Signals off</button><button data-field="signal_mode" data-value="recorded">Recorded signals</button><a id="galaxy" target="_blank" rel="noopener">Open Galaxy</a></section>
+<p id="error"></p><small>Saved soundtrack · live presentation controls · actual comma screen</small></main>
+<script>
+const view=document.querySelector('#view'),notice=document.querySelector('#notice'),buttons=[...document.querySelectorAll('[data-field]')];let session,frame=0,blob;
+document.querySelector('#fullscreen').onclick=()=>document.fullscreenElement?document.exitFullscreen():document.documentElement.requestFullscreen();
+buttons.forEach(button=>button.onclick=async()=>{try{const response=await fetch('/control',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({session_id:session,field:button.dataset.field,value:button.dataset.value})});const result=await response.json();if(!response.ok)throw Error(result.error);document.querySelector('#error').textContent=''}catch(error){document.querySelector('#error').textContent=error.message}});
+async function status(){try{const response=await fetch('/status');const data=await response.json();if(!response.ok)throw Error(data.error);const demo=data.peer.demo;session=demo.session_id;buttons.forEach(button=>{button.disabled=!demo.available;button.classList.toggle('on',demo[button.dataset.field]===button.dataset.value)});document.querySelector('#state').textContent=data.phase+(data.muted?' · muted test':' · audio from comma');document.querySelector('#galaxy').href=data.galaxy_url}catch(error){buttons.forEach(button=>button.disabled=true);document.querySelector('#state').textContent='Connection unavailable'}setTimeout(status,250)}
+async function picture(){try{const response=await fetch('/frame?n='+frame++);if(!response.ok)throw Error('No fresh frame');const next=URL.createObjectURL(await response.blob());view.src=next;if(blob)URL.revokeObjectURL(blob);blob=next;notice.hidden=true}catch(error){notice.hidden=false;notice.textContent='Waiting for the comma screen…'}setTimeout(picture,30)}
+status();picture();
+</script>
@@ -0,0 +1,148 @@
+"""Launch one saved native replay and mirror its actual UI on the Mac."""
+import argparse
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import threading
+import time
+import uuid
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlencode
+from urllib.request import Request, build_opener, ProxyHandler
+
+from paired_demo_controls import GalaxyPeer, _NoRedirect, _http_json
+
+HERE = Path(__file__).resolve().parent
+
+
+class DemoPeer:
+  def __init__(self, url, *, transport=_http_json):
+    self.peer = GalaxyPeer(url, allow_lan_http=True)
+    self.transport = transport
+
+  def call(self, action, data=None):
+    return self.transport('GET' if action == 'status' else 'POST',
+                          self.peer.base_url+'/api/roadscore/'+action, data,
+                          {'Accept':'application/json','Content-Type':'application/json','Cache-Control':'no-store'}, 4.)
+
+  def frame(self, request_id):
+    url=self.peer.base_url+'/api/roadscore/demo_frame?'+urlencode({'request_id':request_id})
+    request=Request(url,headers={'Accept':'image/jpeg','Cache-Control':'no-store'})
+    with build_opener(ProxyHandler({}),_NoRedirect()).open(request,timeout=2.) as response:
+      if response.headers.get_content_type() != 'image/jpeg':raise ValueError('Expected a live replay image')
+      frame=response.read(1500001)
+      if len(frame)>1500000:raise ValueError('Replay frame is too large')
+      return frame, {key:response.headers.get(key,'') for key in ('X-RoadScore-Frame','X-RoadScore-Age-Ms')}
+
+
+def control_server(peer, request_id, port, state):
+  class Handler(BaseHTTPRequestHandler):
+    def log_message(self,*args):pass
+    def send(self, body, kind='application/json', code=200, headers=None):
+      self.send_response(code)
+      self.send_header('Content-Type',kind);self.send_header('Cache-Control','no-store')
+      self.send_header('Content-Length',str(len(body)))
+      for name,value in (headers or {}).items():self.send_header(name,value)
+      self.end_headers()
+      try:self.wfile.write(body)
+      except (BrokenPipeError,ConnectionResetError):pass
+    def json(self,value,code=200):self.send(json.dumps(value).encode(),code=code)
+    def do_GET(self):
+      if self.path == '/':self.send((HERE/'paired_showcase.html').read_bytes(),'text/html; charset=utf-8');return
+      if self.path == '/status':
+        try:self.json({'phase':state['phase'],'peer':peer.call('status'),'galaxy_url':peer.peer.base_url+'/mobile/#/roadscore','muted':state['muted']})
+        except Exception as error:self.json({'phase':state['phase'],'error':str(error)},503)
+        return
+      if self.path.startswith('/frame?'):
+        try:
+          frame,headers=peer.frame(request_id);self.send(frame,'image/jpeg',headers=headers)
+        except Exception:self.json({'error':'Waiting for a fresh comma replay frame'},503)
+        return
+      self.send_error(404)
+    def do_POST(self):
+      if self.path!='/control':self.send_error(404);return
+      expected={f'http://127.0.0.1:{self.server.server_port}',f'http://localhost:{self.server.server_port}'}
+      if ('http://'+self.headers.get('Host','') not in expected
+          or self.headers.get('Origin') not in expected
+          or self.headers.get('Sec-Fetch-Site','same-origin') != 'same-origin'):
+        self.json({'error':'Open controls from this local demo page'},403);return
+      try:
+        length=int(self.headers.get('Content-Length','0'))
+        if not 0<length<=1024:raise ValueError('Invalid control payload')
+        data=json.loads(self.rfile.read(length))
+        if not isinstance(data,dict) or set(data)!={'field','value','session_id'}:raise ValueError('Invalid control payload')
+        field=data['field'];allowed={'mode':('engaged','disengaged','recorded'),'signal_mode':('left','right','off','recorded')}
+        if field not in allowed or data['value'] not in allowed[field]:raise ValueError('Unknown replay control')
+        action='demo_engagement' if field=='mode' else 'demo_signal'
+        self.json(peer.call(action,{'session_id':data['session_id'],field:data['value']}))
+      except Exception as error:self.json({'error':str(error)},409)
+  server=ThreadingHTTPServer(('127.0.0.1',port),Handler)
+  threading.Thread(target=server.serve_forever,daemon=True).start()
+  return server
+
+
+def main():
+  parser=argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--roadscore',action='store_true');parser.add_argument('--demo',action='store_true')
+  parser.add_argument('route',nargs='?',default='route1')
+  parser.add_argument('--muted',action='store_true');parser.add_argument('--check',action='store_true')
+  parser.add_argument('--no-browser',action='store_true');parser.add_argument('--port',type=int)
+  parser.add_argument('--peer');parser.add_argument('--duration',type=float,default=float('inf'))
+  args=parser.parse_args()
+  root=HERE.parent
+  from demo_catalog import entry
+  selected=entry(root,args.route)
+  config=json.loads((root/'assets/demo_catalog.json').read_text())
+  peer=DemoPeer(args.peer or config.get('paired_comma'))
+  port=args.port if args.port is not None else config.get('controls_port',56976)
+  if type(port) is not int or not 0<=port<=65535:raise SystemExit('Invalid local controls port')
+  status=peer.call('status')
+  if status.get('offroad') is not True:raise SystemExit('Comma must be offroad for the saved demo')
+  if args.check:
+    print(json.dumps({'route':selected['route'],'alias':args.route,'peer_offroad':True,'generation_invoked':False,'controls_port':port}));return
+  request_id=uuid.uuid4().hex
+  state={'phase':'Preparing saved replay','muted':args.muted}
+  # Claim the requested local page before starting anything on the peer.
+  server=control_server(peer,request_id,port,state)
+  launched=False;released=False;began=time.monotonic()
+  def interrupt(*_):raise KeyboardInterrupt
+  signal.signal(signal.SIGTERM,interrupt)
+  try:
+    launched=True
+    peer.call('demo_start',{'alias':args.route,'request_id':request_id,'muted':args.muted})
+    url=f'http://127.0.0.1:{server.server_port}'
+    print('RoadScore saved demo. No model generation. Controls and mirrored screen: '+url,flush=True)
+    if not args.no_browser:subprocess.Popen(['/usr/bin/open',url],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
+    deadline=time.monotonic()+90
+    while True:
+      ready=peer.call('demo_ready',{})
+      if ready.get('request_id')!=request_id:raise RuntimeError('Device demo ownership changed')
+      if ready.get('failure'):raise RuntimeError(ready['failure'])
+      if not ready.get('running'):raise RuntimeError('Saved demo failed to prepare')
+      if ready.get('prepared'):break
+      if time.monotonic()>deadline:raise TimeoutError('Saved replay did not prepare within 90 seconds; no model is being compiled')
+      time.sleep(.25)
+    peer.call('demo_play',{'request_id':request_id,'session_id':ready['ready_session_id']});released=True
+    state['phase']='Playing saved replay'
+    print('Playing on comma; this Mac mirrors its actual UI. Galaxy controls both views.',flush=True)
+    started=time.monotonic()
+    while time.monotonic()-started<args.duration:
+      current=peer.call('demo_ready',{})
+      if current.get('request_id')!=request_id:raise RuntimeError('Device demo ownership changed')
+      if current.get('failure'):raise RuntimeError(current['failure'])
+      if not current.get('running'):
+        if current.get('complete'):break
+        raise RuntimeError('Device demo stopped before completion')
+      time.sleep(.5)
+    state['phase']='Replay finished'
+  finally:
+    if launched:
+      try:peer.call('demo_stop',{'request_id':request_id})
+      except Exception as error:print('Could not confirm device demo stop: '+str(error),flush=True)
+    server.shutdown();server.server_close()
+    print('Saved demo stopped. Elapsed %.1f seconds.'%(time.monotonic()-began),flush=True)
+
+
+if __name__=='__main__':main()
@@ -72,7 +72,8 @@ class PreparedPresentation:
    curve_grid=None
    if state['curve'].get('demo_build_drop') and state.get('route_t') is not None:
      curve_grid=self.rhythm.at(frame+round((state['curve']['predicted_peak']-state['route_t'])*self.rhythm.rate))
-    result = self.curve.process(result, frame, state['curve'], source_fresh=state['model_fresh'], blocked=self.alert.priority_active or (self.motion.stopped and self.motion.fresh),route_time=state.get('route_t'),payoff_grid=curve_grid)
+    curve_state={**state['curve'],'manual_signal_priority':signal_mode in ('left','right') and state['car_fresh']}
+    result = self.curve.process(result, frame, curve_state, source_fresh=state['model_fresh'], blocked=self.alert.priority_active or (self.motion.stopped and self.motion.fresh),route_time=state.get('route_t'),payoff_grid=curve_grid)
    result = self.motion.process(result, speed=state['speed'], source_fresh=state['car_fresh'])
    result = self.engagement.process(result, active, self.config)
    gain = self.contained_gain+(1-self.contained_gain)*min(self.engagement.mix, self.motion.dsp.mix)
@@ -0,0 +1,45 @@
+import json
+from types import SimpleNamespace
+import unittest
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+from paired_showcase import DemoPeer, control_server
+
+
+class MirrorControlTests(unittest.TestCase):
+  def setUp(self):
+    self.calls=[]
+    def call(action,data=None):
+      self.calls.append((action,data));return {'demo':{'available':True,'session_id':'device-session'}}
+    self.peer=SimpleNamespace(peer=SimpleNamespace(base_url='http://192.168.8.156:8082'),call=call,
+                              frame=lambda request:(b'\xff\xd8image\xff\xd9',{'X-RoadScore-Age-Ms':'12'}))
+    self.server=control_server(self.peer,'owned-request',0,{'phase':'Playing saved replay','muted':True})
+    self.url='http://127.0.0.1:'+str(self.server.server_port)
+    self.addCleanup(self.server.server_close);self.addCleanup(self.server.shutdown)
+  def post(self,data,origin=None):
+    return urlopen(Request(self.url+'/control',data=json.dumps(data).encode(),headers={'Content-Type':'application/json','Origin':origin or self.url}),timeout=2)
+  def test_controls_change_one_authoritative_peer(self):
+    with self.post({'session_id':'device-session','field':'mode','value':'disengaged'}) as response:
+      self.assertEqual(response.status,200)
+    self.assertEqual(self.calls,[('demo_engagement',{'session_id':'device-session','mode':'disengaged'})])
+  def test_cross_origin_and_arbitrary_action_rejected(self):
+    with self.assertRaises(HTTPError) as error:self.post({'session_id':'s','field':'mode','value':'engaged'},'http://elsewhere.invalid')
+    self.assertEqual(error.exception.code,403)
+    with self.assertRaises(HTTPError):self.post({'session_id':'s','field':'launch','value':'shell'})
+    self.assertEqual(self.calls,[])
+  def test_frame_is_native_image_with_age(self):
+    with urlopen(self.url+'/frame?n=1',timeout=2) as response:
+      self.assertEqual(response.headers.get_content_type(),'image/jpeg')
+      self.assertEqual(response.headers['X-RoadScore-Age-Ms'],'12')
+      self.assertEqual(response.read(),b'\xff\xd8image\xff\xd9')
+  def test_peer_stays_explicit_and_operations_are_bounded(self):
+    calls=[]
+    peer=DemoPeer('http://192.168.8.156:8082',transport=lambda *args:calls.append(args) or {'ok':True})
+    self.assertEqual(calls,[])
+    peer.call('demo_start',{'alias':'route1'})
+    self.assertEqual(calls[0][0],'POST');self.assertEqual(calls[0][-1],4.)
+    with self.assertRaises(ValueError):DemoPeer('http://unrelated.example:8082')
+
+
+if __name__=='__main__':unittest.main()
@@ -51,7 +51,7 @@ class PreparedTests(unittest.TestCase):
    rate=48000;payoff=2.5
    samples=np.arange(4*rate)/rate
    core=np.repeat((.15*np.sin(2*np.pi*90*samples))[:,None],2,axis=1).astype(np.float32)
-    before=core.copy();chunks=[]
+    before=core.copy();chunks=[];signal_blocks_during_build=0
    for frame in range(0,len(core),960):
      now=frame/rate
      curve=dict(kind='curve',phase='anticipation' if now<payoff else 'event',
@@ -60,10 +60,14 @@ class PreparedTests(unittest.TestCase):
      if now<.2 or now>=3.5:curve=dict(kind='curve',phase='neutral',amount=0.,activation=None)
      state=dict(active=True,signal_on=True,fresh=True,car_fresh=True,model_fresh=True,
                 speed=10.,alert_key='',alert_meaningful=False,curve=curve,route_t=now)
-      wet,_=engine.process(core[frame:frame+960],frame,state,('engaged','left'))
+      wet,cues=engine.process(core[frame:frame+960],frame,state,('engaged','left'))
+      if .5<now<2 and cues['signal_shaker']['rendered_active']:
+        signal_blocks_during_build+=1
+        self.assertEqual(cues['curve_reaction']['build_drop']['rendered_roll_peak'],0.)
      chunks.append(wet)
    result=np.concatenate(chunks)
    self.assertEqual(engine.curve.impact.actual_payoff,round(payoff*rate))
+    self.assertGreater(signal_blocks_during_build,5)
    self.assertLess(float(np.max(abs(result[round((payoff-.08)*rate):round((payoff-.02)*rate)]))),1e-6)
    self.assertGreater(float(np.sqrt(np.mean(result[round((payoff+.04)*rate):round((payoff+.12)*rate)]**2))),.09)
    np.testing.assert_array_equal(core,before)
@@ -296,12 +296,31 @@ class Operator:


 def register(app, params):
-  from flask import jsonify, request
+  from flask import jsonify, request, Response
  prototype=Path('/data/roadscore/prototype')
  if str(prototype) not in sys.path:sys.path.insert(0,str(prototype))
  from operator_output import real_offroad
  operator = Operator(offroad=real_offroad)

+  @app.route('/api/roadscore/demo_frame')
+  def roadscore_demo_frame():
+    if not real_offroad():return jsonify(error='Saved replay mirror is offroad only'),409
+    owner=operator.demo_controller().status()
+    if not owner.get('running') or request.args.get('request_id')!=owner.get('request_id'):
+      return jsonify(error='No matching saved replay is active'),409
+    folder=Path(owner['out'])/'mirror'
+    frame=read_json(folder/'frame.json')
+    stamp=frame.get('captured_wall')
+    age=time.monotonic()-stamp if type(stamp) in (int,float) and math.isfinite(stamp) else float('inf')
+    if not 0<=age<=1 or frame.get('session_id')!=owner.get('ready_session_id'):
+      return jsonify(error='Waiting for a fresh replay image'),503
+    try:data=(folder/'latest.jpg').read_bytes()
+    except OSError:return jsonify(error='Replay image unavailable'),503
+    if not data.startswith(b'\xff\xd8') or len(data)>1500000:return jsonify(error='Invalid replay image'),503
+    response=Response(data,mimetype='image/jpeg')
+    response.headers.update({'Cache-Control':'no-store','X-RoadScore-Frame':str(frame.get('frame_id','')),'X-RoadScore-Age-Ms':str(round(age*1000,1))})
+    return response
+
  @app.route('/api/roadscore/status')
  def roadscore_status():
    response = jsonify(operator.status(real_offroad()))
@@ -70,5 +70,24 @@ class DemoTests(unittest.TestCase):
  self.assertEqual(state['backend'],'Prepared local audio')
  self.assertEqual(state['demo']['playhead']['route_t'],12.)
  self.assertGreaterEqual(state['demo']['status_age_seconds'],0.)
+ def test_mirror_frame_requires_fresh_owned_session_and_offroad(self):
+  from flask import Flask
+  from unittest.mock import patch
+  import operator_output
+  app=Flask(__name__)
+  folder=self.run/'mirror';folder.mkdir()
+  (folder/'latest.jpg').write_bytes(b'\xff\xd8fixture\xff\xd9')
+  frame={'session_id':'ready-session','frame_id':1,'captured_wall':time.monotonic()}
+  (folder/'frame.json').write_text(json.dumps(frame))
+  with patch.object(module,'Operator') as factory,patch.object(operator_output,'real_offroad',return_value=True) as parked:
+   factory.return_value.demo_controller.return_value.status.return_value={'running':True,'request_id':'owned-request','ready_session_id':'ready-session','out':str(self.run)}
+   module.register(app,None);client=app.test_client()
+   self.assertEqual(client.get('/api/roadscore/demo_frame?request_id=wrong').status_code,409)
+   response=client.get('/api/roadscore/demo_frame?request_id=owned-request')
+   self.assertEqual(response.status_code,200);self.assertEqual(response.mimetype,'image/jpeg')
+   frame['captured_wall']=time.monotonic()-2;(folder/'frame.json').write_text(json.dumps(frame))
+   self.assertEqual(client.get('/api/roadscore/demo_frame?request_id=owned-request').status_code,503)
+   parked.return_value=False
+   self.assertEqual(client.get('/api/roadscore/demo_frame?request_id=owned-request').status_code,409)

 if __name__=='__main__':unittest.main()