Add session-bound replay-only musical engagement controls

starpilot/system/the_galaxy/assets/mobile/js/views/RoadScore.js

@@ -135,6 +135,16 @@ export const RoadScore = {
         <p class="gx-row__desc">{{status.live?.reason || 'Live driving readiness has not been confirmed.'}}</p>
         <button class="gx-btn" @click="setLive(false)">Stop RoadScore</button>
       </div></GalaxySection>
+      <GalaxySection title="Replay simulation" icon="bi-play-circle" :collapsible="false"><div style="padding:16px">
+        <p class="gx-row__desc">Music presentation only. These buttons do not engage or control the vehicle.</p>
+        <p><strong>{{status.demo?.available ? (status.demo.mode === 'engaged' ? 'Simulated engage' : status.demo.mode === 'disengaged' ? 'Simulated disengage' : 'Using recorded state') : 'Replay unavailable'}}</strong></p>
+        <div style="display:flex;gap:8px;flex-wrap:wrap">
+          <button class="gx-btn" :disabled="busy || !status.demo?.available" @click="action('demo_engagement', {session_id:status.demo.session_id, mode:'engaged'})">Simulate engage</button>
+          <button class="gx-btn" :disabled="busy || !status.demo?.available" @click="action('demo_engagement', {session_id:status.demo.session_id, mode:'disengaged'})">Simulate disengage</button>
+          <button class="gx-btn" :disabled="busy || !status.demo?.available" @click="action('demo_engagement', {session_id:status.demo.session_id, mode:'recorded'})">Use recorded state</button>
+        </div>
+        <p v-if="!status.demo?.available" class="gx-row__desc">{{status.demo?.reason}}</p>
+      </div></GalaxySection>
       <GalaxySection title="Composer" icon="bi-music-note-beamed" :collapsible="false"><div style="padding:16px">
         <div class="gx-row"><div><strong>{{ status.state || 'UNAVAILABLE' }}</strong><div class="gx-row__desc">{{ status.composer ? status.composer.toUpperCase() : 'Composer not connected' }}{{ status.backend ? ' · ' + status.backend : '' }}{{ status.composer && status.profile ? ' · ' + status.profile.toUpperCase() : '' }}</div></div></div>
         <div class="gx-row"><label for="roadscore-style">Next style</label><select class="gx-field" id="roadscore-style" v-model="profile" :disabled="busy || !status.can_edit" @change="action('settings', {profile})"><option v-for="p in status.profiles || []" :value="p.id">{{p.name}}</option></select></div>

starpilot/system/the_galaxy/roadscore.py

@@ -1,5 +1,7 @@
 """Galaxy's local-only RoadScore operator API. Never discovers or contacts a bench."""
 import json
+import math
+import tempfile
 import os
 from pathlib import Path
 import importlib.util
@@ -141,6 +143,33 @@ class Operator:
         self.output_owner = module.OutputOwner(self.root, self.offroad)
     return self.output_owner.dispatch(action, **payload)
 
+  def demo_status(self):
+    state=read_json(self.root/'results/current/status.json')
+    stamp=state.get('command_wall');session=state.get('presentation_session_id')
+    fresh=type(stamp) in (int,float) and math.isfinite(stamp) and 0<=time.monotonic()-stamp<=2
+    available=bool(fresh and state.get('engagement_presentation',{}).get('enabled') is True and state.get('input_mode')=='replay' and state.get('route') not in (None,'','live') and isinstance(session,str) and session)
+    mode=state.get('demo_engagement_mode','recorded')
+    return {'available':available,'session_id':session if available else None,'mode':mode if mode in ('recorded','engaged','disengaged') else 'recorded',
+            'reason':'' if available else 'Start a RoadScore replay to simulate musical engagement.'}
+
+  def demo_engagement(self,data):
+    if not isinstance(data,dict) or set(data)!={'session_id','mode'} or data['mode'] not in ('recorded','engaged','disengaged'):
+      raise ValueError('Invalid replay simulation command')
+    with self.lock:
+      status=self.demo_status()
+      if not status['available'] or data['session_id']!=status['session_id']:
+        raise ValueError('A fresh matching RoadScore replay is required; live driving cannot be simulated')
+      run=(self.root/'results/current').resolve(strict=True)
+      command={'version':1,'session_id':status['session_id'],'mode':data['mode'],'created_wall':time.monotonic()}
+      temporary=None
+      try:
+        with tempfile.NamedTemporaryFile(mode='w',dir=run,prefix='.demo-engagement-',delete=False) as stream:
+          temporary=Path(stream.name);json.dump(command,stream)
+        temporary.replace(run/'demo_engagement.json')
+      finally:
+        if temporary is not None:temporary.unlink(missing_ok=True)
+      return {'requested_mode':data['mode'],'demo':self.demo_status()}
+
   def status(self, offroad):
     settings = read_json(self.root / 'generated/operator_settings.json')
     worker=current_worker(self.root)
@@ -156,7 +185,9 @@ class Operator:
     if live and (self.root/'generated/busy').exists():state='GENERATING'
     if output.get('state') in STATES:
       state = output['state']
-    return dict(available=self.device and self.root.exists(), state=state, profiles=PROFILES, live=self.live_status(),
+    demo=self.demo_status()
+    if not offroad:demo.update(available=False,reason='Replay simulation requires the vehicle to be offroad.')
+    return dict(available=self.device and self.root.exists(), state=state, profiles=PROFILES, live=self.live_status(), demo=demo,
                 profile=worker.get('profile') if live else settings.get('profile', 'prism'),
                 selected_profile=settings.get('profile', 'prism'),
                 composer='ace' if live else None, backend='Chestnut' if live else None,
@@ -168,6 +199,9 @@ class Operator:
                 calibrating=output.get('calibrating', False), session_muted=output.get('session_muted', True), output=output.get('output'), latency_ms=output.get('latency_ms'), error=self.error or output.get('error'))
 
   def operate(self, action, data, offroad):
+    if action == 'demo_engagement':
+      if not offroad:raise ValueError('Replay simulation requires the vehicle to be offroad')
+      return self.demo_engagement(data)
     if action == 'live':
       if set(data) != {'enabled'} or type(data['enabled']) is not bool:
         raise ValueError('Live RoadScore requires an enabled boolean')

starpilot/system/the_galaxy/tests/test_roadscore_demo.py

@@ -0,0 +1,40 @@
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import time
+import unittest
+
+spec=importlib.util.spec_from_file_location('galaxy_demo_operator',Path(__file__).resolve().parents[1]/'roadscore.py')
+module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+
+class DemoTests(unittest.TestCase):
+ def setUp(self):
+  self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
+  self.root=Path(self.temp.name);self.run=self.root/'results/current';self.run.mkdir(parents=True)
+  self.operator=module.Operator(root=self.root,device=True)
+  self.state={'route':'private-route','input_mode':'replay','command_wall':time.monotonic(),'presentation_session_id':'session-a','engagement_presentation':{'enabled':True}}
+  self.write()
+ def write(self): (self.run/'status.json').write_text(json.dumps(self.state))
+ def command(self,mode='engaged',session='session-a',offroad=True):
+  return self.operator.operate('demo_engagement',{'mode':mode,'session_id':session},offroad)
+ def test_modes_write_only_session_command(self):
+  for mode in ('engaged','disengaged','recorded'):
+   result=self.command(mode);value=json.loads((self.run/'demo_engagement.json').read_text())
+   self.assertEqual(value['version'],1);self.assertEqual(value['mode'],mode);self.assertEqual(value['session_id'],'session-a');self.assertEqual(result['requested_mode'],mode)
+   self.assertLessEqual(value['created_wall'],time.monotonic())
+  self.assertEqual({p.name for p in self.run.iterdir()},{'status.json','demo_engagement.json'})
+ def test_live_stale_future_unknown_and_disabled_rejected(self):
+  original=dict(self.state)
+  for patch in ({'input_mode':'live'},{'route':'live'},{'input_mode':None},{'command_wall':time.monotonic()-3},{'command_wall':time.monotonic()+10},{'command_wall':float('nan')},{'engagement_presentation':{'enabled':False}},{'presentation_session_id':None}):
+   self.state={**original,**patch};self.write()
+   with self.assertRaises(ValueError):self.command()
+  self.assertFalse((self.run/'demo_engagement.json').exists())
+ def test_session_onroad_and_invalid_commands_rejected(self):
+  for kwargs in ({'session':'old-session'},{'offroad':False},{'mode':'controlsAllowed'}):
+   with self.assertRaises(ValueError):self.command(**kwargs)
+  for payload in ({'mode':'engaged'}, {'mode':'engaged','session_id':'session-a','pid':1}):
+   with self.assertRaises(ValueError):self.operator.operate('demo_engagement',payload,True)
+  self.assertFalse((self.run/'demo_engagement.json').exists())
+
+if __name__=='__main__':unittest.main()