Expose bounded sanitized receiver failure details in normal launcher

roadscore/prototype/normal_onroad.py

@@ -11,6 +11,7 @@ from presentation_policy import select_launch
 from hook_launch import start_planner
 from composition_launch import configure as configure_composition
 from launch_health import check_children, describe_failure, write_failure
+from receiver_diagnostics import receiver_failure
 from session_seed import select_session, seed_argument, seed_environment, remote_assignments
 R=Path(__file__).resolve().parents[1]
 native=Path('/TICI').exists()
@@ -128,7 +129,7 @@ try:
      temporary=out/'roadscore_status.tmp';temporary.write_text(json.dumps({**initial_display,**progress}));temporary.replace(out/'roadscore_status.json')
     last_preparation_status=time.monotonic()
    check_children(named_children,remote=not native,include_receiver=True)
-   if receiver.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Bench receiver failed: '+(out/'receiver.log').read_text())
+   if receiver.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Bench receiver failed or timed out; see receiver.log')
    time.sleep(.2)
   audio_host=None
   if not native:
@@ -198,9 +199,10 @@ except Exception as error:
   check_children(named_children,remote=not native,include_receiver=True,allow_clean_receiver=True)
  except Exception as child_error:
   error=child_error
+ error=receiver_failure(error,out/'receiver.log')
  failure=describe_failure(error)
  write_failure(out/'roadscore_status.json',failure)
- raise
+ raise error from None
 finally:
  for c in reversed(children):
   if c.poll() is None:

roadscore/prototype/receiver_diagnostics.py

@@ -0,0 +1,24 @@
+"""Bounded terminal diagnostics for a failed RoadScore receiver."""
+from pathlib import Path
+import re
+
+
+def receiver_failure(error, log_path):
+  if getattr(error, 'component', None) != 'receiver':
+    return error
+  path=Path(log_path)
+  try:
+    with path.open('rb') as handle:
+      handle.seek(0,2);size=handle.tell();handle.seek(max(0,size-8192))
+      text=handle.read(8192).decode('utf-8',errors='replace')
+    text=re.sub(r'\x1b\[[0-?]*[ -/]*[@-~]', '', text)
+    text=''.join(c for c in text if c in '\n\t' or ord(c)>=32)
+    text=re.sub(r'(?i)(bearer\s+)[^\s]+', r'\1[redacted]', text)
+    text=re.sub(r'''(?ix)((?:token|password|secret|authorization|api[_-]?key)["']?\s*[:=]\s*["']?)[^\s,"'}]+''',r'\1[redacted]',text)
+    text=re.sub(r'(https?://)[^\s/@]+:[^\s/@]+@',r'\1[redacted]@',text)
+    excerpt='\n'.join(text.strip().splitlines()[-12:])[-1800:]
+  except OSError:
+    excerpt='Receiver log is unavailable.'
+  error.args=(f'RoadScore receiver exited with status {getattr(error,"code","unknown")}.\n'
+              f'Log: {path}\n{excerpt or "Receiver log is empty."}',)
+  return error

roadscore/prototype/test_receiver_diagnostics.py

@@ -0,0 +1,24 @@
+from pathlib import Path
+import tempfile
+import unittest
+from launch_health import LaunchFailure, describe_failure
+from receiver_diagnostics import receiver_failure
+
+class ReceiverDiagnosticsTests(unittest.TestCase):
+ def test_preserves_actual_final_error_and_log_location(self):
+  with tempfile.TemporaryDirectory() as d:
+   p=Path(d)/'receiver.log';p.write_text('noise\n'*10000+'Traceback (most recent call last):\n  app.py:200\nRuntimeError: Selected Bluetooth speaker missing\n')
+   e=receiver_failure(LaunchFailure('receiver',1),p)
+   self.assertIn('Selected Bluetooth speaker missing',str(e));self.assertIn(str(p),str(e))
+   self.assertLess(len(str(e)),2100);self.assertEqual(describe_failure(e)['failure_exit_code'],1)
+ def test_secret_redaction_and_missing_log(self):
+  with tempfile.TemporaryDirectory() as d:
+   p=Path(d)/'receiver.log';p.write_text('token=abcdef\nAuthorization: Bearer secret123\n{"password":"hidden"}\nhttps://user:pass@example.test\nValueError: output unavailable\n')
+   text=str(receiver_failure(LaunchFailure('receiver',1),p))
+   for secret in ['abcdef','secret123','hidden','user:pass']:self.assertNotIn(secret,text)
+   self.assertIn('ValueError: output unavailable',text)
+   self.assertIn('unavailable',str(receiver_failure(LaunchFailure('receiver',1),Path(d)/'absent')))
+ def test_other_failures_unchanged(self):
+  e=ValueError('original');self.assertIs(receiver_failure(e,'/unused'),e);self.assertEqual(str(e),'original')
+
+if __name__=='__main__':unittest.main()