diff --git a/roadscore/prototype/preparation_progress.py b/roadscore/prototype/preparation_progress.py
new file mode 100644
index 0000000000..04a06d868b
--- /dev/null
+++ b/roadscore/prototype/preparation_progress.py
@@ -0,0 +1,22 @@
+"""Read preparation progress only from the current launch's worker evidence."""
+import json
+import math
+from pathlib import Path
+
+
+def current_progress(path, *, seed, profile, launch_wall):
+    path=Path(path)
+    try:
+        if path.stat().st_mtime < launch_wall:return None
+        state=json.loads(path.read_text())
+    except (OSError,ValueError):return None
+    if not isinstance(state,dict) or state.get('generation_seed')!=seed or state.get('profile')!=profile:return None
+    phase=str(state.get('phase','')).upper()
+    if phase not in ('PREPARING','READY'):return None
+    result={'readiness':'PREPARING','preparation_worker_phase':phase,'preparation_worker_pid':state.get('pid')}
+    for source,target in [('accepted_buffer_seconds','buffered'),('initial_buffer_seconds','buffered'),
+                          ('first_accepted_audio_seconds','first_accepted_audio_seconds'),('elapsed_seconds','preparation_elapsed_seconds')]:
+        value=state.get(source)
+        if isinstance(value,(int,float)) and not isinstance(value,bool) and math.isfinite(value) and value>=0:result[target]=value
+    # Worker READY means audio is prepared; the launcher still must start replay.
+    return result
diff --git a/roadscore/prototype/test_preparation_progress.py b/roadscore/prototype/test_preparation_progress.py
new file mode 100644
index 0000000000..8ce1fd45e6
--- /dev/null
+++ b/roadscore/prototype/test_preparation_progress.py
@@ -0,0 +1,26 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from preparation_progress import current_progress
+
+class ProgressTests(unittest.TestCase):
+ def test_only_current_matching_worker_can_update_preparing(self):
+  with tempfile.TemporaryDirectory() as directory:
+   path=Path(directory)/'worker.json'
+   path.write_text(json.dumps({'generation_seed':12,'profile':'prism','phase':'preparing','accepted_buffer_seconds':53,'elapsed_seconds':120}))
+   args=dict(seed=12,profile='prism',launch_wall=path.stat().st_mtime-1)
+   result=current_progress(path,**args)
+   self.assertEqual(result['buffered'],53);self.assertEqual(result['readiness'],'PREPARING')
+   self.assertIsNone(current_progress(path,**{**args,'seed':13}))
+   self.assertIsNone(current_progress(path,**{**args,'profile':'aurora'}))
+   self.assertIsNone(current_progress(path,**{**args,'launch_wall':path.stat().st_mtime+1}))
+ def test_ready_worker_does_not_claim_started_playback(self):
+  with tempfile.TemporaryDirectory() as directory:
+   path=Path(directory)/'worker.json'
+   path.write_text(json.dumps({'generation_seed':12,'profile':'prism','phase':'READY','initial_buffer_seconds':81}))
+   result=current_progress(path,seed=12,profile='prism',launch_wall=0)
+   self.assertEqual(result['readiness'],'PREPARING');self.assertEqual(result['preparation_worker_phase'],'READY')
+   path.write_text('{');self.assertIsNone(current_progress(path,seed=12,profile='prism',launch_wall=0))
+
+if __name__=='__main__':unittest.main()