Require fresh live playback and explicit verified diagnostic promotion

roadscore/prototype/live_controller.py

@@ -27,7 +27,7 @@ class LiveController:
             if status.get('available') is False:return status
             observation,authorization=self.adapter.health_and_authorization()
             reason=blocked_reason(observation,authorization,self.clock(),require_parked=True)
-            return {**status,'available':True,'can_enable':not status.get('enabled',False) and not reason,
+            return {**status,'available':True,'can_enable':(not status.get('enabled',False) or (status.get('diagnostic') and status.get('state')=='READY')) and not reason,
                     'reason':reason or status.get('reason','')}
         except Exception as error:
             return dict(available=False,enabled=None,state='DEGRADED',can_enable=False,reason='Live supervisor status unavailable: '+str(error))

roadscore/prototype/live_owned_processes.py

@@ -86,12 +86,16 @@ class OwnedLiveProcesses:
         env['PYTHONPATH']=':'.join(['/data/openpilot',str(self.root/'prototype'),'/data/roadscore-feasibility/venv/lib/python3.12/site-packages'])
         env['ROADSCORE_LIVE_SESSION_ID']=session_id
         log=(self.folder/'app.log').open('ab');self.logs.append(log)
+        self.app_started=self.clock()
         self.app=self.popen(['/usr/local/venv/bin/python','-u',str(self.root/'prototype/app.py'),'--root',str(self.root),'--input','live']+(['--audible'] if audible else []),cwd='/data/openpilot',env=env,stdout=log,stderr=log,stdin=subprocess.DEVNULL,start_new_session=True)
 
     def health(self):
+        status=read(self.root/'results/current/status.json')
+        live_ready=(status.get('route')=='live' and isinstance(status.get('command_wall'),(int,float))
+                    and 0<=self.clock()-status['command_wall']<=1. and status.get('elapsed',0)>0)
         return dict(worker_healthy=self.worker is not None and self.worker.poll() is None,
                     accepted_ready=self.accepted_ready(),app_healthy=self.app is None or self.app.poll() is None,
-                    app_ready=self.app is not None and self.app.poll() is None and (self.root/'results/current/ready').exists())
+                    app_ready=live_ready and self.app is not None and self.app.poll() is None and (self.root/'results/current/ready').exists())
 
     def signal_stop(self):
         self.stop_requested.set()

roadscore/prototype/live_supervisor_README.md

@@ -27,10 +27,16 @@ preflight/model-local evidence is required, but **production coexistence
 authorization is not required** so it can be measured. It launches only the muted
 worker and never app/audio. Health evidence is recorded in a unique local
 results/live/<session>/ directory through preparation and at most120 seconds of
-ready observation. It cannot confirm driver-ready or promote itself to a
-production session. Diagnostic readiness is not GPU/coexistence approval.
+ready observation. It cannot confirm driver-ready or automatically promote itself. Once the owner
+records genuine measured coexistence authorization, an explicit ON request can
+promote the same parked READY session after all production guards pass; this is
+recorded separately and still requires driver-ready. Keeping that worker avoids
+a circular requirement to prove fresh worker health after first killing it.
+Diagnostic readiness alone is not GPU/coexistence approval.
 
-Production ON uses the original full authorization guards. Explicit
+Production ON uses the original full authorization guards. Live readiness requires
+a fresh current-status record with route=live and advancing playback as well as
+the app ready file; a PortAudio initialization marker alone is insufficient. Explicit
 `confirm_driver_ready(session_id, audible=False)` rechecks the exact prepared
 session and fresh parked health, then starts app --input live using the real
 default namespace. Audible output must be explicitly requested and still passes

roadscore/prototype/live_target_daemon.py

@@ -11,7 +11,7 @@ import socketserver
 import threading
 import time
 import uuid
-from live_supervisor import LiveSupervisor
+from live_supervisor import LiveSupervisor,blocked_reason
 from live_owned_processes import OwnedLiveProcesses
 
 
@@ -61,7 +61,15 @@ class Engine:
             if self.diagnostic:raise RuntimeError('Diagnostic preparation cannot start playback or authorize driving')
             self.audible=payload.get('audible',False)
             return self.supervisor.confirm_driver_ready(payload.get('session_id'),self.observation,self.authorization)
-        if self.supervisor.enabled:raise RuntimeError('A live session already exists; stop it before changing mode')
+        if self.supervisor.enabled:
+            if action=='enable' and self.diagnostic and self.supervisor.state=='READY':
+                reason=blocked_reason(self.observation,self.authorization,self.clock(),require_parked=True)
+                if reason:raise RuntimeError(reason)
+                self.diagnostic=False;self.owned.diagnostic=False;self.supervisor.reason='Verified live preparation; explicit current driver-ready confirmation required'
+                if getattr(self.owned,'folder',None) is not None:
+                    (self.owned.folder/'production_authorization.json').write_text(json.dumps({'wall':time.time(),'session_id':self.supervisor.session_id,'authorization':asdict(self.authorization)}))
+                return self.status()
+            raise RuntimeError('A live session already exists; stop it before changing mode')
         self.diagnostic=action=='diagnostic';self.audible=False;self.diagnostic_ready_at=None
         if self.diagnostic:
             if not self.diagnostic_ok():raise RuntimeError('Fresh healthy parked diagnostic preflight is required')
@@ -94,7 +102,12 @@ class Engine:
                 if self.clock()-self.diagnostic_ready_at>120:
                     self.supervisor.stop('Parked diagnostic observation window completed');return
                 self.supervisor.state='READY';self.supervisor.reason='Diagnostic accepted audio ready; music disabled, no driving authorization'
-        else:self.supervisor.tick(self.observation,self.authorization,**health)
+        else:
+            if self.supervisor.state=='LIVE' and not health['app_ready']:
+                self.supervisor.stop('Live app status is stale or playback has stopped');return
+            self.supervisor.tick(self.observation,self.authorization,**health)
+            if self.supervisor.state=='STARTING' and self.clock()-getattr(self.owned,'app_started',self.clock())>30:
+                self.supervisor.stop('Live app failed to establish current input/playback readiness')
         if self.started is not None and self.clock()-self.started>1500 and self.supervisor.state=='PREPARING':
             self.supervisor.stop('Preparation timed out')
 

roadscore/prototype/test_live_target_adapter.py

@@ -122,3 +122,23 @@ def test_production_app_uses_live_namespace_preserves_prior_current(tmp_path):
     assert 'OPENPILOT_PREFIX' not in env and 'ZMQ' not in env
     assert (owned.folder/'previous_current/evidence').read_text()=='keep'
     with patch('live_owned_processes.os.killpg'):owned.stop_owned()
+
+def test_diagnostic_promotes_only_after_explicit_verified_on(engine):
+    engine.collector.authorization=replace(AUTH,coexistence_verified=False)
+    engine.command({'command':'diagnostic'});engine.tick();session=engine.supervisor.session_id
+    with pytest.raises(RuntimeError):engine.command({'command':'enable'})
+    engine.collector.authorization=AUTH
+    engine.command({'command':'enable'})
+    assert not engine.diagnostic and engine.supervisor.session_id==session
+    engine.owned.start_app.assert_not_called()
+    engine.command({'command':'driver_ready','session_id':session})
+    engine.owned.start_app.assert_called_once()
+
+def test_ready_file_alone_does_not_claim_live_playback(tmp_path):
+    owned,popen=fixture_owned(tmp_path);owned.app=Mock();owned.app.poll.return_value=None
+    current=tmp_path/'results/current';current.mkdir(parents=True);(current/'ready').write_text('ready')
+    assert not owned.health()['app_ready']
+    (current/'status.json').write_text(json.dumps({'route':'live','command_wall':owned.clock(),'elapsed':1}))
+    assert owned.health()['app_ready']
+    (current/'status.json').write_text(json.dumps({'route':'live','command_wall':owned.clock()-3,'elapsed':1}))
+    assert not owned.health()['app_ready']