Avoid Mac replay and UI publisher port collisions

roadscore/MAC_SHOWCASE.md

@@ -34,6 +34,8 @@ Route2 completed its full saved recording on the Mac, including the recorded tai
 
 The local configuration is ignored by Git: `roadscore/assets/demo_catalog.json` stores the explicit comma address, local controls port and registered routes. The comma has its own catalog pointing to persistent local assets. Each entry needs the route identity, a compatible completed archive and an optional matching curve plan. No discovery or automatic target switching occurs.
 
+The Mac launcher checks the complete runtime message registry before choosing its session namespace. This prevents the UI's own publishers, such as `uiDebug`, from hashing to the same TCP port as replay messages. The camera port is also reserved. The selected namespace is recorded in the Mac session's `launch.json`.
+
 The dual-native mode completed a full301.7-second FiiO audible run with Galaxy controls reaching both screens and successful cleanup. Tracking skew was85ms median and171ms at the95th percentile. Network delays sometimes paused corrections while replay continued. There were no callback or clock exceptions, but28 audio status flags and21 forward clock corrections were recorded, so completion does not establish glitch-free output. Fullscreen also passed an8-second native Mac replay check with no output flags or clock exceptions. The public JLab speaker still needs its own physical output/timing check.
 
 For a Mac-only interactive fallback, use:

roadscore/prototype/mac_showcase.py

@@ -6,6 +6,7 @@ import json
 import math
 import os
 from pathlib import Path
+import runpy
 import signal
 import subprocess
 import sys
@@ -496,11 +497,16 @@ def parent_main(a):
   for path in (py,rt/'tools/replay/replay',rt/'selfdrive/ui/ui.py'):
     if not path.exists():raise SystemExit('Missing existing host runtime: '+str(path))
   session=uuid.uuid4().hex
+  from replay_namespace import choose_namespace
+  # The UI also publishes services (including uiDebug), so validate the whole
+  # runtime registry rather than just the replay's --allow list.
+  namespace=choose_namespace(session,runpy.run_path(str(rt/'cereal/services.py'))['SERVICE_LIST'])
+  session=namespace.removeprefix('roadscore-showcase-')
   out=a.out or project/'roadscore/results'/('mac_showcase_'+str(int(time.time())))
   out.mkdir(parents=True,exist_ok=False)
   env=os.environ.copy()
-  env.update(PYTHONDONTWRITEBYTECODE='1',ZMQ='1',OPENPILOT_ZMQ_NAMESPACE='roadscore-showcase-'+session,ROADSCORE_SHOWCASE_SESSION=session,ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_REPLAY_UI_CONTROLS='1',PARAMS_ROOT=str(out/'params'),BASEDIR=str(rt),NOBOARD='1',SIMULATION='1',SKIP_FW_QUERY='1',BIG='0',SP_ALLOW_DESKTOP_FAKE_WIFI='0',SP_ALLOW_DESKTOP_FAKE_BLUETOOTH='0',SP_ONROAD_NAV_DEMO='0',SP_ONROAD_CEM_DEMO='0',ROADSCORE_CLEAN_DEMO_UI='1',ROADSCORE_OVERLAY='1',ROADSCORE_STATUS_FILE=str(out/'status.json'),ROADSCORE_UI_AUDIT=str(out/'ui_audit.jsonl'),ROADSCORE_OVERLAY_CAPTURE=str(out/'overlay.png'),ROADSCORE_PRESENTATION_POLICY='conservative-v4')
-  env['OPENPILOT_PREFIX']='roadscore-showcase-'+session
+  env.update(PYTHONDONTWRITEBYTECODE='1',ZMQ='1',OPENPILOT_ZMQ_NAMESPACE=namespace,ROADSCORE_SHOWCASE_SESSION=session,ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_REPLAY_UI_CONTROLS='1',PARAMS_ROOT=str(out/'params'),BASEDIR=str(rt),NOBOARD='1',SIMULATION='1',SKIP_FW_QUERY='1',BIG='0',SP_ALLOW_DESKTOP_FAKE_WIFI='0',SP_ALLOW_DESKTOP_FAKE_BLUETOOTH='0',SP_ONROAD_NAV_DEMO='0',SP_ONROAD_CEM_DEMO='0',ROADSCORE_CLEAN_DEMO_UI='1',ROADSCORE_OVERLAY='1',ROADSCORE_STATUS_FILE=str(out/'status.json'),ROADSCORE_UI_AUDIT=str(out/'ui_audit.jsonl'),ROADSCORE_OVERLAY_CAPTURE=str(out/'overlay.png'),ROADSCORE_PRESENTATION_POLICY='conservative-v4')
+  env['OPENPILOT_PREFIX']=namespace
   env['ROADSCORE_FULLSCREEN']='1' if a.fullscreen and not a.headless else '0'
   env['ROADSCORE_AUDIO_DRAIN_FILE']=str(out/'audio_drained.json')
   env['PWD']=str(rt)
@@ -510,7 +516,7 @@ def parent_main(a):
   if '--no-hw-decoder' in args:args.remove('--no-hw-decoder')
   if a.follow_playhead and '--headless' not in args:args.append('--headless')
   write_json(out/'status.json',dict(readiness='PREPARING',style='Prism',compute='prepared-core'))
-  write_json(out/'launch.json',dict(core_sha256=hashlib.sha256((a.score_archive/'dry.wav').read_bytes()).hexdigest(),curve_plan_sha256=hashlib.sha256(a.curve_plan.read_bytes()).hexdigest() if a.curve_plan else None,mode='prepared-interactive-showcase',route=a.route,source=str(a.score_archive),runtime=str(rt),native_replay_args=args,generation_invoked=False,network_required=False,session_id=session,muted=a.muted or a.headless,paired_comma=a.paired_comma,paired_controls_scope=DISCLOSURE))
+  write_json(out/'launch.json',dict(core_sha256=hashlib.sha256((a.score_archive/'dry.wav').read_bytes()).hexdigest(),curve_plan_sha256=hashlib.sha256(a.curve_plan.read_bytes()).hexdigest() if a.curve_plan else None,mode='prepared-interactive-showcase',route=a.route,source=str(a.score_archive),runtime=str(rt),native_replay_args=args,generation_invoked=False,network_required=False,session_id=session,zmq_namespace=namespace,muted=a.muted or a.headless,paired_comma=a.paired_comma,paired_controls_scope=DISCLOSURE))
   check=subprocess.run([str(py),'-c','from cereal import messaging; import sounddevice,soundfile; from prepared_core import load_archive; import sys; a,r,m=load_archive(sys.argv[1],sys.argv[2]); print("Prepared core:",len(a)/r,"seconds; local replay ready")',str(a.score_archive),a.route],cwd=rt,env=env)
   if check.returncode:raise SystemExit(check.returncode)
   if a.check:return

roadscore/prototype/replay_namespace.py

@@ -0,0 +1,31 @@
+"""Choose Mac replay message ports without collisions between its services."""
+
+
+def zmq_port(namespace, service):
+  # Match the 64-bit FNV-1a mapping in msgq/impl_zmq.cc.
+  value = 0xcbf29ce484222325
+  endpoint = f'{namespace}:{service}' if namespace else service
+  for byte in endpoint.encode():
+    value = ((value ^ byte) * 0x100000001b3) & 0xffffffffffffffff
+  return 8023 + value % (65535 - 8023)
+
+
+def namespace_collisions(namespace, services):
+  ports = {9000: ['camera']}
+  for service in sorted(set(services)):
+    ports.setdefault(zmq_port(namespace, service), []).append(service)
+  return {port: names for port, names in ports.items() if len(names) > 1}
+
+
+def choose_namespace(session, services, max_attempts=128):
+  if max_attempts < 1:
+    raise ValueError('At least one replay namespace attempt is required')
+  services = tuple(services)
+  if not services:
+    raise ValueError('Replay services must be available before selecting message ports')
+  base = 'roadscore-showcase-' + session
+  for attempt in range(max_attempts):
+    candidate = base if attempt == 0 else f'{base}-{attempt}'
+    if not namespace_collisions(candidate, services):
+      return candidate
+  raise RuntimeError('Could not choose collision-free Mac replay message ports')

roadscore/prototype/test_replay_namespace.py

@@ -0,0 +1,76 @@
+"""Regression for the real Mac uiDebug/replay publisher port collision."""
+import unittest
+
+from cereal.services import SERVICE_LIST
+from replay_namespace import choose_namespace, namespace_collisions, zmq_port
+from replay_ui_controls import isolated_replay
+
+
+FAILED_SESSION = 'a2daf09be6c143b98efea8e7471bfd49'
+FAILED_NAMESPACE = 'roadscore-showcase-' + FAILED_SESSION
+GOOD_SESSION = '00000000000000000000000000000001'
+
+
+class ReplayNamespaceTests(unittest.TestCase):
+  def test_hash_matches_cpp_fnv1a_port_vectors(self):
+    # impl_zmq.cc hashes namespace + ':' + service as unsigned 64-bit FNV1a,
+    # then maps into [8023, 65535). Empty namespace hashes the service alone.
+    vectors = [
+      ('', 'carState', 9041),
+      ('test', 'modelV2', 50273),
+      ('roadscore-showcase-simple', 'carState', 28976),
+      (FAILED_NAMESPACE, 'uiDebug', 54090),
+      (FAILED_NAMESPACE, 'starpilotSelfdriveState', 54090),
+    ]
+    for namespace, service, expected in vectors:
+      with self.subTest(namespace=namespace, service=service):
+        self.assertEqual(zmq_port(namespace, service), expected)
+
+  def test_failed_session_has_real_ui_and_replay_collision(self):
+    collisions = namespace_collisions(FAILED_NAMESPACE, SERVICE_LIST)
+    self.assertEqual(set(collisions), {54090})
+    self.assertEqual(set(collisions[54090]), {'uiDebug', 'starpilotSelfdriveState'})
+
+  def test_selector_avoids_collision_across_every_runtime_service(self):
+    namespace = choose_namespace(FAILED_SESSION, SERVICE_LIST)
+    self.assertEqual(namespace, FAILED_NAMESPACE + '-1')
+    self.assertEqual(namespace_collisions(namespace, SERVICE_LIST), {})
+    ports = [zmq_port(namespace, service) for service in SERVICE_LIST]
+    self.assertEqual(len(ports), len(set(ports)))
+    self.assertNotIn(9000, ports)
+
+  def test_good_base_namespace_is_preserved_deterministically(self):
+    expected = 'roadscore-showcase-' + GOOD_SESSION
+    self.assertEqual(namespace_collisions(expected, SERVICE_LIST), {})
+    self.assertEqual(choose_namespace(GOOD_SESSION, SERVICE_LIST), expected)
+    self.assertEqual(choose_namespace(GOOD_SESSION, reversed(list(SERVICE_LIST))), expected)
+
+  def test_salted_session_keeps_the_exact_replay_isolation_guard(self):
+    namespace = choose_namespace(FAILED_SESSION, SERVICE_LIST)
+    session = namespace.removeprefix('roadscore-showcase-')
+    self.assertEqual(session, FAILED_SESSION + '-1')
+    environ = dict(ROADSCORE_REPLAY_UI_CONTROLS='1', SIMULATION='1', ZMQ='1',
+                   ROADSCORE_PREPARED_SHOWCASE='1', ROADSCORE_SHOWCASE_SESSION=session,
+                   OPENPILOT_ZMQ_NAMESPACE=namespace, OPENPILOT_PREFIX=namespace)
+    self.assertTrue(isolated_replay(environ))
+    self.assertFalse(isolated_replay({**environ, 'ROADSCORE_SHOWCASE_SESSION': FAILED_SESSION}))
+    self.assertFalse(isolated_replay({**environ, 'OPENPILOT_ZMQ_NAMESPACE': FAILED_NAMESPACE}))
+
+  def test_camera_port_is_reserved_even_without_service_pair_collision(self):
+    session = '00000000000000000000000000001c59'
+    namespace = 'roadscore-showcase-' + session
+    self.assertEqual(zmq_port(namespace, 'carState'), 9000)
+    collisions = namespace_collisions(namespace, ['carState'])
+    self.assertIn(9000, collisions)
+    self.assertIn('carState', collisions[9000])
+    selected = choose_namespace(session, ['carState'])
+    self.assertNotEqual(selected, namespace)
+    self.assertNotEqual(zmq_port(selected, 'carState'), 9000)
+
+  def test_retry_limit_fails_instead_of_returning_conflicting_namespace(self):
+    with self.assertRaises(RuntimeError):
+      choose_namespace(FAILED_SESSION, SERVICE_LIST, max_attempts=1)
+
+
+if __name__ == '__main__':
+  unittest.main()