diff --git a/roadscore/prototype/stream_clock_bridge.py b/roadscore/prototype/stream_clock_bridge.py
new file mode 100644
index 0000000000..f29c1b83e8
--- /dev/null
+++ b/roadscore/prototype/stream_clock_bridge.py
@@ -0,0 +1,36 @@
+"""Map PortAudio timestamps to monotonic time once per output stream."""
+import math
+import time
+
+
+class StreamClockBridge:
+  def __init__(self, offset, uncertainty):
+    self.offset = offset
+    self.uncertainty = uncertainty
+
+  @classmethod
+  def measure(cls, stream_time, *, clock=time.monotonic, samples=5):
+    """Call outside the audio callback, while it is still awaiting route input."""
+    candidates = []
+    for _ in range(samples):
+      before = clock()
+      source = stream_time()
+      after = clock()
+      if (not all(type(value) in (int, float) and math.isfinite(value) for value in (before,source,after))
+          or source <= 0 or after < before):
+        raise ValueError('Invalid output stream clock sample')
+      candidates.append((after-before, (before+after)/2-source))
+    if not candidates:raise ValueError('No output stream clock samples')
+    width, offset = min(candidates)
+    if width > .05:raise ValueError('Output stream clock measurement is too uncertain')
+    return cls(offset,width/2)
+
+  def dac_wall(self, output_buffer_dac_time):
+    if type(output_buffer_dac_time) not in (int,float) or not math.isfinite(output_buffer_dac_time):
+      raise ValueError('Invalid output DAC timestamp')
+    return output_buffer_dac_time+self.offset
+
+  def snapshot(self):
+    return {'portaudio_to_monotonic_seconds':self.offset,
+            'measurement_uncertainty_seconds':self.uncertainty,
+            'mapping':'fixed-stream-clock-offset'}
diff --git a/roadscore/prototype/test_stream_clock_bridge.py b/roadscore/prototype/test_stream_clock_bridge.py
new file mode 100644
index 0000000000..a92cfd57cb
--- /dev/null
+++ b/roadscore/prototype/test_stream_clock_bridge.py
@@ -0,0 +1,38 @@
+import unittest
+
+from stream_clock_bridge import StreamClockBridge
+
+
+class StreamClockBridgeTests(unittest.TestCase):
+  def test_uses_narrowest_bracket_and_supports_distinct_clock_origins(self):
+    walls=iter([100.,100.02,101.,101.002,102.,102.01])
+    source=iter([500.01,501.001,502.005])
+    bridge=StreamClockBridge.measure(lambda:next(source),clock=lambda:next(walls),samples=3)
+    self.assertAlmostEqual(bridge.offset,-400.)
+    self.assertAlmostEqual(bridge.uncertainty,.001)
+    self.assertAlmostEqual(bridge.dac_wall(505.02),105.02)
+
+  def test_delayed_callback_then_burst_cannot_shift_stream_clock_mapping(self):
+    # The old conversion added callback-entry delay to the DAC timestamp.
+    # A delayed callback followed by prompt callbacks looked like a seek back.
+    bridge=StreamClockBridge(-400.,.0001)
+    driver_current=[500.,500.02,500.04,500.06]
+    entered=[100.,100.08,100.04,100.06]
+    dac=[stamp+.056 for stamp in driver_current]
+    old=[now+d-current for now,d,current in zip(entered,dac,driver_current)]
+    self.assertLess(old[2],old[1])
+    mapped=[bridge.dac_wall(stamp) for stamp in dac]
+    for a,b in zip(mapped,mapped[1:]):self.assertAlmostEqual(b-a,.02)
+
+  def test_invalid_samples_and_uncertain_bracket_fail_closed(self):
+    for source in (0.,float('nan'),float('inf')):
+      with self.assertRaises(ValueError):StreamClockBridge.measure(lambda:source,clock=lambda:1.,samples=1)
+    walls=iter([1.,1.1])
+    with self.assertRaises(ValueError):StreamClockBridge.measure(lambda:10.,clock=lambda:next(walls),samples=1)
+    with self.assertRaises(ValueError):StreamClockBridge.measure(lambda:10.,samples=0)
+
+  def test_invalid_dac_timestamp_is_rejected(self):
+    with self.assertRaises(ValueError):StreamClockBridge(0.,0.).dac_wall(float('nan'))
+
+
+if __name__=='__main__':unittest.main()