Bind normal RoadScore sessions to causal host semantic plans and native sampling

roadscore/experiments/ace_chestnut_20260916/ace_worker.py

@@ -18,6 +18,11 @@ from generation_seed import configured_seed,sample_seed
 base_seed=configured_seed(required=True)
 from quality_gate import QualifiedGenerator,POLICY
 from link_health import LinkProbe
+from hook_service import Client
+from planned_composition import PlannedComposition
+composition_policy=os.environ.get('ROADSCORE_COMPOSITION_POLICY','prepared-v1')
+if composition_policy not in ('prepared-v1','hook-v2'):raise ValueError('Unknown composition policy')
+if composition_policy=='hook-v2' and not windowed:raise ValueError('Hook planning requires windowed native sampler')
 from tinygrad import Device
 profile=selected();preparation_id=f'{profile}_{time.time_ns()}'
 lock=open(G/'gpu.lock','w');fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
@@ -28,16 +33,23 @@ def write_json(path,data):
  tmp=path.with_suffix('.tmp');tmp.write_text(json.dumps(data,indent=2));tmp.replace(path)
 def save_wave(path,wave):sf.write(path,wave*POLICY.output_gain,48000,subtype='FLOAT')
 try:
- write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'profile':profile,'phase':'preparing'})
+ write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'composition_policy':composition_policy,'profile':profile,'phase':'preparing'})
  load_started=time.monotonic()
  c=Composer(P,profile=profile) if windowed else Composer(P)
  model_load_seconds=time.monotonic()-load_started
  probe=LinkProbe(Device['AMD'],G/'ace_link.jsonl');probe.install_failure_hook(contain=True)
  if windowed:c.decoder.trace=probe.trace
+ planned=None
+ if composition_policy=='hook-v2':
+  from ace_runtime import Composer as NativeComposer
+  from window_policy import retained_end
+  planned=PlannedComposition(Client(os.environ['ROADSCORE_PLANNER_URL'],os.environ['ROADSCORE_PLANNER_TOKEN']),G/'hook_sessions'/preparation_id,base_seed,profile)
+  def planned_generate(role,seed,previous):return planned.generate(lambda case,seed,previous:NativeComposer.generate(c,case,seed,previous),seed,previous,retained_end)
+ sample=planned_generate if planned else c.generate
  def generate(role,seed,previous):
   probe.preflight(role=role,seed=seed)
   try:
-   wave,latent,stats=c.generate(role,seed,previous);stats['power_limit_watts']=float(os.environ['AM_POWER_LIMIT']) if os.environ.get('AM_POWER_LIMIT') else None;stats['link_session']=probe.session;stats['host_peak_rss_kib']=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss;stats['tracked_allocation_bytes']=probe.last.get('allocator_bytes');return wave,latent,stats
+   wave,latent,stats=sample(role,seed,previous);stats['power_limit_watts']=float(os.environ['AM_POWER_LIMIT']) if os.environ.get('AM_POWER_LIMIT') else None;stats['link_session']=probe.session;stats['host_peak_rss_kib']=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss;stats['tracked_allocation_bytes']=probe.last.get('allocator_bytes');return wave,latent,stats
   except Exception as e:
    probe.sample('generation_exception',error=str(e),role=role,seed=seed);raise
  qualified=QualifiedGenerator(generate)
@@ -52,9 +64,11 @@ try:
  initial=None;last=None;preparation=[];slot=0
  while initial is None or len(initial)/48000<POLICY.initial_buffer_seconds:
   role=('initial' if windowed else 'verse') if initial is None else ('verse' if windowed else 'repaint_verse')
+  if planned:role=planned.begin(last)
   wave,last_new,stats=qualified.run(role,sample_seed(base_seed,"prepare",slot),last,record=record_for('prepare_'+preparation_id+'_'+str(slot)))
   preparation.append(stats)
   if wave is None:raise RuntimeError('Preparation rejected after bounded quality retries; inspect generated/quality')
+  if planned:planned.accept(wave,last_new)
   if initial is None:initial=wave.copy()
   else:
    overlap=2*48000;prefix=round(stats['prefix_seconds']*48000);alpha=np.linspace(0,1,overlap)[:,None]
@@ -63,8 +77,8 @@ try:
   last=last_new;slot+=1
   if slot>8:raise RuntimeError('Initial buffer did not fill within bounded preparation')
  save_wave(G/'ace_initial.wav',initial);np.save(G/'ace_initial.npy',last)
- write_json(G/'ace_initial.json',{'generation_seed':base_seed,'composer':'ace','startup_seconds':time.monotonic()-BOOT,'model_load_seconds':model_load_seconds,'host_peak_rss_kib':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,'duration':len(initial)/48000,'source_identity':'kpop_control','prepared_profile':profile if windowed else 'legacy','preparation_id':preparation_id,'continuation_policy':'quality-gated fixed lookahead','generation':preparation,'prepared_identity':True,'output_gain':POLICY.output_gain,'created_wall':time.time()})
- write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'profile':profile,'phase':'READY','initial_buffer_seconds':len(initial)/48000})
+ write_json(G/'ace_initial.json',{'generation_seed':base_seed,'composition_policy':composition_policy,'composer':'ace','startup_seconds':time.monotonic()-BOOT,'model_load_seconds':model_load_seconds,'host_peak_rss_kib':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,'duration':len(initial)/48000,'source_identity':'kpop_control','prepared_profile':profile if windowed else 'legacy','preparation_id':preparation_id,'continuation_policy':'quality-gated fixed lookahead','generation':preparation,'prepared_identity':True,'output_gain':POLICY.output_gain,'created_wall':time.time()})
+ write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'composition_policy':composition_policy,'profile':profile,'phase':'READY','initial_buffer_seconds':len(initial)/48000})
  (G/'request.json').unlink(missing_ok=True);ready.write_text('ace');print('ACE_READY',flush=True)
  while True:
   request=G/'request.json'
@@ -81,10 +95,12 @@ try:
    if case is None:raise ValueError('Unsupported ACE section intent')
    if windowed:case=case.removeprefix('repaint_')
    previous=np.load(req['latents'])
+   if planned:case=planned.begin(previous,arrival=case=='outro')
    deadline=req.get('playback_deadline_monotonic')
-   wave,latent,stats=qualified.run(case,int(req.get('seed',job%(2**32))),previous,deadline=deadline,record=record_for(job))
+   wave,latent,stats=qualified.run(case,int(req['seed']),previous,deadline=deadline,record=record_for(job))
    if wave is None:
     write_json(G/f'result_{job}.json',{**req,**stats,'id':job,'composer':'ace','seconds':time.monotonic()-start});continue
+   if planned:planned.accept(wave,latent)
    wav=G/f'ace_job_{job}.wav';lat=G/f'ace_job_{job}.npy';save_wave(wav,wave);np.save(lat,latent)
    result={**req,**stats,'id':job,'composer':'ace','backend':'ACE-Step1.5 turbo native Chestnut, prepared identity','wav':str(wav),'latents':str(lat),'seconds':time.monotonic()-start,'retained_seconds':stats['prefix_seconds'],'new_audio_start_frame':round(stats['prefix_seconds']*48000),'overlap_frames':96000,'sample_rate':48000,'output_gain':POLICY.output_gain,'conditioning':req['conditioning']}
    write_json(G/f'result_{job}.json',result);print('ACE_RESULT',job,result['seconds'],flush=True)
@@ -93,4 +109,4 @@ try:
   finally:busy.unlink(missing_ok=True)
 finally:
  ready.unlink(missing_ok=True)
- write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'profile':profile,'phase':'Stopped'})
+ write_json(G/'ace_worker_state.json',{'pid':os.getpid(),'generation_seed':base_seed,'composition_policy':composition_policy,'profile':profile,'phase':'Stopped'})

roadscore/prototype/hook_launch.py

@@ -0,0 +1,35 @@
+"""Automatic Mac planner lifecycle for ordinary remote native sessions."""
+import json
+import os
+from pathlib import Path
+import secrets
+import subprocess
+import time
+
+
+def enabled(session, replay, composer):
+ return bool(session and not replay and composer=='ace' and session['seed_origin']!='judging-route')
+
+
+def start_planner(launch, env, out, roadscore, bench, native=False):
+ if native:
+  if not env.get('ROADSCORE_PLANNER_URL') or not env.get('ROADSCORE_PLANNER_TOKEN'):
+   raise RuntimeError('Hook composition needs a connected host semantic planner; launch on the Mac or configure its authenticated loopback tunnel')
+  return
+ assets=Path(env.get('ROADSCORE_ASSETS_ROOT',str(Path.home()/'Desktop/RoadScore')))
+ python=Path(env.get('ROADSCORE_PLANNER_PYTHON',str(assets/'experiments/composition_20260916/venv/bin/python')))
+ if not python.is_file():raise RuntimeError('Existing Mac ACE preparation environment missing: '+str(python))
+ env['ROADSCORE_PLANNER_TOKEN']=secrets.token_urlsafe(32)
+ ready=out/'planner_ready.json'
+ planner=launch([str(python),str(roadscore/'prototype/hook_service.py'),'--assets-root',str(assets),'--cache',str(assets/'cache/hook-plans-v2'),'--ready',str(ready)],'semantic_planner')
+ deadline=time.monotonic()+300
+ while not ready.exists():
+  if planner.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Host semantic planner failed; see semantic_planner.log')
+  time.sleep(.1)
+ port=int(json.loads(ready.read_text())['port']);remote_port=int(env.get('ROADSCORE_PLANNER_REMOTE_PORT','8767'))
+ tunnel=launch(['ssh','-T','-o','ExitOnForwardFailure=yes','-o','ServerAliveInterval=15','-R',f'127.0.0.1:{remote_port}:127.0.0.1:{port}',bench,'printf "PLANNER_TUNNEL_READY\\n"; cat'],'semantic_tunnel',stdin=subprocess.PIPE)
+ deadline=time.monotonic()+30
+ while b'PLANNER_TUNNEL_READY' not in (out/'semantic_tunnel.log').read_bytes():
+  if tunnel.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Exclusive planner tunnel unavailable; see semantic_tunnel.log')
+  time.sleep(.1)
+ env['ROADSCORE_PLANNER_URL']=f'http://127.0.0.1:{remote_port}'

roadscore/prototype/hook_service.py

@@ -0,0 +1,82 @@
+"""Loopback-only semantic preparation transport; no diffusion or route input."""
+import argparse
+import base64
+import hmac
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.request import Request, urlopen
+from hook_planning import PlanCache, PlanRequest, request_plan
+
+LIMIT = 32 * 1024 * 1024
+
+class Client:
+ def __init__(self, url, token, timeout=600):
+  if not url.startswith('http://127.0.0.1:'):raise ValueError('Planner must use authenticated loopback tunnel')
+  self.url,self.token,self.timeout=url.rstrip('/'),token,timeout
+ def call(self, endpoint, data=None):
+  body=None if data is None else json.dumps(data).encode()
+  req=Request(self.url+endpoint,data=body,headers={'Authorization':'Bearer '+self.token,'Content-Type':'application/json'})
+  with urlopen(req,timeout=self.timeout) as response:
+   result=response.read(LIMIT+1)
+  if len(result)>LIMIT:raise ValueError('Oversized planner response')
+  return json.loads(result)
+ def fingerprints(self):return self.call('/identity')
+ def prepare(self, request, sources, output):
+  data=self.call('/plan',{'request':request.identity(),'sources':{k:base64.b64encode(Path(v).read_bytes()).decode() for k,v in sources.items()}})
+  for name,encoded in data['files'].items():
+   if Path(name).name!=name or name in ('.','..','cache.json'):raise ValueError('Invalid planner filename')
+   (output/name).write_bytes(base64.b64decode(encoded,validate=True))
+
+
+def make_server(adapter, cache, token, port=0):
+ lock=threading.Lock()
+ class Handler(BaseHTTPRequestHandler):
+  def log_message(self,*args):pass
+  def respond(self,status,data):
+   body=json.dumps(data).encode();self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)
+  def authorized(self):return hmac.compare_digest(self.headers.get('Authorization',''),'Bearer '+token)
+  def do_GET(self):
+   if not self.authorized():return self.respond(403,{'error':'Authentication required'})
+   if self.path!='/identity':return self.respond(404,{'error':'Unknown endpoint'})
+   try:self.respond(200,adapter.fingerprints())
+   except Exception as exc:self.respond(500,{'error':str(exc)})
+  def do_POST(self):
+   if not self.authorized():return self.respond(403,{'error':'Authentication required'})
+   if self.path!='/plan':return self.respond(404,{'error':'Unknown endpoint'})
+   try:
+    size=int(self.headers.get('Content-Length','0'))
+    if not 0<size<=LIMIT:raise ValueError('Invalid planner request size')
+    data=json.loads(self.rfile.read(size));request=PlanRequest(**data['request'])
+    canonical=request_plan(**{k:v for k,v in request.identity().items() if k in ('session_seed','plan_index','profile','section','window_seconds','model_fingerprint','preparation_fingerprint','hook_reference_sha256','committed_prefix_sha256','previous_plan_sha256')})
+    if request!=canonical:raise ValueError('Request does not match current composition policy')
+    if any(request.identity()[k]!=v for k,v in adapter.fingerprints().items()):raise ValueError('Planner identity mismatch')
+    with lock,tempfile.TemporaryDirectory(dir=cache.root) as scratch:
+     sources={}
+     for name,value in data['sources'].items():
+      if name not in ('hook_reference','committed_prefix'):raise ValueError('Unknown musical context')
+      path=Path(scratch)/name;path.write_bytes(base64.b64decode(value,validate=True));sources[name]=path
+     directory,hit=cache.resolve(request,adapter,sources=sources)
+     files={p.name:base64.b64encode(p.read_bytes()).decode() for p in directory.iterdir() if p.name!='cache.json'}
+    self.respond(200,{'files':files,'cache_hit':hit})
+   except Exception as exc:self.respond(400,{'error':str(exc)})
+ cache.root.mkdir(parents=True,exist_ok=True)
+ return ThreadingHTTPServer(('127.0.0.1',port),Handler)
+
+
+def main():
+ parser=argparse.ArgumentParser();parser.add_argument('--assets-root',type=Path,required=True);parser.add_argument('--cache',type=Path,required=True);parser.add_argument('--ready',type=Path,required=True);args=parser.parse_args()
+ sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'experiments/ace_chestnut_20260916'))
+ from host_hook_adapter import HostHookAdapter
+ token=os.environ['ROADSCORE_PLANNER_TOKEN'];adapter=HostHookAdapter(args.assets_root)
+ server=make_server(adapter,PlanCache(args.cache),token)
+ identity=adapter.fingerprints()
+ args.ready.write_text(json.dumps({'port':server.server_port,**identity}))
+ try:server.serve_forever()
+ finally:server.server_close();args.ready.unlink(missing_ok=True)
+
+if __name__=='__main__':main()

roadscore/prototype/native_receiver.sh

@@ -32,6 +32,7 @@ for attempt in $(seq 1 "$preparation_wait"); do
   sleep 1
 done
 [ -f generated/worker_ready ] || { echo 'Worker preparation timed out'; exit 1; }
+if [ "${ROADSCORE_COMPOSER:-ace}" = ace ]; then /usr/local/venv/bin/python prototype/prepared_session.py; fi
 export OPENPILOT_PREFIX=roadscore_native
 export PYTHONPATH=/data/openpilot:/data/roadscore/prototype:/data/roadscore-feasibility/venv/lib/python3.12/site-packages
 mkdir -p /dev/shm/msgq_roadscore_native

roadscore/prototype/normal_onroad.py

@@ -6,19 +6,23 @@ No RoadScore event annotations, route allowlist, custom camera drawing, or model
 import argparse,os,subprocess,time,signal,shlex,json
 from pathlib import Path
 from clock_sync import measure
+from presentation_policy import select_launch
+from hook_launch import enabled as hook_enabled, start_planner
 from session_seed import select_session, seed_argument, seed_environment, remote_assignments
 R=Path(__file__).resolve().parents[1]
 native=Path('/TICI').exists()
 def interrupt(*_):raise KeyboardInterrupt
 signal.signal(signal.SIGTERM,interrupt)
-p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default='current',help='Explicit ACE output-only gold core bypass');p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--start',type=int,default=0);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
-if a.replay and a.render_mode!='current':raise SystemExit('Stored scores retain their recorded rendering; do not apply gold-core to a finished mix')
+p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default=None,help='ACE rendering mode');p.add_argument('--roadscore-presentation',choices=['conservative-v1','off','frozen']);p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--start',type=int,default=0);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
 if a.render_mode=='gold-core' and a.composer!='ace':raise SystemExit('Gold core requires ACE')
 if a.roadscore_seed is not None and (a.replay or a.composer!='ace'):raise SystemExit('--roadscore-seed applies only to fresh ACE generation')
 session=None
 if not a.replay and a.composer=='ace':
  judging_seed=os.environ.get('ROADSCORE_GENERATION_SEED') if os.environ.get('ROADSCORE_SEED_ORIGIN')=='judging-route' and a.roadscore_seed is not None else None
  session=select_session(a.roadscore_seed,judging_seed=judging_seed)
+presentation=select_launch(a.composer,a.profile,replay=a.replay,judging=bool(session and session['seed_origin']=='judging-route'),render_mode=a.render_mode,policy=a.roadscore_presentation)
+a.render_mode=presentation['render_mode']
+composition_policy='hook-v2' if hook_enabled(session,a.replay,a.composer) and not a.transport_only else 'prepared-v1'
 from settings import Settings,resolve_route
 a.routeid=resolve_route(p,a.route,a.routeid)
 settings=Settings(mode='stored' if a.replay else 'generate',muted=a.muted,overlay=not a.no_overlay,output_device=a.audio_device)
@@ -34,8 +38,10 @@ if session:
  env.update(seed_environment(session));(out/'session_seed.json').write_text(json.dumps(session,indent=2));print('RoadScore session seed:',session['generation_seed'],'('+session['seed_origin']+')',flush=True)
 else:
  env.pop('ROADSCORE_GENERATION_SEED',None);env.pop('ROADSCORE_SEED_ORIGIN',None)
-(out/'settings.json').write_text(json.dumps({**settings.snapshot(a.headless),'composer':a.composer,'profile':a.profile,'render_mode':a.render_mode,**(session or {})},indent=2))
+(out/'settings.json').write_text(json.dumps({**settings.snapshot(a.headless),'composer':a.composer,'profile':a.profile,'render_mode':a.render_mode,'presentation_policy':presentation['policy'],'composition_policy':composition_policy,**(session or {})},indent=2))
 env['ROADSCORE_RENDER_MODE']=a.render_mode
+env['ROADSCORE_PRESENTATION_POLICY']=presentation['policy']
+env['ROADSCORE_COMPOSITION_POLICY']=composition_policy
 env['ROADSCORE_COMPOSER']=a.composer
 env['ROADSCORE_ACE_PROFILE']=a.profile
 env['ROADSCORE_OVERLAY_CAPTURE']=str(out/'overlay.png')
@@ -70,6 +76,7 @@ def launch(cmd,name,**kw):
  f=(out/(name+'.log')).open('wb');logs.append(f);c=subprocess.Popen(cmd,stdout=f,stderr=f,env=env,cwd=rt,start_new_session=True,**kw);children.append(c);return c
 try:
  if not native and Path('/usr/bin/caffeinate').exists():launch(['/usr/bin/caffeinate','-i'],'wake_assertion')
+ if composition_policy=='hook-v2' and not a.transport_only:start_planner(launch,env,out,R,a.bench,native=native)
  # Native parameter seeding reads metadata; it does not feed future route data to RoadScore.
  subprocess.run([str(py),str(rt/'tools/replay/onroad_config.py'),'seed',*args],env=env,cwd=rt,check=True,stdout=(out/'seed.log').open('w'),stderr=subprocess.STDOUT)
  if native and not a.headless:
@@ -88,7 +95,8 @@ try:
    if sender.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Stored audio failed; see stored_audio.log')
    time.sleep(.1)
  else:
-  receiver_command=(['env','-u','ZMQ','ROADSCORE_AUDIBLE='+('1' if a.audible else '0'),'bash',str(R/'prototype/native_receiver.sh'),a.routeid] if native else ['ssh',a.bench,(remote_assignments(session) if session else '')+'ROADSCORE_RENDER_MODE='+a.render_mode+' ROADSCORE_COMPOSER='+a.composer+' ROADSCORE_ACE_PROFILE='+a.profile+' ROADSCORE_PCM_RETURN=1 '+('ROADSCORE_NO_GENERATION=1 ' if a.transport_only else '')+'bash /data/roadscore/prototype/native_receiver.sh '+shlex.quote(a.routeid)])
+  remote_policy=' '.join(k+'='+shlex.quote(env[k]) for k in ('ROADSCORE_PRESENTATION_POLICY','ROADSCORE_COMPOSITION_POLICY','ROADSCORE_PLANNER_URL','ROADSCORE_PLANNER_TOKEN') if k in env)+' '
+  receiver_command=(['env','-u','ZMQ','ROADSCORE_AUDIBLE='+('1' if a.audible else '0'),'bash',str(R/'prototype/native_receiver.sh'),a.routeid] if native else ['ssh',a.bench,(remote_assignments(session) if session else '')+remote_policy+'ROADSCORE_RENDER_MODE='+a.render_mode+' ROADSCORE_COMPOSER='+a.composer+' ROADSCORE_ACE_PROFILE='+a.profile+' ROADSCORE_PCM_RETURN=1 '+('ROADSCORE_NO_GENERATION=1 ' if a.transport_only else '')+'bash /data/roadscore/prototype/native_receiver.sh '+shlex.quote(a.routeid)])
   receiver=launch(receiver_command,'receiver',stdin=subprocess.PIPE)
   deadline=time.monotonic()+(1560 if a.composer=='ace' else 420)
   while b'BRIDGE_READY' not in (out/'receiver.log').read_bytes():

roadscore/prototype/planned_composition.py

@@ -0,0 +1,53 @@
+"""Transactional semantic-plan binding to the native sampler and accepted audio."""
+import json
+import hashlib
+from pathlib import Path
+import time
+import numpy as np
+import soundfile as sf
+from hook_planning import PlanCache, digest, request_plan, next_section
+
+class PlannedComposition:
+ def __init__(self, client, root, session_seed, profile):
+  self.client,self.root,self.seed,self.profile=client,Path(root),session_seed,profile
+  self.root.mkdir(parents=True,exist_ok=True)
+  self.cache=PlanCache(self.root/'plans');self.identity=client.fingerprints()
+  self.index=0;self.previous_key=None;self.pending=None;self.accepted={}
+  self.hook=self.root/'accepted_hook.wav'
+ def begin(self, previous, arrival=False):
+  sources={};context={}
+  if previous is not None:
+   identity=hashlib.sha256(previous.tobytes()).hexdigest()
+   if identity not in self.accepted:raise ValueError('Continuation source was not accepted in this session')
+   self.index,self.previous_key=self.accepted[identity]
+  role='initial' if self.index==0 else next_section(self.index-1,arrival_intent=arrival)
+  if self.index:
+   if previous is None or previous.ndim!=3 or previous.shape[0]!=1 or previous.shape[1]<200 or previous.shape[2]!=64:raise ValueError('Committed musical tail required')
+   prefix=self.root/'committed_prefix.npy';np.save(prefix,previous[:,-200:].copy())
+   sources={'hook_reference':self.hook,'committed_prefix':prefix}
+   context={'hook_reference_sha256':digest(self.hook),'committed_prefix_sha256':digest(prefix),'previous_plan_sha256':self.previous_key}
+  elif previous is not None:raise ValueError('Fresh composition cannot reuse a previous session latent')
+  request=request_plan(session_seed=self.seed,plan_index=self.index,profile=self.profile,section=role,window_seconds=30 if self.index==0 else 45,**self.identity,**context)
+  started=time.monotonic();directory,hit=self.cache.resolve(request,self.client.prepare,sources=sources)
+  self.pending={'request':request.identity(),'plan_key':request.cache_key,'directory':str(directory),'host_preparation_seconds':time.monotonic()-started,'plan_cache_hit':hit}
+  (self.root/f'plan_{self.index}.json').write_text(json.dumps(self.pending,indent=2))
+  return role
+ def generate(self, native_generate, seed, previous, retain):
+  if self.pending is None:raise RuntimeError('Begin a semantic plan before sampling')
+  wave,latent,stats=native_generate(self.pending['directory'],seed,previous)
+  role=self.pending['request']['section'];seconds=28 if self.index==0 else 36
+  try:frames,endpoint=retain(wave,48000,stats['prefix_seconds'],seconds,allow_fade=role=='outro')
+  except ValueError as exc:
+   frames=seconds*25;endpoint={'rejected':str(exc)};stats['endpoint_error']=str(exc)
+  committed=wave[:frames*1920];z=latent[:,:frames]
+  if len(committed)!=frames*1920 or z.shape[1]!=frames:raise ValueError('Native sampler returned incomplete planned window')
+  model_duration=stats['duration'];duration=frames/25;new=duration-stats['prefix_seconds']
+  if new<=0 or not np.isfinite(committed).all():raise ValueError('Invalid committed planned audio')
+  stats.update(self.pending,case=role,prepared_profile=self.profile,model_duration=model_duration,duration=duration,new_seconds=new,discarded_lookahead_seconds=model_duration-duration,endpoint=endpoint,composition_policy='hook-v2',preparation_host='authenticated host semantic planner',warm_rtf_new_audio=(stats['generation_seconds']+stats['decode_seconds'])/new)
+  return committed,z,stats
+ def accept(self,wave,latent):
+  if self.pending is None:raise RuntimeError('No pending plan')
+  if self.index==0:sf.write(self.hook,wave,48000,subtype='FLOAT')
+  self.previous_key=self.pending['plan_key'];self.index+=1
+  self.accepted[hashlib.sha256(latent.tobytes()).hexdigest()]=(self.index,self.previous_key)
+  self.pending=None

roadscore/prototype/prepared_session.py

@@ -0,0 +1,12 @@
+"""Refuse accidental reuse of a different resident composition."""
+import json
+import os
+from pathlib import Path
+
+
+def verify(metadata,environ):
+ if int(metadata.get('generation_seed',-1))!=int(environ['ROADSCORE_GENERATION_SEED']):raise ValueError('Resident composer belongs to another seed; its owner must finish/release that session')
+ if metadata.get('prepared_profile')!=environ.get('ROADSCORE_ACE_PROFILE','prism'):raise ValueError('Resident composer profile differs from this launch')
+ if metadata.get('composition_policy','prepared-v1')!=environ.get('ROADSCORE_COMPOSITION_POLICY','prepared-v1'):raise ValueError('Resident composer policy differs from this launch')
+
+if __name__=='__main__':verify(json.loads(Path('/data/roadscore/generated/ace_initial.json').read_text()),os.environ)

roadscore/prototype/test_hook_launch.py

@@ -0,0 +1,20 @@
+import unittest
+from hook_launch import enabled
+from prepared_session import verify
+from session_seed import select_session
+
+class LaunchTests(unittest.TestCase):
+ def test_fresh_and_explicit_but_not_judging_or_replay(self):
+  self.assertTrue(enabled(select_session(123),False,'ace'))
+  self.assertTrue(enabled(select_session(random_bits=lambda _:234),False,'ace'))
+  self.assertFalse(enabled(select_session(123,judging_seed=123),False,'ace'))
+  self.assertFalse(enabled(select_session(123),True,'ace'))
+  self.assertFalse(enabled(None,False,'sa3'))
+ def test_resident_policy_and_seed_must_match(self):
+  metadata={'generation_seed':123,'prepared_profile':'prism','composition_policy':'hook-v2'}
+  env={'ROADSCORE_GENERATION_SEED':'123','ROADSCORE_ACE_PROFILE':'prism','ROADSCORE_COMPOSITION_POLICY':'hook-v2'}
+  verify(metadata,env)
+  for key,value in [('generation_seed',124),('prepared_profile','aurora'),('composition_policy','prepared-v1')]:
+   with self.assertRaises(ValueError):verify({**metadata,key:value},env)
+
+if __name__=='__main__':unittest.main()

roadscore/prototype/test_planned_composition.py

@@ -0,0 +1,73 @@
+import json
+from pathlib import Path
+import tempfile
+import threading
+import unittest
+from urllib.error import HTTPError
+import numpy as np
+from hook_service import Client,make_server
+from hook_planning import PlanCache
+from planned_composition import PlannedComposition
+from test_hook_planning import fake_prepare
+
+class Adapter:
+ def fingerprints(self):return {'model_fingerprint':'a'*64,'preparation_fingerprint':'b'*64}
+ def __call__(self,req,sources,out):
+  fake_prepare(req,sources,out)
+  if req.prefix_seconds:
+   n=req.window_seconds*25;prefix=np.load(sources['committed_prefix'])
+   context=np.zeros((1,n,128),np.float32);context[:,:200,:64]=prefix
+   source=np.zeros((1,n,64),np.float32);source[:,:200]=prefix
+   np.save(out/'context_latents.npy',context);np.save(out/'sampler_clean_src_latents.npy',source)
+   np.save(out/'sampler_repaint_mask.npy',np.arange(n)[None,:]>=200)
+   (out/'sampler.json').write_text(json.dumps({'repaint_crossfade_frames':12,'repaint_injection_ratio':.5}))
+
+def retain(wave,sr,prefix,seconds,allow_fade=False):return seconds*25,{}
+
+def sampler(path,seed,previous):
+ context=np.load(Path(path)/'context_latents.npy');n=context.shape[1]
+ latent=np.full((1,n,64),seed,np.float16)
+ wave=np.full((n*1920,2),.1,np.float32)
+ return wave,latent,dict(duration=n/25,prefix_seconds=0 if previous is None else 8,generation_seconds=1,decode_seconds=.1)
+
+class IntegrationTests(unittest.TestCase):
+ def setUp(self):
+  self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name)
+  self.server=make_server(Adapter(),PlanCache(self.root/'server'),'secret')
+  self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start()
+  self.client=Client('http://127.0.0.1:'+str(self.server.server_port),'secret')
+ def tearDown(self):
+  self.server.shutdown();self.server.server_close();self.thread.join();self.temp.cleanup()
+ def test_freshplans_causal_chain_and_discarded_result(self):
+  run=PlannedComposition(self.client,self.root/'session',101,'prism')
+  self.assertEqual(run.begin(None),'initial')
+  wave,latent,stats=run.generate(sampler,11,None,retain)
+  self.assertEqual(len(wave),28*48000);self.assertEqual(latent.shape,(1,700,64))
+  initial_key=stats['plan_key'];run.accept(wave,latent)
+  committed=latent.copy()
+  self.assertEqual(run.begin(committed),'verse')
+  wave,latent,stats=run.generate(sampler,12,committed,retain)
+  self.assertEqual(len(wave),36*48000);self.assertEqual(stats['new_seconds'],28)
+  self.assertEqual(stats['request']['previous_plan_sha256'],initial_key)
+  prefix=np.load(self.root/'session/committed_prefix.npy');np.testing.assert_array_equal(prefix,committed[:,-200:])
+  run.accept(wave,latent)
+  # Audio scheduler discarded this result: its next request supplies the old committed tail.
+  self.assertEqual(run.begin(committed),'verse')
+  self.assertEqual(run.pending['request']['previous_plan_sha256'],initial_key)
+  self.assertTrue(run.pending['plan_cache_hit'])
+  run.accept(wave,latent)
+  self.assertEqual(run.begin(latent),'prechorus')
+  wave,new,stats=run.generate(sampler,13,latent,retain);run.accept(wave,new)
+  self.assertEqual(run.begin(new),'chorus')
+  other=PlannedComposition(self.client,self.root/'other',102,'prism');other.begin(None)
+  self.assertNotEqual(other.pending['plan_key'],initial_key)
+ def test_unknown_context_and_bad_auth_fail_closed(self):
+  run=PlannedComposition(self.client,self.root/'session',101,'prism')
+  with self.assertRaises(ValueError):run.begin(np.zeros((1,700,64)))
+  with self.assertRaises(HTTPError):Client(self.client.url,'wrong').fingerprints()
+ def test_seed_and_prompt_tampering_rejected(self):
+  run=PlannedComposition(self.client,self.root/'session',101,'prism');run.begin(None)
+  request=run.pending['request'].copy();request['caption']='stale generic plan'
+  with self.assertRaises(HTTPError):self.client.call('/plan',{'request':request,'sources':{}})
+
+if __name__=='__main__':unittest.main()