Add sparse beat-aligned alert accents and assess short openings

roadscore/prototype/alert_accent.py

@@ -0,0 +1,47 @@
+"""Sparse native-alert accents using the signal shaker's percussion vocabulary."""
+import math
+import numpy as np
+from signal_shaker import SignalShaker
+
+
+class AlertAccent:
+ def __init__(self, grid, rate=48000, enabled=False):
+  self.grid=grid;self.rate=rate;self.enabled=enabled
+  self.grain=SignalShaker(grid,rate,peak=.016).grain
+  self.key=None;self.since=0;self.handled=False;self.pending=[]
+  self.last=-60*rate;self.events=[];self.rendered_active=False
+
+ def process(self, pcm, start, key, meaningful, fresh, competing=False):
+  self.rendered_active=False
+  if not self.enabled or not self.grid.usable:return pcm
+  current=key if meaningful and fresh else None
+  if current!=self.key:
+   self.key=current;self.since=start;self.handled=False;self.pending=[]
+  if not fresh or competing:
+   self.pending=[]
+   if current:self.handled=True
+   return pcm
+  if current and not self.handled and start-self.since>=round(.2*self.rate):
+   self.handled=True
+   if start-self.last>=12*self.rate and sum(start-e['frame']<60*self.rate for e in self.events)<3:
+    beat=self.rate*60/self.grid.bpm;origin=self.grid.beat_phase*self.rate
+    frame=round(origin+math.ceil((start-origin)/beat-1e-10)*beat)
+    self.pending=[(frame,1.),(frame+round(beat/2),.55)]
+    self.last=start;self.events.append({'kind':'native_alert_accent','frame':start,'scheduled_frame':frame,'alert':current})
+  end=start+len(pcm);overlay=None;remaining=[]
+  for frame,gain in self.pending:
+   left=max(start,frame);right=min(end,frame+len(self.grain))
+   if right>left:
+    if overlay is None:overlay=np.zeros_like(pcm)
+    overlay[left-start:right-start]+=self.grain[left-frame:right-frame]*gain
+   if frame+len(self.grain)>end:remaining.append((frame,gain))
+  self.pending=remaining
+  # Never create an isolated notification sound in a musical rest.
+  if overlay is None or np.max(np.abs(pcm),initial=0)<.002:return pcm
+  self.rendered_active=bool(np.any(overlay))
+  return pcm+overlay
+
+ def snapshot(self):
+  return {'enabled':self.enabled,'rhythm_enabled':self.enabled and self.grid.usable,
+          'rendered_active':self.rendered_active,'minimum_spacing_seconds':12,
+          'maximum_per_minute':3,'source':'selfdriveState.alertStatus'}

roadscore/prototype/app.py

@@ -16,6 +16,7 @@ from presentation_policy import effective_config,selected as presentation_policy
 from engagement_presentation import EngagementPresentation,PresentationConfig,engagement_active
 from signal_shaker import SignalShaker,assess_grid,profile_tempo_prior
 from core_apex import CoreApex
+from alert_accent import AlertAccent
 from presentation_status import export_status
 from rolling import anchor_options,INITIAL_END,WINDOW,LATENT_SECONDS,trajectory
 from musical import Arrival,MusicalDSP,analyze_music,ending_gesture,match_continuation
@@ -59,9 +60,10 @@ drive_events=config.get('drive_events',False) or config.get('event_music',False)
 presentation_config=PresentationConfig.read(config.get('engagement_presentation'))
 shaker_enabled=config.get('signal_shaker',{}).get('enabled') is True
 apex_enabled=config.get('core_apex',{}).get('enabled') is True
-if (presentation_config.enabled or shaker_enabled or apex_enabled) and render_mode!='gold-core':raise ValueError('Optional presentation layers require gold-core baseline')
+alert_enabled=config.get('alert_accent',{}).get('enabled') is True
+if (presentation_config.enabled or shaker_enabled or apex_enabled or alert_enabled) and render_mode!='gold-core':raise ValueError('Optional presentation layers require gold-core baseline')
 presentation=EngagementPresentation(rate,block) if presentation_config.enabled else None
-engagement=(False,False,0,0.,0);signal_state=(False,False,0,0.,0)
+engagement=(False,False,0,0.,0);signal_state=(False,False,0,0.,0);alert_state=(None,False)
 initial_identity=identity;musical_mode=config.get('musical',False);arrival=Arrival();ending_start=None;ending_audio=None;ending_info={};playing_identity=identity;identity_queue=[];last_requested_identity=identity
 usable_frames=WINDOW if rolling_mode else (301 if musical_mode else 323)
 source_path=root/f'assets/source_{identity}.wav' if identity!='legacy' else root/'assets/source.wav'
@@ -99,12 +101,13 @@ if config.get('gesture_layer',False) and render_mode=='current':
  gesture_grid=grid(gesture_source,rate,gesture_pulse['bpm'] if gesture_pulse['confidence']>=.25 else music_info['bpm'])
  gestures=MusicalGestures(gesture_source,rate,gesture_grid['bpm'],gesture_grid['beat_phase'])
  (run/'gesture_grid.json').write_text(json.dumps(gesture_grid))
-shaker=None;apex=None
-if shaker_enabled or apex_enabled:
+shaker=None;apex=None;alert_accent=None
+if shaker_enabled or apex_enabled or alert_enabled:
  profile_manifest=json.loads((root/'experiments/ace_chestnut_20260916/profiles'/ace_profile/'profile.json').read_text())
  shaker_grid,shaker_analysis=assess_grid(source,rate,profile_tempo_prior(profile_manifest))
  shaker=SignalShaker(shaker_grid,rate,enabled=shaker_enabled)
- apex=CoreApex(shaker_grid,rate,enabled=apex_enabled)
+ apex=CoreApex(shaker_grid,rate,enabled=apex_enabled,dip_db=config.get('core_apex',{}).get('dip_db',-1.))
+ alert_accent=AlertAccent(shaker_grid,rate,enabled=alert_enabled)
  (run/'shaker_grid.json').write_text(json.dumps({'grid':shaker.snapshot(),'analysis':shaker_analysis},indent=2))
 if config.get('composition_control',False):
  from composition_policy import CompositionPolicy
@@ -173,6 +176,10 @@ def callback(out,n,ti,status):
  signal_on,signal_fresh=engagement_active(signal_state[1],signal_state[0],signal_state[2],signal_state[4],signal_state[3],callback_wall)
  if shaker is not None:rendered=shaker.process(rendered,frames-n,signal_on,signal_fresh)
  if apex is not None:rendered=apex.process(rendered,frames-n,event_state)
+ if alert_accent is not None:
+  alert_key,meaningful=alert_state
+  competing=bool((shaker is not None and shaker.active) or (apex is not None and apex.rendered_active) or (presentation is not None and abs(presentation.mix-float(active))>.01))
+  rendered=alert_accent.process(rendered,frames-n,alert_key,meaningful,fresh,competing)
  if presentation is not None:rendered=presentation.process(rendered,active,presentation_config)
  out[:]=0 if a.mute else rendered
  try:capture.put_nowait((rendered,chunk,{'signal_shaker_enabled':shaker_enabled,'signal_on':signal_on,'signal_fresh':signal_fresh,'engagement_presentation_enabled':presentation_config.enabled,'engagement_active':active,'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}))
@@ -207,6 +214,7 @@ try:
    sm.update(100)
    control_wall=time.monotonic();latest_source=max(sm.logMonoTime.values())
    if sm.updated['selfdriveState']:
+    alert_state=(str(sm['selfdriveState'].alertType),int(sm['selfdriveState'].alertStatus.raw)>0 and int(sm['selfdriveState'].alertSize.raw)>0)
     engagement=(bool(sm['selfdriveState'].active),bool(sm.valid['selfdriveState']),sm.logMonoTime['selfdriveState'],control_wall,latest_source)
    else:engagement=(*engagement[:4],latest_source)
    if sm.updated['carState']:
@@ -352,6 +360,7 @@ try:
      snapshot.update(presentation.snapshot(presentation_config,engagement_on,engagement_fresh))
     if shaker is not None:snapshot['signal_shaker']=shaker.snapshot()
     if apex is not None:snapshot['core_apex']=apex.snapshot()
+    if alert_accent is not None:snapshot['alert_accent']=alert_accent.snapshot()
     if songform:snapshot.update(songform.snapshot())
     if composition:snapshot.update(composition.snapshot(frames/rate))
     if gestures:snapshot.update(gestures.status())
@@ -359,6 +368,7 @@ try:
    if time.monotonic()-last_progress>15:raise RuntimeError('Replay model input stalled')
 finally:
  audio_started=False
+ if alert_accent is not None:(run/'alert_accent_events.json').write_text(json.dumps(alert_accent.events,indent=2))
  if apex is not None:(run/'core_apex_events.json').write_text(json.dumps(apex.events,indent=2))
  if shaker is not None:(run/'shaker_events.json').write_text(json.dumps({'sequences':shaker.events,'pulse_frames':shaker.pulse_frames,'settings':shaker.snapshot()},indent=2))
  if gestures:(run/'gestures.json').write_text(json.dumps({'events':gestures.events,'bpm':gestures.bpm,'source_derived':True,'tonal_key_not_inferred':True},indent=2))

roadscore/prototype/core_apex.py

@@ -3,8 +3,8 @@ import math
 import numpy as np
 
 class CoreApex:
- def __init__(self, grid, rate=48000, enabled=False):
-  self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.events=[];self.rendered_active=False;self.rendered_end=0.
+ def __init__(self, grid, rate=48000, enabled=False,dip_db=-1.):
+  self.dip_db=max(-3.,min(0.,float(dip_db)));self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.events=[];self.rendered_active=False;self.rendered_end=0.
  def process(self, pcm, start_frame, state):
   self.rendered_active=False;self.rendered_end=(start_frame+len(pcm))/self.rate
   if not self.enabled or not self.grid.usable:return pcm
@@ -15,12 +15,12 @@ class CoreApex:
    if (not self.events or now-self.events[-1]['trigger_seconds']>=8) and sum(now-e['trigger_seconds']<60 for e in self.events)<3:
     subdivision=60/self.grid.bpm/2
     target=self.grid.beat_phase+round((now+lead-self.grid.beat_phase)/subdivision)*subdivision
-    self.peak=round(target*self.rate);self.events.append({'trigger_seconds':now,'peak_seconds':target,'activation':activation,'dip_db':-1.,'peak_gain':1.})
+    self.peak=round(target*self.rate);self.events.append({'trigger_seconds':now,'peak_seconds':target,'activation':activation,'dip_db':self.dip_db,'peak_gain':1.})
   if self.peak is None:return pcm
   offset=(np.arange(start_frame,start_frame+len(pcm))-self.peak)/self.rate
   # Half-second breath, then original unity source at the estimated apex. Never boosts/clips the hook.
   gain=np.ones(len(pcm),np.float32);inside=(offset>-.7)&(offset<0)
-  gain[inside]=1-(1-10**(-1/20))*np.sin(np.pi*(offset[inside]+.7)/.7)**2
+  gain[inside]=1-(1-10**(self.dip_db/20))*np.sin(np.pi*(offset[inside]+.7)/.7)**2
   self.rendered_active=bool(inside.any() and np.any(pcm[inside]))
   return pcm if not self.rendered_active else pcm*gain[:,None]
  def snapshot(self):

roadscore/prototype/presentation_policy.py

@@ -2,7 +2,7 @@
 import copy
 import os
 
-POLICIES = ('conservative-v1', 'off', 'frozen')
+POLICIES = ('conservative-v1', 'conservative-v2', 'off', 'frozen')
 CONSERVATIVE = {
  'presentation_policy_version': 'conservative-v1',
  'signal_shaker': {'enabled': True},
@@ -10,6 +10,8 @@ CONSERVATIVE = {
  'engagement_presentation': {'version': 2, 'enabled': True, 'attack_ms': 220, 'release_ms': 650},
 }
 
+CONSERVATIVE_V2 = {**copy.deepcopy(CONSERVATIVE), 'presentation_policy_version':'conservative-v2', 'alert_accent':{'enabled':True}, 'core_apex':{'enabled':True,'dip_db':-3.}}
+
 def select_launch(composer, profile, replay=False, judging=False, render_mode=None, policy=None):
  if policy is not None and policy not in POLICIES:raise ValueError('Unknown presentation policy')
  if render_mode is not None and render_mode not in ('current','gold-core'):raise ValueError('Unknown rendering mode')
@@ -25,8 +27,8 @@ def select_launch(composer, profile, replay=False, judging=False, render_mode=No
   return {'render_mode':'current','policy':'frozen'}
  normal_prism=composer=='ace' and profile=='prism'
  mode=render_mode or ('gold-core' if normal_prism else 'current')
- chosen=policy or ('conservative-v1' if normal_prism and mode=='gold-core' else 'off')
- if chosen=='conservative-v1' and (composer!='ace' or mode!='gold-core'):
+ chosen=policy or ('conservative-v2' if normal_prism and mode=='gold-core' else 'off')
+ if chosen in ('conservative-v1','conservative-v2') and (composer!='ace' or mode!='gold-core'):
   raise ValueError('Conservative presentation requires ACE gold-core rendering')
  return {'render_mode':mode,'policy':chosen}
 
@@ -39,7 +41,9 @@ def effective_config(config,environ=None):
  """Return a copy; never mutate the global runtime file or an archived policy."""
  result=copy.deepcopy(config);policy=selected(environ)
  if policy=='frozen':return result
- if policy=='conservative-v1':result.update(copy.deepcopy(CONSERVATIVE))
+ if policy in ('conservative-v1','conservative-v2'):
+  result.update(copy.deepcopy(CONSERVATIVE_V2 if policy=='conservative-v2' else CONSERVATIVE))
+  if policy=='conservative-v1':result['alert_accent']={'enabled':False}
  else:
-  result.update(presentation_policy_version='off',signal_shaker={'enabled':False},core_apex={'enabled':False},engagement_presentation={'version':2,'enabled':False})
+  result.update(presentation_policy_version='off',alert_accent={'enabled':False},signal_shaker={'enabled':False},core_apex={'enabled':False},engagement_presentation={'version':2,'enabled':False})
  return result

roadscore/prototype/presentation_status.py

@@ -1,5 +1,5 @@
 """Status passed with PCM; preserve actual rendered state and its audio clock."""
-EXPORT_FIELDS = ['composer','job_inflight','generation_elapsed_seconds','readiness','profile','safe_extensions','quality_failures','holding_accepted_music','identity','style','playing_identity','phase','lead','buffered','arrival_at','worker_failed','section','next_section','scheduled','gesture_active','gesture_queued','turn_signal_music','outro_heard_seconds','form_labels_are_intent'] + ['render_mode','presentation_policy','elapsed','signal_shaker','core_apex','engagement_presentation']
+EXPORT_FIELDS = ['composer','job_inflight','generation_elapsed_seconds','readiness','profile','safe_extensions','quality_failures','holding_accepted_music','identity','style','playing_identity','phase','lead','buffered','arrival_at','worker_failed','section','next_section','scheduled','gesture_active','gesture_queued','turn_signal_music','outro_heard_seconds','form_labels_are_intent'] + ['render_mode','presentation_policy','elapsed','alert_accent','signal_shaker','core_apex','engagement_presentation']
 
 def export_status(snapshot):
     return {key: snapshot.get(key) for key in EXPORT_FIELDS}

roadscore/prototype/render_presentation_auditions.py

@@ -6,6 +6,7 @@ import numpy as np
 import soundfile as sf
 from signal_shaker import SignalShaker,assess_grid
 from core_apex import CoreApex
+from alert_accent import AlertAccent
 from engagement_presentation import EngagementPresentation,PresentationConfig
 
 def main():
@@ -15,7 +16,8 @@ def main():
  # Explicit audition hypothesis only: never written into runtime configuration.
  audition=replace(estimated,bpm=128.,beat_phase=estimated.beat_phase,confirmed=True,reason='OFFLINE PROVISIONAL128BPM hypothesis; NOT human verified or runtime authorized')
  shaker=SignalShaker(audition,rate,True);safe=SignalShaker(estimated,rate,True);apex=CoreApex(audition,rate,True);engage=EngagementPresentation(rate)
- arms={k:[] for k in ['core','shaker_provisional128','shaker_confidence_gated','engagement_separate','apex_separate_provisional128']};timings=[]
+ alerts=AlertAccent(estimated,rate,True)
+ arms={k:[] for k in ['core','alerts_confidence_gated','shaker_provisional128','shaker_confidence_gated','engagement_separate','apex_separate_provisional128']};timings=[]
  for frame in range(0,len(wave),4800):
   x=wave[frame:frame+4800];t=frame/rate
   signal=any(start<=t<end for start,end in [(8.,14.),(32.,38.)]) and int(t*2)%2==0
@@ -24,11 +26,11 @@ def main():
   target=22. if t<28 else 46.;lead=target-t
   state={'kind':'curve','phase':'anticipation' if .0<lead<2 else 'neutral','lead':lead,'activation':target}
   started=time.perf_counter()
-  arms['core'].append(x);arms['shaker_provisional128'].append(shaker.process(x,frame,signal,True));arms['shaker_confidence_gated'].append(safe.process(x,frame,signal,True));arms['engagement_separate'].append(engage.process(x,active,PresentationConfig(enabled=True)));arms['apex_separate_provisional128'].append(apex.process(x,frame,state));timings.append(time.perf_counter()-started)
+  arms['core'].append(x);arms['alerts_confidence_gated'].append(alerts.process(x,frame,'synthetic_user_prompt',3<=t<6,True));arms['shaker_provisional128'].append(shaker.process(x,frame,signal,True));arms['shaker_confidence_gated'].append(safe.process(x,frame,signal,True));arms['engagement_separate'].append(engage.process(x,active,PresentationConfig(enabled=True)));arms['apex_separate_provisional128'].append(apex.process(x,frame,state));timings.append(time.perf_counter()-started)
  records={}
  for name,blocks in arms.items():
   output=np.concatenate(blocks);path=a.output/(name+'.wav');sf.write(path,output,rate,subtype='FLOAT');records[name]={'sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'peak':float(abs(output).max()),'difference_rms':float(np.sqrt(np.mean((output-wave)**2)))}
  records['core']['sample_identical_to_source']=bool(np.array_equal(sf.read(a.output/'core.wav',dtype='float32')[0],wave))
- report={'source':str(a.source),'source_sha256':hashlib.sha256(a.source.read_bytes()).hexdigest(),'seconds':len(wave)/rate,'common_gain':'source unchanged; no normalization, limiter, secondgain or resampling','events_source':'SYNTHETIC OFFLINE timeline; not route telemetry','synthetic_signal_windows_seconds':[[8,14],[32,38]],'synthetic_engaged_windows_seconds':[[16,28],[42,56]],'synthetic_apex_seconds':[22,46],'estimated_grid':asdict(estimated),'grid_analysis':analysis,'provisional_audition_grid':asdict(audition),'shaker':shaker.snapshot(),'shaker_sequence_events':shaker.events,'shaker_pulse_frames':shaker.pulse_frames,'apex_events':apex.events,'engagement_config':asdict(PresentationConfig(enabled=True)),'callback_work_allarms_mean_ms':float(np.mean(timings)*1000),'callback_work_allarms_max_ms':max(timings)*1000,'timing_scope':'Mac offline sequential5arms, not native callback deadline validation','files':records,'hardware_validated':False}
+ report={'source':str(a.source),'source_sha256':hashlib.sha256(a.source.read_bytes()).hexdigest(),'seconds':len(wave)/rate,'common_gain':'source unchanged; no normalization, limiter, secondgain or resampling','events_source':'SYNTHETIC OFFLINE timeline; not route telemetry','synthetic_signal_windows_seconds':[[8,14],[32,38]],'synthetic_engaged_windows_seconds':[[16,28],[42,56]],'synthetic_apex_seconds':[22,46],'estimated_grid':asdict(estimated),'grid_analysis':analysis,'provisional_audition_grid':asdict(audition),'shaker':shaker.snapshot(),'shaker_sequence_events':shaker.events,'shaker_pulse_frames':shaker.pulse_frames,'apex_events':apex.events,'engagement_config':asdict(PresentationConfig(enabled=True)),'callback_work_allarms_mean_ms':float(np.mean(timings)*1000),'callback_work_allarms_max_ms':max(timings)*1000,'timing_scope':'Mac offline sequential5arms, not native callback deadline validation','files':records,'alert_events':alerts.events,'hardware_validated':False}
  (a.output/'report.private.json').write_text(json.dumps(report,indent=2));print(json.dumps({'output':str(a.output),'grid':asdict(estimated),'files':records},indent=2))
 if __name__=='__main__':main()

roadscore/prototype/signal_shaker.py

@@ -25,13 +25,14 @@ def profile_tempo_prior(manifest):
 
 def assess_grid(wave, rate, bpm_prior=None):
  """Assess once outside callback. Conflicting acoustic tempo wins a veto, not a guessed beat."""
- if len(wave)<30*rate:return ShakerGrid(128.,0.,0.,0.,False,'insufficient audio for grid'),{}
- g=estimate_grid(wave[:30*rate],rate,bpm_prior)
+ if len(wave)<12*rate:return ShakerGrid(128.,0.,0.,0.,False,'insufficient audio for grid'),{}
+ wave=wave[:30*rate];half=len(wave)//2;half_seconds=half/rate
+ g=estimate_grid(wave,rate,bpm_prior)
  best=max(x['autocorrelation'] for x in g['tempo_candidates'])
  agreement=g['pulse_confidence']/max(best,1e-9)
  # Bar accent confidence is not beat-phase confidence. Compare phase of two halves.
- a=estimate_grid(wave[:15*rate],rate,g['bpm']);b=estimate_grid(wave[15*rate:30*rate],rate,g['bpm'])
- period=60/g['bpm'];drift=abs(((b['beat_phase']+15-a['beat_phase']+period/2)%period)-period/2)
+ a=estimate_grid(wave[:half],rate,g['bpm']);b=estimate_grid(wave[half:],rate,g['bpm'])
+ period=60/g['bpm'];drift=abs(((b['beat_phase']+half_seconds-a['beat_phase']+period/2)%period)-period/2)
  phase=max(0.,1.-drift/(period*.25))
  coherent=bpm_prior is not None and agreement>=.85 and abs(g['bpm']/bpm_prior-1)<.08 and abs(a['bpm']/b['bpm']-1)<.03
  reason='coherent estimated pulse; downbeat unverified' if coherent else 'tempo/phase ambiguity; rhythmic additions bypassed'

roadscore/prototype/test_alert_accent.py

@@ -0,0 +1,42 @@
+import unittest
+import numpy as np
+from alert_accent import AlertAccent
+from signal_shaker import ShakerGrid, assess_grid
+
+GRID=ShakerGrid(120,.125,.8,.9,True,'test')
+
+class AccentTests(unittest.TestCase):
+ def render(self, alerts, fresh=True, competing=False, grid=GRID):
+  accent=AlertAccent(grid,rate=8000,enabled=True)
+  source=np.full((800,2),.1,np.float32);heard=False
+  for i,key in enumerate(alerts):
+   wet=accent.process(source,i*800,key,bool(key),fresh,competing)
+   heard |= bool(np.any(wet!=source))
+   self.assertLessEqual(float(np.max(np.abs(wet-source))),.016001)
+  return accent,heard
+ def test_persistent_alert_once_and_grid_aligned(self):
+  accent,heard=self.render(['takeover']*200)
+  self.assertTrue(heard);self.assertEqual(len(accent.events),1)
+  self.assertEqual(accent.events[0]['scheduled_frame'],5000)
+ def test_tiny_changes_and_competing_cues_are_omitted(self):
+  self.assertFalse(self.render(['a','b']*100)[1])
+  self.assertFalse(self.render(['a']*100,competing=True)[1])
+  self.assertFalse(self.render(['a']*100,fresh=False)[1])
+ def test_density_limit(self):
+  accent,_=self.render([str(i//30) for i in range(600)])
+  self.assertLessEqual(len(accent.events),3)
+ def test_ambiguous_grid_bypasses(self):
+  grid=ShakerGrid(120,0,0,0,False,'uncertain')
+  self.assertFalse(self.render(['a']*100,grid=grid)[1])
+ def test_cancel_pending_on_stale_input(self):
+  a=AlertAccent(GRID,rate=8000,enabled=True);x=np.full((800,2),.1,np.float32)
+  for i in range(3):a.process(x,i*800,'a',True,True)
+  self.assertTrue(a.pending)
+  self.assertIs(a.process(x,2400,'a',True,False),x);self.assertFalse(a.pending)
+ def test_short_opening_assessed_without_assuming_confidence(self):
+  # Real estimator must accept a 23s opening; silence must still veto cues.
+  grid,_=assess_grid(np.zeros((23*8000,2),np.float32),8000,128)
+  self.assertNotEqual(grid.reason,'insufficient audio for grid')
+  self.assertFalse(grid.usable)
+
+if __name__=='__main__':unittest.main()

roadscore/prototype/test_presentation_policy.py

@@ -7,7 +7,7 @@ class PolicyTests(unittest.TestCase):
   fragment=Path(__file__).resolve().parents[1]/'tools/presentation_conservative_v1.json'
   self.assertEqual(json.loads(fragment.read_text()),CONSERVATIVE)
  def test_normal_prism_gets_integrated_policy_without_extra_flags(self):
-  self.assertEqual(select_launch('ace','prism'),{'render_mode':'gold-core','policy':'conservative-v1'})
+  self.assertEqual(select_launch('ace','prism'),{'render_mode':'gold-core','policy':'conservative-v2'})
  def test_explicit_optouts(self):
   self.assertEqual(select_launch('ace','prism',policy='off'),{'render_mode':'gold-core','policy':'off'})
   self.assertEqual(select_launch('ace','prism',render_mode='current'),{'render_mode':'current','policy':'off'})
@@ -27,6 +27,13 @@ class PolicyTests(unittest.TestCase):
   self.assertEqual(effective_config(original,{}),original)
   off=effective_config(original,{'ROADSCORE_PRESENTATION_POLICY':'off'})
   self.assertFalse(off['signal_shaker']['enabled']);self.assertFalse(off['engagement_presentation']['enabled'])
+ def test_v2_keeps_frozen_and_legacy_alerts_off(self):
+  config={'alert_accent':{'enabled':True}}
+  for policy in ('off','conservative-v1'):
+   self.assertFalse(effective_config(config,{'ROADSCORE_PRESENTATION_POLICY':policy})['alert_accent']['enabled'])
+  latest=effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':'conservative-v2'})
+  self.assertTrue(latest['alert_accent']['enabled'])
+  self.assertEqual(latest['core_apex']['dip_db'],-3.)
  def test_other_backend_and_legacy_incompatibility(self):
   self.assertEqual(select_launch('sa3','prism'),{'render_mode':'current','policy':'off'})
   with self.assertRaises(ValueError):select_launch('ace','prism',render_mode='current',policy='conservative-v1')