Validate final prepared clock residual after bounded recovery

roadscore/prototype/mac_showcase.py

@@ -353,7 +353,7 @@ def audio_worker(a):
         if sync_trace is not None:sync_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,prepared_clock=playback_clock.snapshot(),stream_clock_bridge=bridge.snapshot() if bridge else None,callback_errors=errors,clock_errors=clock_errors,muted=a.muted,session_id=session,manual_scope='isolated replay display and presentation only'))
   if errors:raise RuntimeError(errors[0])
-  if playback_clock.snapshot()['max_post_error_seconds']>.05:raise RuntimeError('Prepared audio clock drift remained above 50 ms after recovery')
+  if abs(playback_clock.snapshot()['current_post_error_seconds'])>.05:raise RuntimeError('Prepared audio clock drift remained above 50 ms after recovery')
 
 
 def apply_showcase_config(a, config):

roadscore/prototype/prepared_clock.py

@@ -37,6 +37,7 @@ class PreparedClock:
     self.explicit_seeks=0
     self.max_pre_error=0
     self.max_post_error=0
+    self.current_post_error=0
     self.events=deque(maxlen=64)
 
   @staticmethod
@@ -90,6 +91,7 @@ class PreparedClock:
       if self.fade_done>=self.fade_frames:self.fade_from=None
     self.position+=frames
     post_error=expected_frame-start
+    self.current_post_error=post_error
     self.max_post_error=max(self.max_post_error,abs(post_error))
     return result,{'source_frame':start,'next_source_frame':self.position,'expected_frame':expected_frame,
                    'pre_error_frames':delta,'post_error_frames':post_error,'crossfade_frames_rendered':faded,
@@ -98,6 +100,7 @@ class PreparedClock:
   def snapshot(self):
     return {'next_source_frame':self.position,'corrections':self.corrections,'explicit_seeks':self.explicit_seeks,
             'max_pre_error_seconds':self.max_pre_error/self.rate,'max_post_error_seconds':self.max_post_error/self.rate,
+            'current_post_error_seconds':self.current_post_error/self.rate,
             'crossfade_ms':1000*self.fade_frames/self.rate,'recovery_in_progress':self.fade_from is not None,
             'maximum_backward_recovery_seconds':self.maximum_backward/self.rate,
             'events':list(self.events),'added_delay_samples':0,'resampling':False}

roadscore/prototype/test_replay_sync.py

@@ -147,6 +147,29 @@ class ReplaySyncTests(unittest.TestCase):
     self.assertEqual(context['clock_errors'][0]['portaudio_current_time'],500.04)
     self.assertIn('next_source_frame',context['clock_errors'][0])
 
+  def test_worker_final_residual_rejects_unfinished_bounce_but_accepts_recovery(self):
+    import numpy as np
+    from prepared_clock import PreparedClock
+    # Exercise the real worker's final gate without creating runtime services.
+    path=Path(__file__).with_name('mac_showcase.py')
+    worker=next(node for node in ast.parse(path.read_text()).body if isinstance(node,ast.FunctionDef) and node.name=='audio_worker')
+    final_gate=compile(ast.Module(body=[worker.body[-1]],type_ignores=[]),str(path),'exec')
+    core=np.zeros((480000,2),np.float32)
+    for delta in (-3149,3149):
+      with self.subTest(delta=delta):
+        clock=PreparedClock();clock.render(core,96000,960)
+        previous=clock.position;target=previous+delta
+        clock.render(core,target,960)
+        clock.render(core,previous+960,960)
+        self.assertGreater(abs(clock.snapshot()['current_post_error_seconds']),.05)
+        with self.assertRaisesRegex(RuntimeError,'drift remained'):
+          exec(final_gate,{'playback_clock':clock})
+        clock.render(core,previous+1920,960)
+        clock.render(core,previous+2880,960)
+        self.assertEqual(clock.snapshot()['current_post_error_seconds'],0)
+        self.assertGreater(clock.snapshot()['max_post_error_seconds'],.05)
+        exec(final_gate,{'playback_clock':clock})
+
   def test_cleanup_stops_all_groups_before_reaping_and_survives_race(self):
     import signal
     events=[]