Stage a measured showcase curve build with a musical bass return

roadscore/SHOWCASE.md

@@ -14,6 +14,10 @@ Mac seed auditions use `roadscore/experiments/ace_chestnut_20260916/audition_cac
 
 In Galaxy's RoadScore section, **Simulate disengage** gives the replay UI its native blue AOL appearance and contains the music; **Simulate engage** restores enabled UI/music. **Recorded engagement** resets that selection. **Left signal**, **Right signal**, **Signals off**, and **Use recorded signals** separately drive native arrows and the existing rhythm-aware shaker. The controls change only the isolated replay UI view and music: no recorded messages, vehicle controls or real Params are rewritten. They follow acknowledged audible-time state. Restore both selections independently. Device code checks passed; an audible/visual operator rehearsal remains required before judging. Proposed engagement sequence: contain at0:10, open at0:15, recorded at0:19.
 
+The next presentation revision uses native **Steer Left/Right** prompts labeled **Replay simulation**, with existing alerts taking priority. Conservative-v4 strengthens the engagement contrast and mixes the shaker after containment so its high frequencies remain audible. Manual direction changes can restart the motif after a one-bar cooldown. Stale inputs and uncertain beat timing still suppress percussion; a button press is not proof that a shaker pulse rendered.
+
+For this explicitly staged showcase, an optional private `assets/showcase_curve_plan.json` identifies measured route events and the exact replay start. Only a matching native fresh v4 replay may apply it; live input and official judging ignore it. It shapes the existing song's bass during the approach and restores full bandwidth at the next suitable eighth-note boundary around the measured apex. The original composer state is unchanged. The plan and its hash remain in the captured archive. Ordinary routes without a matching plan retain their causal conductor timing.
+
 Use the protected recorded fallback if preparation or hardware is unreliable. The local rehearsal package is `Desktop/RoadScore/demos/showcase-rehearsal/`; `SHOW DEMO.command` opens the route E presentation cut over matching camera footage: 5:00.35, unchanged soundtrack until a two-second closing fade. The unedited exact soundtrack remains protected. The full judge briefing, evidence inventory, timestamp script, and cheat sheet are in that package. This is recorded playback, not fresh generation. The original 315.4-second recording includes an approximately 15-second historical EOF tail; the source app ended with an input-stall error despite uninterrupted audio. It is not the latest device run and has neither native UI nor the newest presentation refinements.
 
 Do not infer audible accents from source alerts alone. In the locked soundtrack, the first strong signal motif begins about 0:24.6, curve payoffs occur around 0:52.7 and 2:01.3, and recorded engagement switches at 2:10.8 and 2:15.9. Fresh generations may have different beat phase. Let the music establish and leave silence around those demonstrations.

roadscore/prototype/app.py

@@ -18,6 +18,7 @@ from motion_presentation import MotionPresentation
 from signal_shaker import SignalShaker,profile_tempo_prior
 from core_apex import CoreApex
 from curve_reaction import CurveReaction
+from demo_curve_plan import ReplayCurvePlan
 from alert_accent import AlertAccent
 from rhythm_timeline import RhythmTimeline
 from presentation_status import export_status
@@ -78,6 +79,8 @@ from ace_profiles import selected,PROFILES
 ace_profile=selected()
 if composer=='ace':
  from ending_policy import safe_ending_gesture as ending_gesture
+demo_curve_plan=ReplayCurvePlan.from_environment(a.input,os.environ)
+if demo_curve_plan:(run/'demo_curve_plan.json').write_text(json.dumps(demo_curve_plan.value,indent=2))
 config=effective_config(configuration());validate_render_mode(render_mode,composer,config.get('song_form_experimental',False));identity=config.get('identity','legacy');identity=identity if identity in styles else 'legacy'
 rolling_mode=config.get('rolling',False);rolling_anchors={k:anchor_options(root,k) for k in styles if (root/f'assets/source_{k}.wav').exists()} if rolling_mode and composer!='ace' else {}
 drive_events=config.get('drive_events',False) or config.get('event_music',False);phrase_runway=config.get('phrase_runway',False)
@@ -141,7 +144,7 @@ if shaker_enabled or apex_enabled or alert_enabled or curve_enabled:
  shaker=SignalShaker(shaker_grid,rate,enabled=shaker_enabled,peak=config.get('signal_shaker',{}).get('peak',.012))
  apex=CoreApex(shaker_grid,rate,enabled=apex_enabled and not curve_enabled,dip_db=config.get('core_apex',{}).get('dip_db',-1.))
  alert_accent=AlertAccent(shaker_grid,rate,enabled=alert_enabled)
- curve_reaction=CurveReaction(shaker_grid,rate,enabled=curve_enabled)
+ curve_reaction=CurveReaction(shaker_grid,rate,enabled=curve_enabled,bass_build=config.get('curve_reaction',{}).get('bass_build') is True)
  (run/'shaker_grid.json').write_text(json.dumps({'grid':shaker.snapshot(),'analysis':rhythm_timeline.snapshot()},indent=2))
 if config.get('composition_control',False):
  from composition_policy import CompositionPolicy
@@ -239,7 +242,7 @@ def callback(out,n,ti,status):
  if curve_reaction is not None:
   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)
-  rendered=curve_reaction.process(rendered,frames-n,event_state,source_fresh=curve_fresh,blocked=alert_priority or motion_blocked)
+  rendered=curve_reaction.process(rendered,frames-n,event_state.get('demo_curve_state',event_state),source_fresh=curve_fresh,blocked=alert_priority or motion_blocked)
  if motion is not None:
   _,motion_fresh=engagement_active(motion_state[1],True,motion_state[2],motion_state[4],motion_state[3],callback_wall)
   rendered=motion.process(rendered,speed=motion_state[0],source_fresh=motion_fresh)
@@ -262,7 +265,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,'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}))
+ 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,'demo_curve_state':event_state.get('demo_curve_state'),'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'
@@ -322,7 +325,8 @@ try:
     else:state=con.state(now)
     samples=[(float(t),float(y)) for t,y in zip(m.orientationRate.t,m.orientationRate.z) if 1<=t<=8]
     state['predicted_turn_radians']=sum((y0+y1)*.5*(t1-t0) for (t0,y0),(t1,y1) in zip(samples,samples[1:]) if 0<t1-t0<2)
-    render_state=(state,now,time.monotonic())
+    presentation_state={**state,'demo_curve_state':demo_curve_plan.state(now,clock['route'],state)} if demo_curve_plan else state
+    render_state=(presentation_state,now,time.monotonic())
     if sm.updated['navRoute']:
      nav={};arrival_since=None;nav_revision+=1
      arrival.route_change(bool(sm.valid['navRoute']))

roadscore/prototype/curve_reaction.py

@@ -1,13 +1,21 @@
 """Live conductor-driven curve contrast, using only the current music samples."""
 import math
+import numpy as np
+from scipy.signal import butter
 from engagement_presentation import EngagementPresentation,PresentationConfig
 
 
 class CurveReaction:
- def __init__(self,grid,rate=48000,enabled=False):
+ def __init__(self,grid,rate=48000,enabled=False,bass_build=False):
   self.grid=grid;self.rate=rate;self.enabled=enabled
-  self.dsp=EngagementPresentation(rate,cutoff_hz=1800.,width=.92,gain=1.)
-  self.config=PresentationConfig(enabled=True,attack_ms=130.,release_ms=180.)
+  self.bass_build=bool(bass_build);self.cutoff=300. if self.bass_build else 1800.
+  self.dsp=EngagementPresentation(rate,cutoff_hz=self.cutoff,width=.92,gain=.92 if self.bass_build else 1.)
+  if self.bass_build:
+   # Fixed coefficients are prepared before playback. The existing sample ramp
+   # blends toward this high-pass, then restores the track's own bass at apex.
+   self.dsp.sos=butter(2,self.cutoff,btype='highpass',fs=rate,output='sos').astype(np.float32)
+  self.config=PresentationConfig(enabled=True,attack_ms=80. if self.bass_build else 130.,release_ms=180.)
+  self.staged=False
   self.activation=None;self.payoff=None;self.target=1.;self.phase='neutral';self.fresh=False;self.end_frame=0;self.valid=False;self.armed=False
 
  def process(self,pcm,start_frame,state,source_fresh=True,blocked=False):
@@ -15,6 +23,7 @@ class CurveReaction:
   amount=state.get('amount',0.)
   self.fresh=bool(source_fresh and isinstance(amount,(int,float)) and math.isfinite(amount))
   phase=state.get('phase','neutral');activation=state.get('activation')
+  self.staged=state.get('demo_staged_curve') is True
   identified=isinstance(activation,(int,float)) and not isinstance(activation,bool) and math.isfinite(activation)
   valid=self.enabled and self.fresh and not blocked and state.get('kind')=='curve' and identified
   self.valid=bool(valid)
@@ -24,7 +33,7 @@ class CurveReaction:
    self.payoff=None;self.armed=False
   target=1.
   if valid and phase=='anticipation':
-   target=1.-.8*min(1.,max(0.,amount));self.payoff=None
+   target=1.-(.98 if self.bass_build else .8)*min(1.,max(0.,amount));self.payoff=None
   elif valid and phase=='event' and self.armed:
    if self.payoff is None:
     step=self.rate*60/self.grid.bpm/2 if self.grid.usable else 0.
@@ -35,7 +44,6 @@ class CurveReaction:
     split=min(len(pcm),self.payoff-start_frame)
     first=self.dsp.process(pcm[:split],False,self.config,target_mix=target)
     if split<len(pcm):
-     import numpy as np
      second=self.dsp.process(pcm[split:],False,self.config,target_mix=1.)
      self.target=1.;self.phase=phase
      return np.concatenate((first,second))
@@ -51,6 +59,7 @@ class CurveReaction:
   build=bool(self.valid and self.armed and self.phase in ('anticipation','event') and self.dsp.mix<.999)
   return {'curve_reaction':dict(enabled=self.enabled,input_fresh=self.fresh,phase=self.phase,
           rendered_phase=('apex' if apex else 'build' if build else 'neutral'),
-          rendered_active=apex or build,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=1800.,maximum_wet_mix=.8,
-          payoff_audio_s=None if self.payoff is None else self.payoff/self.rate,attack_ms=130.,contain_ms=180.,
-          added_delay_samples=0,source='current causal conductor state',unknown_policy='smooth bypass')}
+          rendered_active=apex or build,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=self.cutoff,maximum_wet_mix=.98 if self.bass_build else .8,
+          treatment='bass-return' if self.bass_build else 'filter-opening',filter_type='highpass' if self.bass_build else 'lowpass',
+          payoff_audio_s=None if self.payoff is None else self.payoff/self.rate,attack_ms=self.config.attack_ms,contain_ms=180.,
+          added_delay_samples=0,source='known replay route event' if self.staged else 'current causal conductor state',demo_staged=self.staged,unknown_policy='smooth bypass')}

roadscore/prototype/demo_curve_plan.py

@@ -0,0 +1,65 @@
+"""Optional, disclosed replay staging over a real route's measured curve events.
+
+The launcher validates the private plan and passes it to presentation only.
+No route recording is opened by the audio callback or composition planner.
+"""
+import json
+import math
+from pathlib import Path
+
+
+def validate(value):
+ if not isinstance(value,dict) or value.get('version')!=1 or value.get('demo_only') is not True:
+  raise ValueError('Expected a version 1 demo-only curve plan')
+ if not isinstance(value.get('route'),str) or not value['route']:
+  raise ValueError('Curve plan requires its exact route identity')
+ if type(value.get('replay_start')) is not int or value['replay_start']<0:
+  raise ValueError('Curve plan requires its exact replay start')
+ events=value.get('curves')
+ if not isinstance(events,list) or not 1<=len(events)<=3:
+  raise ValueError('Curve plan requires one to three measured events')
+ previous=-1.
+ for event in events:
+  if not isinstance(event,dict):raise ValueError('Invalid curve event')
+  values=[event.get(k) for k in ('build_start','apex','end')]
+  if not all(type(n) in (int,float) and math.isfinite(n) for n in values):
+   raise ValueError('Curve event timing must be finite')
+  start,apex,end=values
+  if not (0<=start<apex<end and 1<=apex-start<=24 and .4<=end-apex<=5 and start>previous):
+   raise ValueError('Curve event timing is out of bounds or overlaps')
+  if not isinstance(event.get('evidence'),str) or not event['evidence'].strip():
+   raise ValueError('Curve event requires measured route evidence')
+  previous=end
+ return value
+
+
+def launch_plan(path,route,replay_start,*,native,replay,judging,policy):
+ if not native or replay or judging or policy!='conservative-v4':return None
+ path=Path(path)
+ if not path.is_file():return None
+ value=validate(json.loads(path.read_text()))
+ return value if value['route']==route and value['replay_start']==replay_start else None
+
+
+class ReplayCurvePlan:
+ def __init__(self,value):
+  self.value=validate(value)
+
+ @classmethod
+ def from_environment(cls,mode,environ):
+  raw=environ.get('ROADSCORE_DEMO_CURVE_PLAN')
+  if not raw or mode!='replay' or environ.get('ROADSCORE_SEED_ORIGIN')=='judging-route':return None
+  if environ.get('ROADSCORE_PRESENTATION_POLICY')!='conservative-v4':return None
+  return cls(json.loads(raw))
+
+ def state(self,now,route,recorded):
+  if route!=self.value['route'] or type(now) not in (int,float) or not math.isfinite(now):return recorded
+  for event in self.value['curves']:
+   start,apex,end=(event[k] for k in ('build_start','apex','end'))
+   if start<=now<end:
+    progress=min(1.,max(0.,(now-start)/(apex-start)))
+    amount=progress*progress*(3.-2.*progress)
+    return {**recorded,'kind':'curve','activation':start,'predicted_peak':apex,
+            'phase':'anticipation' if now<apex else 'event','amount':amount,
+            'demo_staged_curve':True}
+  return recorded

roadscore/prototype/normal_onroad.py

@@ -8,6 +8,7 @@ from pathlib import Path
 from clock_sync import measure
 from receiver_environment import assignments as receiver_assignments,resolve_compute
 from presentation_policy import select_launch
+from demo_curve_plan import launch_plan
 from hook_launch import start_planner
 from composition_launch import configure as configure_composition
 from launch_health import check_children, describe_failure, write_failure
@@ -20,7 +21,7 @@ if native:
  restore_links(R)
 def interrupt(*_):raise KeyboardInterrupt
 signal.signal(signal.SIGTERM,interrupt)
-p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default=None,help='ACE rendering mode');p.add_argument('--roadscore-presentation',choices=['conservative-v3','conservative-v2','conservative-v1','off','frozen']);p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--score-archive',type=Path,help='Imported local archive for --replay');p.add_argument('--start',type=int,default=None);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
+p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default=None,help='ACE rendering mode');p.add_argument('--roadscore-presentation',choices=['conservative-v4','conservative-v3','conservative-v2','conservative-v1','off','frozen']);p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--score-archive',type=Path,help='Imported local archive for --replay');p.add_argument('--start',type=int,default=None);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
 explicit_start=a.start
 if a.render_mode=='gold-core' and a.composer!='ace':raise SystemExit('Gold core requires ACE')
 if a.roadscore_seed is not None and (a.replay or a.composer!='ace'):raise SystemExit('--roadscore-seed applies only to fresh ACE generation')
@@ -57,6 +58,12 @@ if session:print('Requested Chestnut cap:',compute['requested_power_limit_watts'
 (out/'settings.json').write_text(json.dumps({**settings.snapshot(a.headless),'composer':a.composer,'profile':a.profile,'render_mode':a.render_mode,'presentation_policy':presentation['policy'],'composition_policy':composition_policy,**(session or {})},indent=2))
 env['ROADSCORE_RENDER_MODE']=a.render_mode
 env['ROADSCORE_PRESENTATION_POLICY']=presentation['policy']
+env.pop('ROADSCORE_DEMO_CURVE_PLAN',None)
+demo_curve=launch_plan(R/'assets/showcase_curve_plan.json',a.routeid,a.start,native=native,replay=a.replay,judging=bool(session and session['seed_origin']=='judging-route'),policy=presentation['policy'])
+if demo_curve:
+ env['ROADSCORE_DEMO_CURVE_PLAN']=json.dumps(demo_curve,separators=(',',':'))
+ (out/'demo_curve_plan.json').write_text(json.dumps(demo_curve,indent=2))
+ print('Showcase presentation: staged build and bass return at a measured route curve.',flush=True)
 env['ROADSCORE_COMPOSITION_POLICY']=composition_policy
 env['ROADSCORE_COMPOSER']=a.composer
 env['ROADSCORE_ACE_PROFILE']=a.profile

roadscore/prototype/overlay.py

@@ -53,6 +53,7 @@ def install():
   for service in ('selfdriveState','starpilotSelfdriveState'):
    size=ui_state.sm[service].alertSize
    if int(size.raw)>0:alert_clear_after=now+1.
+  if getattr(ui_state,'roadscore_replay_prompt_active',False):alert_clear_after=now+1.
   frames+=1
   if frames==1:path.with_name('ui_capture_origin.json').write_text(json.dumps({'wall':time.monotonic(),'frame':0}))
   startup=home_footer_right is not None or not ui_state.started

roadscore/prototype/presentation_policy.py

@@ -16,6 +16,7 @@ CONSERVATIVE_V3 = {**copy.deepcopy(CONSERVATIVE_V2), 'presentation_policy_versio
 
 CONSERVATIVE_V4 = {**copy.deepcopy(CONSERVATIVE_V3),
  'presentation_policy_version':'conservative-v4',
+ 'curve_reaction':{'enabled':True,'bass_build':True},
  'signal_shaker':{'enabled':True,'peak':.055,'after_containment':True,'contained_gain':.45},
  'engagement_presentation':{'version':2,'enabled':True,'attack_ms':220,'release_ms':650,
                           'cutoff_hz':1300.,'width':.65,'gain':.7079458},

roadscore/prototype/review_curve_build.py

@@ -0,0 +1,50 @@
+"""Render a measured replay curve over unchanged archived core PCM, offline."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import time
+import numpy as np
+import soundfile as sf
+from curve_reaction import CurveReaction
+from demo_curve_plan import ReplayCurvePlan
+from rhythm_timeline import RhythmTimeline
+from signal_shaker import ShakerGrid
+
+
+def main():
+ parser=argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('archive',type=Path);parser.add_argument('plan',type=Path);parser.add_argument('output',type=Path)
+ args=parser.parse_args();args.output.mkdir(parents=True,exist_ok=False)
+ plan=ReplayCurvePlan(json.loads(args.plan.read_text()));event=plan.value['curves'][0]
+ rows=[json.loads(line) for line in (args.archive/'audio_blocks.jsonl').read_text().splitlines()]
+ timeline=RhythmTimeline(48000,128.)
+ timeline.entries=tuple((row['start_frame'],row['end_frame'],ShakerGrid(**row['grid'])) for row in json.loads((args.archive/'rhythm_timeline.json').read_text()))
+ start=int(event['build_start'])-4;end=int(event['end'])+4;rate=48000
+ with sf.SoundFile(args.archive/'dry.wav') as source:
+  if source.samplerate!=rate or source.channels!=2:raise ValueError('Expected native 48kHz stereo core')
+  source.seek(start*rate);wave=source.read((end-start)*rate,dtype='float32',always_2d=True)
+ lookup={round(row['audio_s']*10):row for row in rows}
+ report={'source':str(args.archive/'dry.wav'),'source_sha256':hashlib.sha256((args.archive/'dry.wav').read_bytes()).hexdigest(),
+         'plan':plan.value,'clip_audio_start':start,'clip_seconds':len(wave)/rate,
+         'camera_start_s':lookup[start*10]['route_t'],
+         'scope':'Same native core PCM and gain. Curve layer only; other presentation effects omitted. No regeneration.',
+         'files':{}}
+ for name,staged in [('before-curve.wav',False),('after-curve.wav',True)]:
+  dsp=CurveReaction(timeline.at(start*rate),rate,enabled=True,bass_build=staged);output=[];events=[];timings=[]
+  for offset in range(0,len(wave),4800):
+   frame=start*rate+offset;row=lookup[round(frame/rate*10)]
+   state={key:row.get(key) for key in ('kind','phase','amount','activation','predicted_peak')}
+   if staged:state=plan.state(row['route_t'],plan.value['route'],state)
+   dsp.grid=timeline.at(frame)
+   began=time.perf_counter();output.append(dsp.process(wave[offset:offset+4800],frame,state));timings.append(time.perf_counter()-began)
+   events.append({'audio_s':frame/rate,'route_s':row['route_t'],**dsp.snapshot()['curve_reaction']})
+  audio=np.concatenate(output);sf.write(args.output/name,audio,rate,subtype='FLOAT')
+  report['files'][name]={'peak':float(abs(audio).max()),'over_full_scale':int(np.count_nonzero(abs(audio)>1)),
+                       'work_mean_ms':float(np.mean(timings)*1000),'work_max_ms':float(max(timings)*1000),'states':events}
+ sf.write(args.output/'unchanged-core.wav',wave,rate,subtype='FLOAT')
+ (args.output/'comparison.json').write_text(json.dumps(report,indent=2)+'\n')
+ print(json.dumps({name:{k:v for k,v in row.items() if k!='states'} for name,row in report['files'].items()},indent=2))
+
+
+if __name__=='__main__':main()

roadscore/prototype/score_archive.py

@@ -1,5 +1,5 @@
 """Route-owned exact rendered score and compact decisions; no regeneration recipe."""
-import json,shutil,subprocess,time
+import hashlib,json,shutil,subprocess,time
 from pathlib import Path
 from route_library import ROOT,identity
 
@@ -12,6 +12,13 @@ def prefer_new_score(new,old):
 def recording_complete(blocks, launch, bridge):
  return bool(blocks) and launch.get('end_reason')=='native final segment exhausted' and not bridge.get('failure')
 
+def staging_provenance(run):
+ path=Path(run)/'demo_curve_plan.json'
+ if not path.is_file():return {}
+ return {'presentation_demo_only':True,'demo_curve_plan_file':path.name,
+         'demo_curve_plan_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
+         'presentation_staging':'Known recorded curve timing shapes presentation only; composer inputs remain causal.'}
+
 def archive(route,run,start=0):
  dongle,name=identity(route);parent=ROOT/'routes'/dongle/name/'roadscore';parent.mkdir(parents=True,exist_ok=True)
  dest=parent/run.name;dest.mkdir(exist_ok=False)
@@ -19,7 +26,7 @@ def archive(route,run,start=0):
  (run/'host_heard.flac').symlink_to(dest/'score.flac')
  if (run/'quality').exists():
   shutil.move(str(run/'quality'),str(dest/'quality'));(run/'quality').symlink_to(dest/'quality',target_is_directory=True)
- for filename in ['host_audio.jsonl','host_audio_summary.json','clock_sync.json','clock_sync_after.json','launch.json','summary.json','jobs.jsonl','boundaries.jsonl','ending.json','bridge.json','runtime_manifest.json','song_form.json','gesture_grid.json','gestures.json','shaker_grid.json','shaker_events.json','core_apex_events.json','composition.json','settings.json','quality_events.jsonl','ace_link.jsonl']:
+ for filename in ['host_audio.jsonl','host_audio_summary.json','clock_sync.json','clock_sync_after.json','launch.json','summary.json','jobs.jsonl','boundaries.jsonl','ending.json','bridge.json','runtime_manifest.json','song_form.json','gesture_grid.json','gestures.json','shaker_grid.json','shaker_events.json','core_apex_events.json','composition.json','settings.json','quality_events.jsonl','ace_link.jsonl','demo_curve_plan.json']:
   if (run/filename).exists():shutil.copy2(run/filename,dest/filename)
  blocks=[b for x in (run/'host_audio.jsonl').read_text().splitlines() if (b:=json.loads(x)).get('audio_s') is not None]
  # The first model's absolute time relative to native route start is recorded by sender.
@@ -38,6 +45,7 @@ def archive(route,run,start=0):
  meta['full_route']=meta['complete'] and start==0
  meta['audio_seconds']=blocks[-1]['audio_s']+.1 if blocks else 0
  meta['output_underruns']=audio_stats.get('portaudio_flags',0)
+ meta.update(staging_provenance(run))
  meta['capture_timing_clean']=not any(audio_stats.get(k,0) for k in ['portaudio_flags','starved_callbacks','late_frames'])
  if not meta['capture_timing_clean']:meta['timing_limitation']='Lossless rendered PCM and per-block DAC timing are retained. Hardware output interruptions are not reproduced by contiguous stored playback.'
  (dest/'metadata.json').write_text(json.dumps(meta,indent=2))
@@ -45,8 +53,8 @@ def archive(route,run,start=0):
  last=None
  with (dest/'events.jsonl').open('w') as out:
   for b in blocks:
-   event={k:b.get(k) for k in ['route_t','audio_s','phase','kind','activation','strength','predicted_peak','cadence_entry_audio_s']}
-   key=tuple(event.get(k) for k in ['phase','kind','activation','cadence_entry_audio_s'])
+   event={k:b.get(k) for k in ['route_t','audio_s','phase','kind','activation','strength','predicted_peak','cadence_entry_audio_s','demo_curve_state']}
+   key=tuple(event.get(k) for k in ['phase','kind','activation','cadence_entry_audio_s'])+(json.dumps(event['demo_curve_state'],sort_keys=True),)
    if key!=last:out.write(json.dumps(event)+'\n');last=key
  if (run/'trace.jsonl').exists():
   last=None

roadscore/prototype/test_curve_reaction.py

@@ -61,5 +61,24 @@ class CurveTests(unittest.TestCase):
   for i in range(10):self.assertFalse(np.any(dsp.process(silence,i*4800,self.state)))
   dsp.process(silence,48000,{**self.state,'phase':'event'})
   self.assertEqual(dsp.payoff,48000)
+ def test_bass_build_removes_low_end_then_restores_exact_song(self):
+  rate=48000;t=np.arange(rate*5)/rate
+  source=np.column_stack([.22*np.sin(2*np.pi*80*t)+.1*np.sin(2*np.pi*2000*t)]*2).astype('float32')
+  original=source.copy();dsp=CurveReaction(GRID,enabled=True,bass_build=True);output=[]
+  for start in range(0,len(source),4800):
+   now=start/rate
+   state={**self.state,'amount':min(1.,now/3),'phase':'anticipation' if now<3.5 else 'event','demo_staged_curve':True}
+   output.append(dsp.process(source[start:start+4800],start,state))
+  wave=np.concatenate(output)
+  def magnitude(samples,hz):
+   spectrum=np.abs(np.fft.rfft(samples[:,0]));return spectrum[round(hz*len(samples)/rate)]
+  before=source[rate*3:rate*3+4800];build=wave[rate*3:rate*3+4800]
+  self.assertLess(magnitude(build,80),magnitude(before,80)*.15)
+  self.assertGreater(magnitude(build,2000),magnitude(before,2000)*.8)
+  self.assertTrue(np.array_equal(wave[rate*4:],source[rate*4:]))
+  self.assertTrue(np.array_equal(source,original));self.assertTrue(np.isfinite(wave).all())
+  self.assertLess(np.abs(wave).max(),1.)
+  self.assertEqual(dsp.snapshot()['curve_reaction']['source'],'known replay route event')
+
 
 if __name__=='__main__':unittest.main()

roadscore/prototype/test_demo_curve_plan.py

@@ -0,0 +1,53 @@
+import copy
+import json
+import hashlib
+import tempfile
+import unittest
+from pathlib import Path
+from demo_curve_plan import ReplayCurvePlan,launch_plan,validate
+from score_archive import staging_provenance
+
+PLAN={'version':1,'demo_only':True,'route':'synthetic-route','replay_start':149,
+      'curves':[{'build_start':10.,'apex':24.,'end':25.5,'evidence':'Synthetic measured curve fixture'}]}
+
+
+class PlanTests(unittest.TestCase):
+ def test_exact_route_start_and_fresh_native_demo_only(self):
+  with tempfile.TemporaryDirectory() as directory:
+   path=Path(directory)/'plan.json';path.write_text(json.dumps(PLAN))
+   options=dict(native=True,replay=False,judging=False,policy='conservative-v4')
+   self.assertEqual(launch_plan(path,'synthetic-route',149,**options),PLAN)
+   for changes in ({'native':False},{'replay':True},{'judging':True},{'policy':'conservative-v3'}):
+    self.assertIsNone(launch_plan(path,'synthetic-route',149,**(options|changes)))
+   self.assertIsNone(launch_plan(path,'another-route',149,**options))
+   self.assertIsNone(launch_plan(path,'synthetic-route',150,**options))
+ def test_no_live_or_judging_plan_and_no_input_mutation(self):
+  env={'ROADSCORE_DEMO_CURVE_PLAN':json.dumps(PLAN),'ROADSCORE_PRESENTATION_POLICY':'conservative-v4'}
+  self.assertIsNone(ReplayCurvePlan.from_environment('live',env))
+  self.assertIsNone(ReplayCurvePlan.from_environment('replay',env|{'ROADSCORE_SEED_ORIGIN':'judging-route'}))
+  plan=ReplayCurvePlan.from_environment('replay',env)
+  source={'kind':'curve','phase':'neutral','activation':None,'amount':0.}
+  self.assertIs(plan.state(15,'other-route',source),source)
+  self.assertIs(plan.state(9,'synthetic-route',source),source)
+  stages=[plan.state(t,'synthetic-route',source) for t in (10,17,23,24,25)]
+  self.assertEqual(source,{'kind':'curve','phase':'neutral','activation':None,'amount':0.})
+  self.assertEqual([s['phase'] for s in stages],['anticipation']*3+['event']*2)
+  self.assertEqual(stages[1]['amount'],.5)
+  self.assertTrue(all(s['demo_staged_curve'] for s in stages))
+  self.assertIs(plan.state(26,'synthetic-route',source),source)
+ def test_invalid_or_unbounded_timeline_is_rejected(self):
+  for changes in ({'apex':float('nan')},{'build_start':-1},{'apex':100},{'end':24.1},{'evidence':''}):
+   candidate=copy.deepcopy(PLAN);candidate['curves'][0].update(changes)
+   with self.assertRaises(ValueError):validate(candidate)
+  candidate=copy.deepcopy(PLAN);candidate['curves'].append(candidate['curves'][0].copy())
+  with self.assertRaises(ValueError):validate(candidate)
+ def test_archive_keeps_staging_disclosure_and_exact_plan_hash(self):
+  with tempfile.TemporaryDirectory() as directory:
+   self.assertEqual(staging_provenance(directory),{})
+   path=Path(directory)/'demo_curve_plan.json';path.write_text(json.dumps(PLAN))
+   meta=staging_provenance(directory)
+   self.assertTrue(meta['presentation_demo_only'])
+   self.assertEqual(meta['demo_curve_plan_sha256'],hashlib.sha256(path.read_bytes()).hexdigest())
+
+
+if __name__=='__main__':unittest.main()