Audio

Provide alert playback and volume control with a native audio callback that does not depend on the Python GIL.

Co-authored-by: Daniel Koepping <elkoled@gmail.com>

openpilot/starpilot/audio/__init__.py

@@ -0,0 +1 @@
+"""Shared audio preference contracts."""

openpilot/starpilot/audio/alert_volume.py

@@ -0,0 +1,71 @@
+"""Validated saved volume choices for the existing stock audible alerts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+AUTO = 101
+MAX_RAW_BYTES = 16
+
+# The first two warnings retain an audible floor. The immediate warning still
+# ramps to full volume in soundd, including the selfdrive timeout warning.
+VOLUMES = (
+  ("WarningImmediateVolume", "Immediate Warning", 25),
+  ("WarningSoftVolume", "Soft Warning", 25),
+  ("RefuseVolume", "Engagement Refused", 0),
+  ("PromptDistractedVolume", "Distracted Driver", 0),
+  ("EngageVolume", "Engagement Chime", 0),
+  ("DisengageVolume", "Disengagement Alert", 0),
+  ("PromptVolume", "General Prompt", 0),
+)
+SPECS = {key: (label, minimum) for key, label, minimum in VOLUMES}
+
+
+@dataclass(frozen=True)
+class SavedVolume:
+  raw: bytes | None
+  value: int | None
+  readable: bool = True
+
+  @property
+  def valid(self) -> bool:
+    return self.readable and self.value is not None
+
+
+def parse_volume(key: str, raw: bytes | None) -> int | None:
+  if key not in SPECS:
+    raise KeyError(key)
+  if raw is None:
+    return AUTO
+  if not raw or len(raw) > MAX_RAW_BYTES or not raw.isascii() or not raw.isdigit():
+    return None
+  value = int(raw)
+  minimum = SPECS[key][1]
+  return value if value == AUTO or minimum <= value <= 100 else None
+
+
+def read_volume(params: Any, key: str) -> SavedVolume:
+  if key not in SPECS:
+    raise KeyError(key)
+  try:
+    with Path(params.get_param_path(key)).open("rb") as file:
+      raw = file.read(MAX_RAW_BYTES + 1)
+  except FileNotFoundError:
+    raw = None
+  except (OSError, ValueError):
+    return SavedVolume(None, None, False)
+  if raw is not None and len(raw) > MAX_RAW_BYTES:
+    return SavedVolume(raw, None, False)
+  return SavedVolume(raw, parse_volume(key, raw))
+
+
+def effective_volume(key: str, saved: int | None, ambient: float, *, immediate_ramp: float | None = None) -> float:
+  """Return gain without changing the stock Auto path or immediate-warning ramp."""
+  if key not in SPECS:
+    raise KeyError(key)
+  if saved is None or saved == AUTO:
+    return ambient
+  selected = saved / 100.0
+  return max(selected, immediate_ramp) if immediate_ramp is not None else selected

openpilot/starpilot/audio/availability_warning.py

@@ -0,0 +1,21 @@
+"""Bounded delivery of an already selected availability chime, independent of authority."""
+MAX_DELIVERY_NS = 400_000_000
+
+class AvailabilityWarningDelivery:
+  def __init__(self):
+    self.context = None
+    self.until_ns = 0
+    self.last_ns = 0
+
+  def observe(self, *, session, drive, now_ns, selected, requested_until_ns):
+    context = (session, drive) if session and drive > 0 else None
+    if context != self.context or now_ns < self.last_ns or context is None:
+      self.until_ns = 0
+    self.context, self.last_ns = context, now_ns
+    if context is not None and selected and now_ns <= requested_until_ns:
+      self.until_ns = max(self.until_ns, min(requested_until_ns, now_ns + MAX_DELIVERY_NS))
+
+  def pending(self, *, session, drive, now_ns, audible_none, priority, low_priority):
+    return (self.context is not None and self.context == (session, drive) and drive > 0 and
+            self.last_ns <= now_ns <= self.until_ns and
+            audible_none and priority <= low_priority)

openpilot/starpilot/audio/axis_alerts.py

@@ -0,0 +1,66 @@
+from openpilot.cereal import log
+from openpilot.starpilot.aol.runtime import current_intent
+from openpilot.starpilot.aol.wire import INTENT_SERVICE
+
+AudibleAlert = log.SelfdriveState.AudibleAlert
+COALESCE_NS = 250_000_000
+
+
+class _Transition:
+  def __init__(self):
+    self.session = None
+    self.sequence = 0
+    self.active = False
+    self.observed_at = 0
+
+  def update(self, session, sequence: int, active: bool, now_ns: int) -> int:
+    changed = 0
+    if session is None:
+      self.session = None
+    elif session != self.session or now_ns - self.observed_at > COALESCE_NS:
+      self.session, self.sequence, self.active, self.observed_at = session, sequence, active, now_ns
+    elif sequence > self.sequence:
+      changed = int(active) - int(self.active)
+      self.sequence, self.active, self.observed_at = sequence, active, now_ns
+    return changed
+
+
+class AxisAlerts:
+  def __init__(self):
+    self.lateral = _Transition()
+    self.longitudinal = _Transition()
+    self.last_sound = AudibleAlert.none
+    self.last_sound_at = 0
+    self.native_sound = AudibleAlert.none
+
+  def update(self, sm, native_sound: int, now_ns: int) -> int:
+    intent = (current_intent(sm, car_state_ns=int(sm.logMonoTime[INTENT_SERVICE]), now_ns=now_ns)
+              if INTENT_SERVICE in sm.logMonoTime else None)
+    # Engagement is persistent driver intent. Temporary native permission loss,
+    # pedal override and output suppression must not replay engagement sounds.
+    state_ns = int(sm.logMonoTime.get('selfdriveState', 0))
+    state_fresh = (state_ns > 0 and 0 <= now_ns - state_ns <= 100_000_000 and
+                   sm.valid['selfdriveState'] and sm.alive['selfdriveState'])
+    lateral_known = intent is not None and (intent.pauseLateral or intent.lateralArmed or state_fresh)
+    lateral = self.lateral.update(intent.producerSessionId if lateral_known else None, intent.sequence if intent else 0,
+                                  bool(intent and not intent.pauseLateral and
+                                       (intent.lateralArmed or (state_fresh and sm['selfdriveState'].active))), now_ns)
+    longitudinal = self.longitudinal.update(
+      intent.producerSessionId if intent is not None and state_fresh else None,
+      intent.sequence if intent is not None else 0,
+      bool(intent is not None and state_fresh and sm['selfdriveState'].enabled and not intent.pauseLongitudinal), now_ns)
+    extra_sound = (AudibleAlert.disengage if min(lateral, longitudinal) < 0 else
+                   AudibleAlert.engage if max(lateral, longitudinal) > 0 else AudibleAlert.none)
+
+    if native_sound != AudibleAlert.none:
+      if native_sound != self.native_sound and native_sound in (AudibleAlert.engage, AudibleAlert.disengage):
+        self.last_sound, self.last_sound_at = native_sound, now_ns
+      self.native_sound = native_sound
+      return native_sound
+
+    self.native_sound = native_sound
+    if extra_sound != AudibleAlert.none:
+      if extra_sound == self.last_sound and now_ns - self.last_sound_at < COALESCE_NS:
+        return AudibleAlert.none
+      self.last_sound, self.last_sound_at = extra_sound, now_ns
+    return extra_sound

openpilot/starpilot/audio/catalog.json

@@ -0,0 +1,33 @@
+{
+  "version": 1,
+  "packs": [
+    {
+      "id": "duck",
+      "name": "Duck",
+      "path": "theme/Themes/duck/sounds.zip",
+      "size": 75095,
+      "sha256": "c37c1f0f177a08a79b4eae723392bedb39b2b99d49a4226b0ca5141a00d523f8"
+    },
+    {
+      "id": "old_engage_sounds",
+      "name": "Old Engage Sounds",
+      "path": "theme/Themes/old_engage_sounds/sounds.zip",
+      "size": 376387,
+      "sha256": "2fff703ee69452789ec7884e6447c146a0f45c6fa39c70455fd9db88e15ebb54"
+    },
+    {
+      "id": "tesla",
+      "name": "Tesla",
+      "path": "theme/Themes/tesla/sounds.zip",
+      "size": 237831,
+      "sha256": "98a13264d9c5c61f0f9f2a8c5e073aa32b3c3679c671c3cb520f9c2c443c79eb"
+    },
+    {
+      "id": "weeb",
+      "name": "Weeb",
+      "path": "theme/Themes/weeb/sounds.zip",
+      "size": 1139378,
+      "sha256": "e367fdfc0d3d6604183e779180b281471f2135a2be5e75b9766a52c03696ba36"
+    }
+  ]
+}

openpilot/starpilot/audio/default_enrollment.py

@@ -0,0 +1,79 @@
+"""Enroll a settings namespace in the built-in sounds once, before consumers start."""
+
+from pathlib import Path
+import json
+import os
+import tempfile
+
+from openpilot.starpilot.state_migration import (
+  _atomic_write, _fsync_dir, _outside_params, _params_lock, _private_dir, _read_value, canonical_json, digest,
+)
+
+
+def _write_selection(namespace):
+  fd, temporary = tempfile.mkstemp(prefix=".sound-default-", dir=namespace.parent)
+  try:
+    with os.fdopen(fd, "wb") as output:
+      output.write(b"starpilot")
+      output.flush()
+      os.fsync(output.fileno())
+    os.replace(temporary, namespace / "SoundPack")
+    _fsync_dir(namespace)
+  finally:
+    Path(temporary).unlink(missing_ok=True)
+
+
+def enroll_default_sounds(params, storage) -> bool:
+  namespace, storage = Path(params.get_param_path()).absolute(), Path(storage).absolute()
+  _outside_params(namespace, storage)
+  with _params_lock(namespace):
+    identity = {"version": 1, "namespace": str(namespace), "target": str(namespace.resolve(strict=True))}
+    marker_dir = storage / "sound-default-once"
+    new_directory = not marker_dir.exists()
+    _private_dir(marker_dir)
+    if new_directory:
+      _fsync_dir(storage)
+    marker = marker_dir / f"{digest(str(namespace).encode())}.json"
+    complete = {**identity, "state": "complete"}
+    try:
+      encoded = _read_value(marker)
+    except FileNotFoundError:
+      encoded = None
+    previous = json.loads(encoded) if encoded is not None else None
+    if previous is not None:
+      if not isinstance(previous, dict) or type(previous.get("version")) is not int or previous["version"] != 1 or encoded != canonical_json(previous):
+        raise ValueError("Invalid sound enrollment marker")
+      if previous == complete:
+        return False
+      state = previous.get("state")
+      required = {*identity, "state"} | ({"before_sha256"} if state == "pending" else set())
+      if state not in ("pending", "complete") or set(previous) != required:
+        raise ValueError("Invalid sound enrollment marker")
+      if previous["namespace"] != identity["namespace"] or not isinstance(previous["target"], str) or not Path(previous["target"]).is_absolute():
+        raise ValueError("Invalid sound enrollment identity")
+      before = previous.get("before_sha256")
+      if before is not None and (not isinstance(before, str) or len(before) != 64 or any(c not in "0123456789abcdef" for c in before)):
+        raise ValueError("Invalid sound enrollment source")
+      if any(previous[key] != value for key, value in identity.items()):
+        previous = None
+
+    selection = namespace / "SoundPack"
+    try:
+      raw = _read_value(selection)
+    except FileNotFoundError:
+      raw = None
+    current = digest(raw) if raw is not None else None
+    if previous is None:
+      previous = {**identity, "state": "pending", "before_sha256": current}
+      _atomic_write(marker, canonical_json(previous))
+    elif current != previous["before_sha256"] and raw != b"starpilot":
+      # A later choice takes precedence over an interrupted startup enrollment.
+      _atomic_write(marker, canonical_json(complete))
+      return False
+
+    if raw != b"starpilot":
+      _write_selection(namespace)
+    if _read_value(selection) != b"starpilot":
+      raise ValueError("Sound enrollment could not verify its selection")
+    _atomic_write(marker, canonical_json(complete))
+    return True

openpilot/starpilot/audio/downloads.py

@@ -0,0 +1,285 @@
+"""Bounded installation of published sound packs while the car is parked."""
+
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import tempfile
+import threading
+import urllib.request
+import uuid
+import zipfile
+
+from openpilot.starpilot.audio.sound_pack import REMOVED_PACKS, cleanup_removed_packs, LEGACY_NAMES, MAX_WAV_BYTES, PACK_ROOT, pack_directory, read_wav
+
+
+ORIGIN = "https://huggingface.co/buckets/StarPilot-Driving/StarPilot-Resources/resolve/"
+CATALOG_PATH = Path(__file__).with_name("catalog.json")
+MAX_ARCHIVE_BYTES = 64 * 1024 * 1024
+MAX_FILES = 32
+MAX_EXPANDED_BYTES = 96 * 1024 * 1024
+CHUNK = 64 * 1024
+ALLOWED_FILES = {f"{name}.wav" for name in (
+  "engage", "disengage", "refuse", "warning", "dm_warning", "pre_alert", "critical", "dm_critical", "prompt_repeat",
+  "distracted", "engage_tizi", "disengage_tizi",
+)} | set(LEGACY_NAMES.values())
+SLUG = re.compile(r"[a-z0-9][a-z0-9_-]{0,63}\Z")
+SHA256 = re.compile(r"[0-9a-f]{64}\Z")
+
+
+class SoundDownloadError(Exception):
+  def __init__(self, message: str, status: int = 400):
+    super().__init__(message)
+    self.status = status
+
+
+class _Cancelled(Exception):
+  pass
+
+
+class SoundDownloads:
+  def __init__(self, parked, root: Path = PACK_ROOT, catalog_path: Path = CATALOG_PATH, opener=None):
+    self.parked = parked
+    self.root = Path(root)
+    self.opener = opener or urllib.request.urlopen
+    self._lock = threading.Lock()
+    self._cancel = threading.Event()
+    self._closed = False
+    self._thread = None
+    self._lock_file = None
+    self._job: dict | None = None
+    try:
+      with Path(catalog_path).open("rb") as source:
+        raw = source.read(65537)
+      if len(raw) > 65536:
+        raise ValueError("catalog too large")
+      catalog = json.loads(raw)
+      if not isinstance(catalog, dict) or catalog.get("version") != 1 or not isinstance(catalog.get("packs"), list) or len(catalog["packs"]) > 64:
+        raise ValueError("unsupported catalog")
+      packs = {}
+      for pack in catalog["packs"]:
+        slug = pack["id"]
+        if (not isinstance(slug, str) or not SLUG.fullmatch(slug) or slug == "stock" or slug in REMOVED_PACKS or slug in packs or
+            pack["path"] != f"theme/Themes/{slug}/sounds.zip" or
+            not isinstance(pack["name"], str) or not 0 < len(pack["name"]) <= 128 or
+            type(pack["size"]) is not int or not 0 < pack["size"] <= MAX_ARCHIVE_BYTES or
+            not isinstance(pack["sha256"], str) or not SHA256.fullmatch(pack["sha256"])):
+          raise ValueError("invalid catalog entry")
+        packs[slug] = pack
+      self._packs = packs
+      cleanup_removed_packs(self.root, parked=self._safe_parked)
+    except (OSError, ValueError, KeyError, TypeError) as exc:
+      raise SoundDownloadError("sound catalog unavailable", 503) from exc
+
+  def snapshot(self) -> dict:
+    with self._lock:
+      return self._snapshot()
+
+  def _snapshot(self) -> dict:
+    try:
+      parked = bool(self.parked())
+    except Exception:
+      parked = False
+    return {
+      "parked": parked,
+      "packs": [{"id": p["id"], "name": p["name"], "installed": pack_directory(p["id"], self.root) is not None}
+                for p in self._packs.values()],
+      "job": self._job.copy() if self._job else None,
+    }
+
+  def action(self, command: str, data: dict) -> dict:
+    cleanup_removed_packs(self.root, parked=self._safe_parked)
+    if not isinstance(data, dict):
+      raise SoundDownloadError("invalid request")
+    with self._lock:
+      if self._closed:
+        raise SoundDownloadError("downloads are unavailable", 503)
+      if command == "cancel":
+        if set(data) != {"job"}:
+          raise SoundDownloadError("invalid request")
+        if not self._job or data.get("job") != self._job["id"] or self._job["state"] not in ("downloading", "verifying"):
+          raise SoundDownloadError("download job not active", 409)
+        self._cancel.set()
+      elif command == "download":
+        if set(data) != {"pack"}:
+          raise SoundDownloadError("invalid request")
+        slug = data.get("pack")
+        if not isinstance(slug, str) or slug not in self._packs:
+          raise SoundDownloadError("unknown sound pack")
+        if not self._safe_parked():
+          raise SoundDownloadError("park the car before downloading", 409)
+        if self._thread and self._thread.is_alive():
+          raise SoundDownloadError("a download is already running", 409)
+        self._prepare_root()
+        if (self.root / slug).exists() or (self.root / slug).is_symlink():
+          raise SoundDownloadError("sound pack already exists", 409)
+        try:
+          descriptor = os.open(self.root / ".sound-download.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+          if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+            os.close(descriptor)
+            raise OSError("unsafe lock file")
+          lock_file = os.fdopen(descriptor, "r+b")
+        except OSError as exc:
+          raise SoundDownloadError("sound storage unavailable", 503) from exc
+        try:
+          fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+          lock_file.close()
+          raise SoundDownloadError("a download is already running", 409) from exc
+        if (self.root / slug).exists() or (self.root / slug).is_symlink():
+          fcntl.flock(lock_file, fcntl.LOCK_UN)
+          lock_file.close()
+          raise SoundDownloadError("sound pack already exists", 409)
+        self._lock_file = lock_file
+        self._cancel.clear()
+        self._job = {"id": uuid.uuid4().hex, "pack": slug, "state": "downloading", "bytes": 0,
+                     "total": self._packs[slug]["size"], "error": None}
+        self._thread = threading.Thread(target=self._run, args=(self._packs[slug],), daemon=True)
+        self._thread.start()
+        return self._snapshot()
+      else:
+        raise SoundDownloadError("unknown action")
+    return self.snapshot()
+
+  def _safe_parked(self) -> bool:
+    try:
+      return bool(self.parked())
+    except Exception:
+      return False
+
+  def _prepare_root(self):
+    try:
+      if self.root.is_symlink():
+        raise OSError("unsafe pack directory")
+      self.root.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+      raise SoundDownloadError("sound storage unavailable", 503) from exc
+
+  def _check_active(self):
+    if self._cancel.is_set() or not self._safe_parked():
+      raise _Cancelled()
+
+  def _run(self, pack: dict):
+    job = self._job
+    assert job is not None
+    stage = None
+    try:
+      stage = Path(tempfile.mkdtemp(prefix=".sound-download-", dir=self.root))
+      archive = stage / "archive.zip"
+      digest = hashlib.sha256()
+      with self.opener(ORIGIN + pack["path"], timeout=20) as response:
+        with archive.open("wb") as target:
+          remaining = pack["size"]
+          while remaining:
+            self._check_active()
+            chunk = response.read(min(CHUNK, remaining + 1))
+            if not chunk or len(chunk) > remaining:
+              raise ValueError("archive length mismatch")
+            target.write(chunk)
+            digest.update(chunk)
+            remaining -= len(chunk)
+            with self._lock:
+              job["bytes"] = pack["size"] - remaining
+          self._check_active()
+          if response.read(1):
+            raise ValueError("archive length mismatch")
+      with self._lock:
+        job["state"] = "verifying"
+      if digest.hexdigest() != pack["sha256"]:
+        raise ValueError("archive checksum mismatch")
+      sounds = stage / "sounds"
+      sounds.mkdir()
+      self._extract(archive, sounds)
+      archive.unlink()
+      with self._lock:
+        self._check_active()
+        if (self.root / pack["id"]).exists() or (self.root / pack["id"]).is_symlink():
+          raise ValueError("sound pack already exists")
+        os.rename(stage, self.root / pack["id"])
+        stage = None
+        job["state"] = "complete"
+    except _Cancelled:
+      with self._lock:
+        job["state"] = "cancelled"
+    except Exception as exc:
+      with self._lock:
+        if self._cancel.is_set() or not self._safe_parked():
+          job["state"] = "cancelled"
+        else:
+          job["state"] = "failed"
+          job["error"] = str(exc) or "download failed"
+    finally:
+      with self._lock:
+        lock_file, self._lock_file = self._lock_file, None
+      if stage is not None:
+        shutil.rmtree(stage, ignore_errors=True)
+      if lock_file is not None:
+        fcntl.flock(lock_file, fcntl.LOCK_UN)
+        lock_file.close()
+
+  def _extract(self, archive: Path, sounds: Path):
+    with zipfile.ZipFile(archive) as source:
+      members = source.infolist()
+      if not 0 < len(members) <= MAX_FILES:
+        raise ValueError("invalid sound archive")
+      wrapped = any(member.filename == "sounds/" for member in members)
+      raw_names = set()
+      names = set()
+      metadata = set()
+      total = 0
+      audio = []
+      for member in members:
+        self._check_active()
+        name = member.filename
+        mode = (member.external_attr >> 16) & 0xffff
+        if (name in raw_names or member.flag_bits & 1 or
+            member.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED)):
+          raise ValueError("invalid sound archive member")
+        raw_names.add(name)
+        if (name == "sounds/" and wrapped and member.is_dir() and member.file_size == 0 and
+            (mode == 0 or stat.S_IFMT(mode) in (0, stat.S_IFDIR))):
+          continue
+        if wrapped and re.fullmatch(r"__MACOSX/sounds/\._[A-Za-z0-9_]+\.wav", name):
+          if member.is_dir() or member.file_size > 512 or (mode and stat.S_IFMT(mode) not in (0, stat.S_IFREG)):
+            raise ValueError("invalid sound archive member")
+          metadata.add(name.removeprefix("__MACOSX/sounds/._"))
+          continue
+        filename = name.removeprefix("sounds/") if wrapped else name
+        if (name != ("sounds/" + filename if wrapped else filename) or filename not in ALLOWED_FILES or
+            filename in names or member.is_dir() or (mode and stat.S_IFMT(mode) not in (0, stat.S_IFREG)) or
+            member.file_size <= 0 or member.file_size > MAX_WAV_BYTES):
+          raise ValueError("invalid sound archive member")
+        names.add(filename)
+        audio.append((member, filename))
+        total += member.file_size
+        if total > MAX_EXPANDED_BYTES:
+          raise ValueError("sound archive too large")
+      if not audio:
+        raise ValueError("sound archive contains no WAV files")
+      if not metadata <= names:
+        raise ValueError("invalid sound archive metadata")
+      for member, filename in audio:
+        with source.open(member) as data, (sounds / filename).open("xb") as output:
+          remaining = member.file_size
+          while remaining:
+            self._check_active()
+            chunk = data.read(min(CHUNK, remaining + 1))
+            if not chunk or len(chunk) > remaining:
+              raise ValueError("sound file length mismatch")
+            output.write(chunk)
+            remaining -= len(chunk)
+          if data.read(1):
+            raise ValueError("sound file length mismatch")
+        read_wav(sounds / filename)
+
+  def close(self):
+    with self._lock:
+      self._closed = True
+      self._cancel.set()
+      thread = self._thread
+    if thread is not None:
+      thread.join(timeout=2)

openpilot/starpilot/audio/native/SConscript

@@ -0,0 +1,6 @@
+Import('env', 'arch')
+
+if arch == "comma_arm64":
+  native_audio = env.Clone()
+  native_audio.Append(CFLAGS=['-std=c11'])
+  native_audio.SharedLibrary('audio_mixer', ['audio_mixer.c'])

openpilot/starpilot/audio/native/audio_mixer.c

@@ -0,0 +1,115 @@
+// Bounded mono float32 alert renderer. PCM allocation/publication/reclamation occur on the control thread only.
+#define _GNU_SOURCE
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <math.h>
+
+#define SLOTS 16
+#define MAX_PCM 1440000U
+#define MAX_FRAMES 4096U
+_Static_assert(ATOMIC_LONG_LOCK_FREE == 2 && ATOMIC_LLONG_LOCK_FREE == 2 && ATOMIC_POINTER_LOCK_FREE == 2, "Audio requires lock-free atomics");
+struct pa_time { double input_adc, current, output_dac; };
+struct bank { uint64_t revision; unsigned lengths[SLOTS]; float *pcm[SLOTS]; };
+struct audio {
+  struct bank *active;
+  _Atomic(struct bank *) pending, retired;
+  _Atomic uint64_t command_seq, command[8], status_seq, status[10];
+  uint64_t applied_seq, restart, frame, stop_frame;
+  uint64_t status_count, underflows, last_flags;
+  unsigned alert, sound;
+  int loops;
+  float gain;
+  bool pending_stop;
+  atomic_flag rendering;
+};
+static void free_bank(struct bank *b) { if (b) { for(unsigned i=0;i<SLOTS;i++) free(b->pcm[i]); free(b); } }
+unsigned audio_mixer_abi(void) { return 3; }
+struct audio *audio_create(void) { struct audio *a=calloc(1,sizeof(*a)); if(a) { *a=(struct audio){.rendering=ATOMIC_FLAG_INIT,.loops=-1}; } return a; }
+void audio_collect(struct audio *a) { free_bank(atomic_exchange(&a->retired,NULL)); }
+int audio_can_publish(struct audio *a) { return a && atomic_load(&a->pending)==NULL; }
+int audio_publish_bank(struct audio *a,const float *const *pcm,const unsigned *lengths,unsigned count,uint64_t revision) {
+  if(!a || !revision || count>SLOTS || atomic_load(&a->pending)) return 0;
+  audio_collect(a);
+  struct bank *b=calloc(1,sizeof(*b)); if(!b) return 0; b->revision=revision;
+  for(unsigned i=0;i<count;i++) {
+    if(!lengths[i] || lengths[i]>MAX_PCM || !pcm[i]) { free_bank(b); return 0; }
+    b->pcm[i]=malloc((size_t)lengths[i]*sizeof(float)); if(!b->pcm[i]) {free_bank(b);return 0;}
+    memcpy(b->pcm[i],pcm[i],(size_t)lengths[i]*sizeof(float)); b->lengths[i]=lengths[i];
+  }
+  atomic_store(&a->pending,b); return 1;
+}
+uint64_t audio_command(struct audio *a,unsigned alert,unsigned sound,int loops,int stop,float gain,uint64_t restart,uint64_t bank_revision,uint64_t stop_frame) {
+  if(!a || sound>=SLOTS || loops < -1 || loops==0 || !isfinite(gain) || gain<0 || gain>1) return 0;
+  uint32_t bits;memcpy(&bits,&gain,sizeof(bits));
+  uint64_t seq=atomic_load(&a->command_seq);atomic_store(&a->command_seq,seq+1);
+  uint64_t values[8]={alert,sound,(uint64_t)(int64_t)loops,stop!=0,bits,restart,bank_revision,stop_frame};
+  for(unsigned i=0;i<8;i++) atomic_store(&a->command[i],values[i]);
+  atomic_store(&a->command_seq,seq+2); return seq+2;
+}
+static void accept_command(struct audio *a) {
+  uint64_t seq=atomic_load(&a->command_seq),v[8];
+  if(seq==a->applied_seq || (seq&1)) return;
+  for(unsigned i=0;i<8;i++) v[i]=atomic_load(&a->command[i]);
+  if(seq!=atomic_load(&a->command_seq)) return;
+  if(!a->active || v[6]!=a->active->revision) {
+    struct bank *next=atomic_load(&a->pending);
+    if(!next || next->revision!=v[6] || atomic_load(&a->retired)) return;
+    next=atomic_exchange(&a->pending,NULL);
+    struct bank *old=a->active;a->active=next;atomic_store(&a->retired,old);
+  }
+  if(v[5]!=a->restart) { a->frame=0;a->restart=v[5]; }
+  a->alert=(unsigned)v[0];a->sound=(unsigned)v[1];a->loops=(int)(int64_t)v[2];
+  a->pending_stop=v[3]!=0;a->stop_frame=v[7];
+  // The main thread may publish dismissal after this cursor has already crossed
+  // its observed loop endpoint. Never replay an additional loop in that case.
+  if(a->pending_stop && a->frame>=a->stop_frame) { a->alert=0;a->pending_stop=false; }
+  uint32_t bits=(uint32_t)v[4];memcpy(&a->gain,&bits,sizeof(bits));a->applied_seq=seq;
+}
+// Initialization only: caller has not yet opened/started a PortAudio stream.
+// Admit the initial bank without rendering samples or publishing fake timing.
+int audio_prepare(struct audio *a) {
+  if(!a) return 0;
+  accept_command(a);
+  return a->active!=NULL && a->applied_seq!=0;
+}
+static int audio_callback(const void *input,void *output,unsigned long frames,const struct pa_time *t,unsigned long flags,void *user) {
+  (void)input;(void)t;struct audio *a=user;
+  if(!a || !output || frames>MAX_FRAMES || atomic_flag_test_and_set(&a->rendering)) return 2;
+  accept_command(a);
+  float *out=output;memset(out,0,frames*sizeof(float));
+  unsigned length=a->active?a->active->lengths[a->sound]:0;
+  if(a->alert && !length) { atomic_flag_clear(&a->rendering);return 2; }
+  unsigned written=0;
+  while(a->alert && written<frames && (a->loops<0 || a->frame/length<(uint64_t)a->loops)) {
+    unsigned pos=(unsigned)(a->frame%length),n=length-pos;
+    if(n>frames-written)n=(unsigned)frames-written;
+    memcpy(out+written,a->active->pcm[a->sound]+pos,n*sizeof(float));written+=n;a->frame+=n;
+    if(a->pending_stop && a->frame%length==0) { a->alert=0;a->pending_stop=false;break; }
+  }
+  for(unsigned i=0;i<frames;i++) out[i]*=a->gain;
+  // Only detached stream health and playback cursor cross the callback boundary.
+  uint64_t status_seq=atomic_load(&a->status_seq);atomic_store(&a->status_seq,status_seq+1);
+  if(flags) { a->status_count++;a->last_flags=flags; }
+  if(flags&4U) a->underflows++;
+  atomic_store(&a->status[0],a->frame);atomic_store(&a->status[1],a->alert);atomic_store(&a->status[2],a->pending_stop);
+  atomic_store(&a->status[3],a->status_count);atomic_store(&a->status[4],a->underflows);atomic_store(&a->status[5],a->last_flags);
+  atomic_store(&a->status[6],1U);atomic_store(&a->status[7],a->applied_seq);
+  atomic_store(&a->status[8],a->restart);atomic_store(&a->status[9],a->sound);
+  atomic_store(&a->status_seq,status_seq+2);
+  atomic_flag_clear(&a->rendering);return 0;
+}
+uintptr_t audio_callback_address(void) { return (uintptr_t)&audio_callback; }
+int audio_status(struct audio *a,uint64_t *out) {
+  if(!a || !out) return 0;
+  for(unsigned attempt=0;attempt<3;attempt++) {
+    uint64_t seq=atomic_load(&a->status_seq);if(seq&1)continue;
+    for(unsigned i=0;i<10;i++)out[i]=atomic_load(&a->status[i]);
+    if(seq==atomic_load(&a->status_seq))return 1;
+  }
+  return 0;
+}
+// Caller MUST stop/close PortAudio before destroying this stream-owned mixer.
+void audio_destroy(struct audio *a) { if(a) {free_bank(a->active);free_bank(atomic_load(&a->pending));free_bank(atomic_load(&a->retired));free(a);} }

openpilot/starpilot/audio/sound_pack.py

@@ -0,0 +1,290 @@
+"""Installed alert WAV packs."""
+
+from pathlib import Path
+import re
+import os
+import stat
+import wave
+
+import numpy as np
+
+PACK_ROOT = Path("/data/themes/theme_packs")
+DEFAULT_PACK = "starpilot"
+BUILTIN_FILES = {
+  "engage.wav": "engage.wav", "disengage.wav": "disengage.wav",
+  "refuse.wav": "warning_1.wav", "warning.wav": "warning_1.wav", "pre_alert.wav": "warning_1.wav",
+  "dm_warning.wav": "warning_2.wav", "critical.wav": "warning_2.wav", "dm_critical.wav": "warning_3.wav",
+}
+REMOVED_PACKS = frozenset(("frog", "frogpilot", "stalin"))
+MAX_WAV_BYTES = 48000 * 2 * 30 + 4096
+LEGACY_NAMES = {
+  "warning.wav": "prompt.wav", "dm_warning.wav": "prompt_distracted.wav",
+  "critical.wav": "warning_soft.wav", "dm_critical.wav": "warning_immediate.wav",
+}
+
+
+def pack_directory(name: str, root: Path = PACK_ROOT) -> Path | None:
+  if name in (DEFAULT_PACK, "stock") or name.casefold() in REMOVED_PACKS or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_~.-]{0,127}", name):
+    return None
+  directory = root / name / "sounds"
+  try:
+    if any(path.is_symlink() for path in (root, root / name, directory)) or not directory.is_dir():
+      return None
+    return directory
+  except OSError:
+    return None
+
+
+def installed_packs(root: Path = PACK_ROOT) -> tuple[str, ...]:
+  try:
+    return (DEFAULT_PACK, "stock") + tuple(sorted(path.name for path in root.iterdir() if pack_directory(path.name, root) is not None))
+  except OSError:
+    return (DEFAULT_PACK, "stock")
+
+
+def read_selection(params) -> tuple[bytes | None, str, bool]:
+  try:
+    try:
+      with Path(params.get_param_path("SoundPack")).open("rb") as source:
+        raw = source.read(129)
+    except FileNotFoundError:
+      raw = None
+    if raw is not None and len(raw) > 128:
+      return raw, DEFAULT_PACK, False
+    try:
+      name = DEFAULT_PACK if raw is None else raw.decode("utf-8")
+    except UnicodeError:
+      return raw, DEFAULT_PACK, True
+    if name.casefold() in REMOVED_PACKS:
+      name = DEFAULT_PACK
+    valid = name in (DEFAULT_PACK, "stock") or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_~.-]{0,127}", name) is not None
+    return raw, name if valid else DEFAULT_PACK, True
+  except (OSError, KeyError, TypeError, ValueError, UnicodeError):
+    return None, DEFAULT_PACK, False
+
+
+def migrate_removed_selection(params, root: Path = PACK_ROOT) -> bool:
+  """Persist the fork default when a saved pack no longer identifies a valid choice."""
+  from openpilot.starpilot.saved_document import commit_exact
+  raw, name, readable = read_selection(params)
+  if not readable or raw == DEFAULT_PACK.encode():
+    return False
+  legacy_default = raw is None or raw.strip().lower() in (b"", b"default", b"frog", b"frogpilot")
+  valid_choice = name == "stock" or pack_directory(name, root) is not None
+  if not legacy_default and valid_choice:
+    return False
+  return commit_exact(params, key="SoundPack", max_bytes=128, raw=DEFAULT_PACK.encode(), expected=raw,
+                      authorized=lambda: True, temp_prefix=".sound-pack-migration-").verified
+
+
+def cleanup_removed_packs(root: Path = PACK_ROOT, *, parked) -> tuple[str, ...]:
+  """Bounded parked migration over captured directory descriptors.
+
+  Enumerate at most 256 root entries and 64 pack entries, depth 4 / 96 MiB.
+  Keep captured descriptors, never follow links, and check entry identities
+  before mutations. External renames between check and syscall remain a POSIX
+  race; symlink targets are never traversed.
+  Concurrent growth/revocation leaves remaining obsolete files for a later retry.
+  """
+  def safe():
+    try:
+      return bool(parked())
+    except Exception:
+      return False
+
+  def inode(value):
+    return value.st_dev, value.st_ino
+
+  def revision(value):
+    return inode(value), value.st_mtime_ns, value.st_ctime_ns
+
+  removed, opened = [], []
+  try:
+    if not safe() or root.is_symlink():
+      return ()
+    root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    opened.append(root_fd)
+    root_identity = inode(os.fstat(root_fd))
+    names = []
+    with os.scandir(root_fd) as entries:
+      for entry in entries:
+        if len(names) == 256:
+          return ()
+        names.append(entry.name)
+    for name in names:
+      if name.casefold() not in REMOVED_PACKS or not safe():
+        continue
+      expected = os.stat(name, dir_fd=root_fd, follow_symlinks=False)
+
+      def guard(name=name, expected=expected):
+        return (safe() and inode(root.lstat()) == root_identity and
+                inode(os.stat(name, dir_fd=root_fd, follow_symlinks=False)) == inode(expected))
+
+      if stat.S_ISLNK(expected.st_mode):
+        if guard():
+          os.unlink(name, dir_fd=root_fd)
+          removed.append(name)
+        continue
+      if not stat.S_ISDIR(expected.st_mode):
+        continue
+      pack_fds, budget = [], [0, 0]
+      try:
+        pack_fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd)
+        pack_fds.append(pack_fd)
+        if inode(os.fstat(pack_fd)) != inode(expected):
+          continue
+
+        def plan(directory_fd, depth, *, budget=budget, pack_fds=pack_fds, guard=guard):
+          if depth > 4 or not guard():
+            raise ValueError('cleanup depth or authority')
+          node = [directory_fd, revision(os.fstat(directory_fd)), []]
+          with os.scandir(directory_fd) as entries:
+            for entry in entries:
+              budget[0] += 1
+              if budget[0] > 64:
+                raise ValueError('cleanup entry bound')
+              saved = os.stat(entry.name, dir_fd=directory_fd, follow_symlinks=False)
+              child = None
+              if stat.S_ISDIR(saved.st_mode):
+                child_fd = os.open(entry.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd)
+                pack_fds.append(child_fd)
+                if inode(os.fstat(child_fd)) != inode(saved):
+                  raise ValueError('cleanup inode changed')
+                child = plan(child_fd, depth + 1)
+              elif not (stat.S_ISREG(saved.st_mode) or stat.S_ISLNK(saved.st_mode)):
+                raise ValueError('cleanup unexpected entry')
+              else:
+                budget[1] += saved.st_size
+                if budget[1] > 96 * 1024 * 1024:
+                  raise ValueError('cleanup byte bound')
+              node[2].append((entry.name, saved, child))
+          if revision(os.fstat(directory_fd)) != node[1]:
+            raise ValueError('cleanup directory changed')
+          return node
+
+        tree = plan(pack_fd, 1)
+
+        def same_entries(directory_fd, expected_names):
+          observed = set()
+          with os.scandir(directory_fd) as entries:
+            for entry in entries:
+              if len(observed) == 64:
+                return False
+              observed.add(entry.name)
+          return observed == expected_names
+
+        def remove(node, *, guard=guard):
+          directory_fd, _, children = node
+          for index, (entry_name, saved, child) in enumerate(children):
+            expected_names = {item[0] for item in children[index:]}
+            if (not guard() or not same_entries(directory_fd, expected_names) or
+                revision(os.fstat(directory_fd)) != node[1]):
+              raise ValueError('cleanup authority or growth')
+            current = os.stat(entry_name, dir_fd=directory_fd, follow_symlinks=False)
+            if revision(current) != revision(saved):
+              raise ValueError('cleanup replacement entry')
+            if child is not None:
+              remove(child)
+              if not guard() or inode(os.stat(entry_name, dir_fd=directory_fd, follow_symlinks=False)) != inode(saved):
+                raise ValueError('cleanup replacement directory')
+              os.rmdir(entry_name, dir_fd=directory_fd)
+            else:
+              os.unlink(entry_name, dir_fd=directory_fd)
+            node[1] = revision(os.fstat(directory_fd))
+          if (not guard() or not same_entries(directory_fd, set()) or
+              revision(os.fstat(directory_fd)) != node[1]):
+            raise ValueError('cleanup authority or growth')
+
+        remove(tree)
+        if guard():
+          os.rmdir(name, dir_fd=root_fd)
+          removed.append(name)
+      except (OSError, ValueError):
+        pass
+      finally:
+        for descriptor in reversed(pack_fds):
+          os.close(descriptor)
+  except (OSError, TypeError, ValueError):
+    pass
+  finally:
+    for descriptor in reversed(opened):
+      os.close(descriptor)
+  return tuple(removed)
+
+
+def read_wav(path: Path) -> np.ndarray:
+  if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_WAV_BYTES:
+    raise ValueError("invalid WAV file")
+  with wave.open(str(path), "rb") as source:
+    frames = source.getnframes()
+    if (source.getnchannels(), source.getsampwidth(), source.getframerate(), source.getcomptype()) != (1, 2, 48000, "NONE"):
+      raise ValueError("invalid WAV format")
+    if not 0 < frames <= 48000 * 30:
+      raise ValueError("invalid WAV duration")
+    raw = source.readframes(frames)
+    if len(raw) != frames * 2:
+      raise ValueError("truncated WAV")
+  result = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / (2**16/2)
+  result.flags.writeable = False
+  return result
+
+
+class SoundPackLoader:
+  def __init__(self, params, stock: Path, root: Path = PACK_ROOT):
+    self.params, self.stock, self.root = params, stock, root
+    self.signature = None
+    self.checked_at = float("-inf")
+    self.stock_sounds: dict[str, np.ndarray] = {}
+    self.builtin_sounds: dict[str, np.ndarray] = {}
+
+  def is_builtin(self, samples: np.ndarray) -> bool:
+    return any(samples is loaded for loaded in self.builtin_sounds.values())
+
+  def refresh(self, filenames: tuple[str, ...], now: float) -> dict[str, np.ndarray] | None:
+    if 0 <= now - self.checked_at < 1.0:
+      return None
+    self.checked_at = now
+    raw, name, readable = read_selection(self.params)
+    directory = pack_directory(name, self.root) if readable else None
+    paths = tuple(directory / candidate for filename in filenames
+                  for candidate in dict.fromkeys((LEGACY_NAMES.get(filename, filename), filename))) if directory else ()
+    paths += tuple(self.stock.parent / "sounds_starpilot" / BUILTIN_FILES[filename]
+                   for filename in filenames if filename in BUILTIN_FILES)
+    metadata = []
+    for path in paths:
+      try:
+        stat = path.lstat()
+        metadata.append((stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns))
+      except OSError:
+        metadata.append(None)
+    signature = (raw, directory, filenames, tuple(metadata))
+    if signature == self.signature:
+      return None
+    result = {}
+    builtin_loaded = {}
+    for filename in filenames:
+      selected = None
+      if (name == DEFAULT_PACK or not readable) and filename in BUILTIN_FILES:
+        builtin = BUILTIN_FILES[filename]
+        try:
+          if builtin not in builtin_loaded:
+            builtin_loaded[builtin] = read_wav(self.stock.parent / "sounds_starpilot" / builtin)
+          selected = builtin_loaded[builtin]
+          self.builtin_sounds[builtin] = selected
+        except (OSError, EOFError, ValueError, wave.Error):
+          # A damaged optional pack must not stop working stock alerts.
+          pass
+      if selected is None:
+        if filename not in self.stock_sounds:
+          self.stock_sounds[filename] = read_wav(self.stock / filename)
+        selected = self.stock_sounds[filename]
+      if directory:
+        for candidate in dict.fromkeys((LEGACY_NAMES.get(filename, filename), filename)):
+          try:
+            selected = read_wav(directory / candidate)
+            break
+          except (OSError, EOFError, ValueError, wave.Error):
+            pass
+      result[filename] = selected
+    self.signature = signature
+    return result

openpilot/starpilot/audio/tests/test_axis_alerts.py

@@ -0,0 +1,226 @@
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pytest
+
+from opendbc.car.structs import car
+from openpilot.cereal import messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.car.card import Car
+from openpilot.selfdrive.ui.soundd import Soundd
+from openpilot.starpilot.aol.intent import AOL_TOGGLE, AolCardIntent, AolSettings
+from openpilot.starpilot.aol.runtime import decide_axes
+from openpilot.starpilot.aol.wire import INTENT_SERVICE, IntentState, decode_intent, encode_intent
+from openpilot.starpilot.audio.axis_alerts import AudibleAlert, AxisAlerts
+
+
+class AudioFixture:
+  def __init__(self):
+    self.sm = messaging.SubMaster(['selfdriveState', 'aolAxisState', INTENT_SERVICE])
+    self.sound = Soundd.__new__(Soundd)
+    self.sound.axis_alerts = AxisAlerts()
+    self.sound.update_alert = Mock()
+    self.now = 1_000_000_000
+    self.sequence = 0
+
+  def publish(self, armed=False, lateral=False, longitudinal=False, native=AudibleAlert.none,
+              *, age_ns=0, valid=True, session='card', sequence=None, acknowledged=True, payload=None,
+              intent_message=None, enabled=None, pause_longitudinal=False, pause_lateral=False,
+              active=False, state_age_ns=0, state_valid=True):
+    self.now += 50_000_000
+    self.sequence += 1
+    stamp = self.now - age_ns
+    if intent_message is None:
+      intent_message = messaging.new_message(INTENT_SERVICE, 0, valid=valid)
+      intent_message.logMonoTime = stamp
+      intent_message.aolIntentWire = payload if payload is not None else encode_intent(IntentState(
+        session, self.sequence if sequence is None else sequence, stamp, stamp, stamp + 200_000_000,
+        False, pause_lateral, pause_longitudinal, True, lateralArmed=armed))
+    else:
+      self.now = intent_message.logMonoTime + 1_000_000
+      stamp = self.now
+    message = messaging.new_message('aolAxisState', valid=True)
+    message.logMonoTime = stamp
+    axis = message.aolAxisState
+    axis.sessionId = 'selfdrived'
+    axis.sequence = self.sequence
+    axis.qualified = True
+    axis.nativeAcknowledged = acknowledged
+    axis.lateralActive, axis.longitudinalActive = lateral, longitudinal
+    axis.desiredLateral, axis.desiredLongitudinal = lateral, longitudinal
+    axis.observedMonoTime = stamp
+    axis.validUntilMonoTime = stamp + 30_000_000
+    state = messaging.new_message('selfdriveState', valid=state_valid)
+    state.logMonoTime = self.now - state_age_ns
+    state.selfdriveState.alertSound = native
+    state.selfdriveState.enabled = longitudinal if enabled is None else enabled
+    state.selfdriveState.active = active
+    self.sm.update_msgs(self.now / 1e9, [message.as_reader(), state.as_reader(), intent_message.as_reader()])
+    with patch('openpilot.selfdrive.ui.soundd.time.monotonic_ns', return_value=self.now):
+      self.sound.get_audible_alert(self.sm)
+    return self.sound.update_alert.call_args.args[0]
+
+
+@pytest.fixture
+def audio():
+  with OpenpilotPrefix():
+    yield AudioFixture()
+
+
+def test_park_lkas_producer_wire_and_sound(audio):
+  owner = AolCardIntent(AolSettings(True, 0.0, AOL_TOGGLE, 0, (0, 0, 0), (0, 0, 0)), explicit_latch=True)
+  published = {}
+  fields = SimpleNamespace(sm=SimpleNamespace(frame=1, all_checks=lambda _: True),
+                         pm=SimpleNamespace(send=lambda service, message: published.__setitem__(service, message)),
+                         CP=car.CarParams.new_message(), car_params_published=True,
+                         slc_replay=False, curve_replay=False, conditional_replay=False,
+                         last_actuators_output=car.CarControl.Actuators.new_message(), can_rcv_cum_timeout_counter=0,
+                         rk=SimpleNamespace(remaining=0), aol_replay=True, aol_qualified=True, aol_sequence=0,
+                         slc_producer_session='card', aol_card_intent=owner, v_cruise_helper=SimpleNamespace(slc_cruise_change=None))
+  card = Car.__new__(Car)
+  card.__dict__.update(vars(fields))
+  cs = car.CarState.new_message(canValid=True, gearShifter=car.CarState.GearShifter.park, standstill=True)
+  for pressed, gear, expected in ((None, 'park', AudibleAlert.none), (True, 'park', AudibleAlert.engage),
+                                  (True, 'park', AudibleAlert.none), (False, 'park', AudibleAlert.none),
+                                  (None, 'drive', AudibleAlert.none), (None, 'park', AudibleAlert.none),
+                                  (True, 'park', AudibleAlert.disengage)):
+    cs.gearShifter = gear
+    cs.buttonEvents = [] if pressed is None else [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=pressed)]
+    owner.update(cs)
+    Car.state_publish(card, cs, None)
+    decoded = decode_intent(published[INTENT_SERVICE].aolIntentWire)
+    assert decoded is not None
+    assert decoded.lateralArmed == owner.allowed_latch
+    assert decoded.allowedLatch == (owner.allowed_latch and gear == 'drive')
+    assert audio.publish(intent_message=published[INTENT_SERVICE]) == expected
+
+
+def test_lateral_activation_without_latch_change_is_silent(audio):
+  assert audio.publish() == AudibleAlert.none
+  assert audio.publish(armed=True) == AudibleAlert.engage
+  for lateral in (True, False, True, False):
+    assert audio.publish(armed=True, lateral=lateral) == AudibleAlert.none
+  assert audio.publish() == AudibleAlert.disengage
+
+
+def test_native_latch_revocation_and_one_lkas_rearm_sound(audio):
+  owner = AolCardIntent(AolSettings(True, 0, 0, 0, (0, 0, 0), (0, 0, 0)), explicit_latch=True)
+  state = car.CarState.new_message(canValid=True, gearShifter=car.CarState.GearShifter.drive)
+  owner.update(state, now_ns=100)
+  assert audio.publish(armed=owner.allowed_latch) == AudibleAlert.none
+  state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=True)]
+  owner.update(state, now_ns=200)
+  assert audio.publish(armed=owner.allowed_latch) == AudibleAlert.engage
+  state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=False)]
+  owner.update(state, now_ns=300)
+  audio.publish(armed=owner.allowed_latch)
+  state.buttonEvents = []
+  owner.update(state, now_ns=400, native_rejection_ns=350)
+  assert audio.publish(armed=owner.allowed_latch) == AudibleAlert.disengage
+  state.buttonEvents = [car.CarState.ButtonEvent(type=car.CarState.ButtonEvent.Type.lkas, pressed=True)]
+  owner.update(state, now_ns=500, native_rejection_ns=350)
+  assert owner.allowed_latch
+  assert audio.publish(armed=owner.allowed_latch) == AudibleAlert.engage
+
+
+def test_longitudinal_transitions_without_aol(audio):
+  assert audio.publish() == AudibleAlert.none
+  assert audio.publish(longitudinal=True) == AudibleAlert.engage
+  assert audio.publish(longitudinal=True) == AudibleAlert.none
+  assert audio.publish() == AudibleAlert.disengage
+
+
+@pytest.mark.parametrize('native', [AudibleAlert.warningImmediate, AudibleAlert.warningSoft, AudibleAlert.refuse,
+                                   AudibleAlert.prompt, AudibleAlert.engage, AudibleAlert.disengage])
+def test_native_alert_wins_without_queued_chime(audio, native):
+  audio.publish()
+  assert audio.publish(armed=True, native=native) == native
+  assert audio.publish(armed=True) == AudibleAlert.none
+
+
+def test_native_then_axis_ack_is_one_engagement(audio):
+  audio.publish()
+  assert audio.publish(native=AudibleAlert.engage) == AudibleAlert.engage
+  assert audio.publish(longitudinal=True) == AudibleAlert.none
+  assert audio.publish(armed=True, longitudinal=True) == AudibleAlert.none
+
+
+@pytest.mark.parametrize('invalid', [{'age_ns': 31_000_000}, {'age_ns': -1}, {'valid': False}, {'payload': b'invalid'}])
+def test_invalid_latch_is_silent_and_recovery_does_not_replay(audio, invalid):
+  audio.publish()
+  assert audio.publish(armed=True, **invalid) == AudibleAlert.none
+  assert audio.publish(armed=True) == AudibleAlert.none
+  assert audio.publish() == AudibleAlert.disengage
+
+
+def test_startup_new_session_and_old_sequence_do_not_chime(audio):
+  assert audio.publish(armed=True) == AudibleAlert.none
+  assert audio.publish(armed=False, session='new-card') == AudibleAlert.none
+  assert audio.publish(armed=True, session='new-card', sequence=1) == AudibleAlert.none
+  assert audio.publish(armed=True, session='new-card') == AudibleAlert.engage
+
+
+def test_runtime_native_pending_permission_and_loss(audio):
+  audio.publish()
+  cs = car.CarState.new_message(canValid=True, gearShifter=car.CarState.GearShifter.drive, vEgo=20.0)
+  for requested, allowed, expected in ((False, False, AudibleAlert.none), (True, True, AudibleAlert.engage),
+                                      (True, False, AudibleAlert.disengage)):
+    native = SimpleNamespace(requestedLateral=False, requestedLongitudinal=requested,
+                             lateralAllowed=False, longitudinalAllowed=allowed)
+    result = decide_axes(standard_lateral=False, standard_longitudinal=True,
+                         intent=SimpleNamespace(allowedLatch=False, pauseLateral=False, pauseLongitudinal=False),
+                         native=native, car_state=cs, initialized=True, model_ready=True,
+                         no_entry=False, immediate_disable=False, dm_lockout=False, pause_brake_mps=0.0)
+    assert result.native_acknowledged
+    assert audio.publish(longitudinal=result.longitudinal_active, acknowledged=result.native_acknowledged) == expected
+
+
+def test_actuator_ack_does_not_own_engagement_sound(audio):
+  audio.publish()
+  assert audio.publish(longitudinal=True, enabled=False, acknowledged=False) == AudibleAlert.none
+
+
+def test_mismatched_payload_timestamp_is_silent(audio):
+  audio.publish()
+  stamp = audio.now + 50_000_000
+  payload = encode_intent(IntentState('card', 2, stamp - 1, stamp, stamp + 200_000_000,
+                                       False, False, False, True, lateralArmed=True))
+  assert audio.publish(armed=True, payload=payload) == AudibleAlert.none
+
+
+def test_native_only_subscriber_retains_alert(audio):
+  sm = messaging.SubMaster(['selfdriveState'])
+  state = messaging.new_message('selfdriveState', valid=True)
+  state.selfdriveState.alertSound = AudibleAlert.engage
+  sm.update_msgs(audio.now / 1e9, [state.as_reader()])
+  audio.sound.get_audible_alert(sm)
+  audio.sound.update_alert.assert_called_once_with(AudibleAlert.engage)
+
+
+def test_output_drop_and_recovery_do_not_sound_when_engagement_is_unchanged(audio):
+  audio.publish(enabled=False)
+  assert audio.publish(enabled=True, longitudinal=True) == AudibleAlert.engage
+  for active, acknowledged in ((False, True), (True, True), (False, False), (True, True)):
+    assert audio.publish(enabled=True, longitudinal=active, acknowledged=acknowledged) == AudibleAlert.none
+  assert audio.publish(enabled=False) == AudibleAlert.disengage
+
+
+def test_explicit_longitudinal_pause_and_resume_sound_once(audio):
+  audio.publish(enabled=True, longitudinal=True)
+  assert audio.publish(enabled=True, pause_longitudinal=True) == AudibleAlert.disengage
+  assert audio.publish(enabled=True, pause_longitudinal=True) == AudibleAlert.none
+  assert audio.publish(enabled=True, longitudinal=True) == AudibleAlert.engage
+
+
+@pytest.mark.parametrize('invalid', [{'state_age_ns': 101_000_000}, {'state_age_ns': -1}, {'state_valid': False}])
+def test_standard_lateral_stale_state_and_recovery_do_not_replay(audio, invalid):
+  audio.publish(active=True, enabled=True)
+  assert audio.publish(active=True, enabled=True, **invalid) == AudibleAlert.none
+  assert audio.publish(active=True, enabled=True) == AudibleAlert.none
+  assert audio.publish(active=True, enabled=True, pause_lateral=True) == AudibleAlert.disengage
+
+
+def test_explicit_lateral_pause_remains_known_without_standard_state(audio):
+  audio.publish(active=True, enabled=True)
+  assert audio.publish(active=True, enabled=True, pause_lateral=True, state_valid=False) == AudibleAlert.disengage
+  assert audio.publish(active=True, enabled=True, pause_lateral=True) == AudibleAlert.none

openpilot/starpilot/audio/tests/test_complete_alert_assets.py

@@ -0,0 +1,46 @@
+import ast
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from openpilot.starpilot.audio.sound_pack import BUILTIN_FILES, SoundPackLoader, read_wav
+
+
+def production_filenames():
+  source = Path(__file__).resolve().parents[3] / 'selfdrive/ui/soundd.py'
+  tree = ast.parse(source.read_text())
+  sounds = next(node for node in tree.body if isinstance(node, ast.AnnAssign) and node.target.id == 'sound_list')
+  return tuple(dict.fromkeys(value.elts[0].value for value in sounds.value.values))
+
+
+@pytest.mark.parametrize('selection', [None, 'stock', 'starpilot'])
+def test_complete_production_alerts_preserve_stock_critical_max(tmp_path, selection):
+  class Params:
+    def get_param_path(self, key):
+      return str(tmp_path / key)
+
+  if selection is not None:
+    (tmp_path / 'SoundPack').write_text(selection)
+  stock = Path(__file__).resolve().parents[3] / 'selfdrive/assets/sounds'
+  loader = SoundPackLoader(Params(), stock, tmp_path / 'packs')
+  filenames = production_filenames()
+  assert 'dm_critical_max.wav' in filenames
+  loaded = loader.refresh(filenames, 0)
+  assert loaded is not None
+  assert set(loaded) == set(filenames)
+  for filename in filenames:
+    path = stock.parent / 'sounds_starpilot' / BUILTIN_FILES[filename] if selection != 'stock' and filename in BUILTIN_FILES else stock / filename
+    np.testing.assert_array_equal(loaded[filename], read_wav(path))
+  assert not loader.is_builtin(loaded['dm_critical_max.wav'])
+  assert loader.refresh(filenames, 1) is None
+
+
+def test_required_stock_only_alert_is_not_silently_masked(tmp_path):
+  class Params:
+    def get_param_path(self, key):
+      return str(tmp_path / key)
+
+  loader = SoundPackLoader(Params(), tmp_path / 'missing', tmp_path / 'packs')
+  with pytest.raises(ValueError, match='invalid WAV file'):
+    loader.refresh(('dm_critical_max.wav',), 0)

openpilot/starpilot/audio/tests/test_default_enrollment.py

@@ -0,0 +1,188 @@
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import ast
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.audio import default_enrollment as enrollment
+
+
+class FilesystemParams:
+  def __init__(self, namespace):
+    self.namespace = namespace
+
+  def get_param_path(self):
+    return str(self.namespace)
+
+
+class SoundDefaultEnrollmentTests(unittest.TestCase):
+  def setUp(self):
+    temporary = tempfile.TemporaryDirectory()
+    self.addCleanup(temporary.cleanup)
+    self.root = Path(temporary.name)
+    self.target = self.root / "params" / ".namespace"
+    self.target.mkdir(parents=True)
+    self.namespace = self.target.parent / "d"
+    self.namespace.symlink_to(self.target, target_is_directory=True)
+    self.params = FilesystemParams(self.namespace)
+    self.storage = self.root / "storage"
+
+  def selection(self, value):
+    if value is None:
+      (self.namespace / "SoundPack").unlink(missing_ok=True)
+    else:
+      (self.namespace / "SoundPack").write_bytes(value)
+
+  def enroll(self):
+    return enrollment.enroll_default_sounds(self.params, self.storage)
+
+  def marker(self):
+    return next((self.storage / "sound-default-once").iterdir())
+
+  def test_first_enrollment_replaces_all_selections_and_preserves_volumes(self):
+    for raw in (None, b"", b"default", b"stock", b"custom", b"starpilot", b"frog", b"\xff"):
+      with self.subTest(raw=raw), tempfile.TemporaryDirectory() as storage:
+        self.selection(raw)
+        volume = self.namespace / "EngageVolume"
+        volume.write_bytes(b"37")
+        self.assertTrue(enrollment.enroll_default_sounds(self.params, Path(storage)))
+        self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"starpilot")
+        self.assertEqual(volume.read_bytes(), b"37")
+
+  def test_completed_enrollment_preserves_menu_optout_across_restarts(self):
+    self.enroll()
+    marker_bytes = self.marker().read_bytes()
+    for choice in (b"stock", b"custom", b"another-custom"):
+      self.selection(choice)
+      with patch.object(enrollment, "_write_selection", side_effect=AssertionError("repeat enrollment")):
+        for _ in range(3):
+          self.assertFalse(self.enroll())
+      self.assertEqual((self.namespace / "SoundPack").read_bytes(), choice)
+      self.assertEqual(self.marker().read_bytes(), marker_bytes)
+
+  def test_new_resolved_namespace_at_same_path_enrolls(self):
+    self.enroll()
+    self.selection(b"stock")
+    old_target = self.target
+    self.target = old_target.parent / ".fresh"
+    self.target.mkdir()
+    self.namespace.unlink()
+    self.namespace.symlink_to(self.target, target_is_directory=True)
+    self.selection(b"custom")
+    self.assertTrue(self.enroll())
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"starpilot")
+    self.assertEqual((old_target / "SoundPack").read_bytes(), b"stock")
+
+  def test_failed_pending_marker_does_not_change_selection_and_retries(self):
+    self.selection(b"stock")
+    with patch.object(enrollment, "_atomic_write", side_effect=OSError("disk full")):
+      with self.assertRaises(OSError):
+        self.enroll()
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"stock")
+    self.assertTrue(self.enroll())
+
+  def test_failed_selection_retries_without_marking_complete(self):
+    self.selection(b"custom")
+    with patch.object(enrollment, "_write_selection", side_effect=OSError("disk full")):
+      with self.assertRaises(OSError):
+        self.enroll()
+    self.assertIn(b'"pending"', self.marker().read_bytes())
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"custom")
+    self.assertTrue(self.enroll())
+
+  def test_failed_completion_retries_without_rewriting_soundpack(self):
+    self.selection(b"stock")
+    writer = enrollment._atomic_write
+
+    def fail_complete(path, raw):
+      if b'"complete"' in raw:
+        raise OSError("disk full")
+      writer(path, raw)
+
+    with patch.object(enrollment, "_atomic_write", side_effect=fail_complete):
+      with self.assertRaises(OSError):
+        self.enroll()
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"starpilot")
+    with patch.object(enrollment, "_write_selection", side_effect=AssertionError("already selected")):
+      self.assertTrue(self.enroll())
+    self.selection(b"stock")
+    self.assertFalse(self.enroll())
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"stock")
+
+  def test_changed_choice_during_pending_retry_is_preserved(self):
+    self.selection(b"stock")
+    with patch.object(enrollment, "_write_selection", side_effect=OSError("disk full")):
+      with self.assertRaises(OSError):
+        self.enroll()
+    self.selection(b"custom")
+    self.assertFalse(self.enroll())
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"custom")
+    self.assertFalse(self.enroll())
+
+  def test_concurrent_enrollments_only_select_once(self):
+    self.selection(b"custom")
+    with ThreadPoolExecutor(max_workers=8) as pool:
+      results = list(pool.map(lambda _: self.enroll(), range(16)))
+    self.assertEqual(results.count(True), 1)
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"starpilot")
+
+  def test_selection_stages_outside_namespace_and_cleans_failed_write(self):
+    self.selection(b"stock")
+    replace = enrollment.os.replace
+    observed = []
+
+    def fail_selection(source, destination):
+      if Path(destination) == self.namespace / "SoundPack":
+        observed.append(Path(source))
+        self.assertEqual(Path(source).parent, self.namespace.parent)
+        self.assertEqual({path.name for path in self.target.iterdir()}, {"SoundPack"})
+        raise OSError("replace failure")
+      replace(source, destination)
+
+    with patch.object(enrollment.os, "replace", side_effect=fail_selection):
+      with self.assertRaises(OSError):
+        self.enroll()
+    self.assertEqual(len(observed), 1)
+    self.assertFalse(observed[0].exists())
+    self.assertEqual({path.name for path in self.target.iterdir()}, {"SoundPack"})
+    self.assertTrue(self.enroll())
+
+  def test_corrupt_marker_preserves_optout(self):
+    self.enroll()
+    self.selection(b"stock")
+    self.marker().write_bytes(b"invalid")
+    with self.assertRaises(ValueError):
+      self.enroll()
+    self.assertEqual((self.namespace / "SoundPack").read_bytes(), b"stock")
+
+  def test_symlink_selection_is_not_followed_or_replaced(self):
+    external = self.root / "external"
+    external.write_bytes(b"custom")
+    (self.namespace / "SoundPack").symlink_to(external)
+    with self.assertRaises(OSError):
+      self.enroll()
+    self.assertEqual(external.read_bytes(), b"custom")
+    self.assertTrue((self.namespace / "SoundPack").is_symlink())
+
+  def test_manager_runs_after_admission_before_cleanup_and_warns_on_failure(self):
+    # Inspect the real startup call without importing its hardware/process graph.
+    manager = Path(enrollment.__file__).parents[2] / "system/manager/manager.py"
+    tree = ast.parse(manager.read_text())
+    init = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "manager_init")
+    calls = [node for node in ast.walk(init) if isinstance(node, ast.Call)]
+    def line(name):
+      return next(node.lineno for node in calls if isinstance(node.func, ast.Name) and node.func.id == name)
+    self.assertLess(line("prepare_manager_start"), line("enroll_default_sounds"))
+    cleanup = next(node.lineno for node in calls if isinstance(node.func, ast.Attribute) and node.func.attr == "clear_all")
+    self.assertLess(line("enroll_default_sounds"), cleanup)
+    guard = next(node for node in init.body if isinstance(node, ast.Try) and any(
+      isinstance(call, ast.Call) and isinstance(call.func, ast.Name) and call.func.id == "enroll_default_sounds" for call in ast.walk(node)))
+    assert guard.handlers[0].type is not None
+    self.assertEqual(ast.unparse(guard.handlers[0].type), "(OSError, ValueError)")
+    self.assertTrue(any(isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "warning"
+                        for node in ast.walk(guard.handlers[0])))
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/audio/tests/test_downloads.py

@@ -0,0 +1,214 @@
+import hashlib
+import io
+import json
+from pathlib import Path
+import stat
+import tempfile
+import threading
+import time
+import unittest
+import wave
+import zipfile
+
+from openpilot.starpilot.audio.downloads import ORIGIN, SoundDownloadError, SoundDownloads
+from openpilot.starpilot.audio.sound_pack import SoundPackLoader, read_wav
+
+
+def wav(value=1000):
+  output = io.BytesIO()
+  with wave.open(output, "wb") as sound:
+    sound.setnchannels(1)
+    sound.setsampwidth(2)
+    sound.setframerate(48000)
+    sound.writeframes(value.to_bytes(2, "little", signed=True) * 100)
+  return output.getvalue()
+
+
+def archive(files, modes=None):
+  output = io.BytesIO()
+  with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as target:
+    for name, content in files.items():
+      info = zipfile.ZipInfo(name)
+      info.compress_type = zipfile.ZIP_DEFLATED
+      if modes and name in modes:
+        info.external_attr = modes[name] << 16
+      target.writestr(info, content)
+  return output.getvalue()
+
+
+class Params:
+  def __init__(self, path):
+    self.path = path
+
+  def get_param_path(self, name):
+    return str(self.path)
+
+
+class SoundDownloadTests(unittest.TestCase):
+  def setUp(self):
+    self.tmp = tempfile.TemporaryDirectory()
+    self.addCleanup(self.tmp.cleanup)
+    self.base = Path(self.tmp.name)
+    self.root = self.base / "packs"
+    self.catalog = self.base / "catalog.json"
+    self.parked = [True]
+    self.managers = []
+
+  def manager(self, payload, opener=None, slug="duck", digest=None, size=None):
+    self.catalog.write_text(json.dumps({"version": 1, "packs": [{
+      "id": slug, "name": "Duck", "path": f"theme/Themes/{slug}/sounds.zip",
+      "size": len(payload) if size is None else size,
+      "sha256": hashlib.sha256(payload).hexdigest() if digest is None else digest,
+    }]}))
+    manager = SoundDownloads(lambda: self.parked[0], self.root, self.catalog,
+                             opener=opener or (lambda url, timeout: io.BytesIO(payload)))
+    self.managers.append(manager)
+    self.addCleanup(manager.close)
+    return manager
+
+  def finish(self, manager):
+    deadline = time.monotonic() + 3
+    while time.monotonic() < deadline:
+      state = manager.snapshot()["job"]["state"]
+      if state in ("complete", "cancelled", "failed"):
+        return manager.snapshot()
+      time.sleep(.005)
+    self.fail("download did not finish")
+
+  def test_install_and_loader(self):
+    payload = archive({"sounds/": b"", "sounds/engage.wav": wav(2000),
+                       "sounds/warning_soft.wav": wav(3000),
+                       "__MACOSX/sounds/._engage.wav": b"metadata"})
+    def open_url(url, timeout):
+      self.assertEqual(url, ORIGIN + "theme/Themes/duck/sounds.zip")
+      self.assertGreater(timeout, 0)
+      return io.BytesIO(payload)
+    manager = self.manager(payload, open_url)
+    self.assertEqual(manager.action("download", {"pack": "duck"})["job"]["pack"], "duck")
+    snapshot = self.finish(manager)
+    self.assertEqual(snapshot["job"]["state"], "complete")
+    self.assertTrue(snapshot["packs"][0]["installed"])
+    self.assertEqual(len(read_wav(self.root / "duck" / "sounds" / "engage.wav")), 100)
+    self.assertFalse((self.root / "duck" / "sounds" / "__MACOSX").exists())
+    stock = self.base / "stock"
+    stock.mkdir()
+    (stock / "engage.wav").write_bytes(wav(1000))
+    (stock / "disengage.wav").write_bytes(wav(1000))
+    selected = self.base / "SoundPack"
+    selected.write_bytes(b"duck")
+    sounds = SoundPackLoader(Params(selected), stock, self.root).refresh(("engage.wav", "disengage.wav"), 1)
+    assert sounds is not None
+    self.assertAlmostEqual(sounds["engage.wav"][0], 2000 / 32768)
+    self.assertAlmostEqual(sounds["disengage.wav"][0], 1000 / 32768)
+
+  def test_truncation_and_checksum_fail_without_publish(self):
+    payload = archive({"engage.wav": wav()})
+    for case, stream, digest in (("short", payload[:-1], None), ("checksum", payload, "0" * 64)):
+      with self.subTest(case=case):
+        manager = self.manager(payload, opener=lambda url, timeout, data=stream: io.BytesIO(data), digest=digest)
+        manager.action("download", {"pack": "duck"})
+        self.assertEqual(self.finish(manager)["job"]["state"], "failed")
+        self.assertFalse((self.root / "duck").exists())
+
+  def test_reject_unsafe_and_invalid_zip_members(self):
+    cases = (
+      archive({"../engage.wav": wav()}),
+      archive({"engage.wav": wav()}, {"engage.wav": stat.S_IFLNK | 0o777}),
+      archive({"engage.wav": b"not a wav"}),
+      archive({"sounds/": b"", "sounds/engage.wav": wav(), "other/engage.wav": wav()}),
+    )
+    for payload in cases:
+      with self.subTest(payload=hashlib.sha256(payload).hexdigest()[:8]):
+        manager = self.manager(payload)
+        manager.action("download", {"pack": "duck"})
+        self.assertEqual(self.finish(manager)["job"]["state"], "failed")
+        self.assertFalse((self.root / "duck").exists())
+
+  def test_cancel_and_departure(self):
+    payload = archive({"engage.wav": wav()})
+    started = threading.Event()
+    release = threading.Event()
+
+    class Slow(io.BytesIO):
+      def read(self, size=-1):
+        started.set()
+        release.wait(2)
+        return super().read(size)
+
+    manager = self.manager(payload, opener=lambda url, timeout: Slow(payload))
+    job = manager.action("download", {"pack": "duck"})["job"]["id"]
+    self.assertTrue(started.wait(1))
+    manager.action("cancel", {"job": job})
+    release.set()
+    self.assertEqual(self.finish(manager)["job"]["state"], "cancelled")
+    self.assertFalse((self.root / "duck").exists())
+    manager = self.manager(payload, opener=lambda url, timeout: Slow(payload))
+    release.clear()
+    started.clear()
+    manager.action("download", {"pack": "duck"})
+    self.assertTrue(started.wait(1))
+    self.parked[0] = False
+    release.set()
+    self.assertEqual(self.finish(manager)["job"]["state"], "cancelled")
+
+  def test_conflicts_and_existing_pack_untouched(self):
+    payload = archive({"engage.wav": wav()})
+    blocker = threading.Event()
+    started = threading.Event()
+
+    class Slow(io.BytesIO):
+      def read(self, size=-1):
+        started.set()
+        blocker.wait(2)
+        return super().read(size)
+
+    manager = self.manager(payload, opener=lambda url, timeout: Slow(payload))
+    manager.action("download", {"pack": "duck"})
+    self.assertTrue(started.wait(1))
+    with self.assertRaises(SoundDownloadError) as caught:
+      manager.action("download", {"pack": "duck"})
+    self.assertEqual(caught.exception.status, 409)
+    second = self.manager(payload)
+    with self.assertRaises(SoundDownloadError) as caught:
+      second.action("download", {"pack": "duck"})
+    self.assertEqual(caught.exception.status, 409)
+    blocker.set()
+    self.assertEqual(self.finish(manager)["job"]["state"], "complete")
+    original = (self.root / "duck" / "sounds" / "engage.wav").read_bytes()
+    with self.assertRaises(SoundDownloadError):
+      second.action("download", {"pack": "duck"})
+    self.assertEqual((self.root / "duck" / "sounds" / "engage.wav").read_bytes(), original)
+
+  def test_cancel_after_verification_prevents_publish(self):
+    payload = archive({"engage.wav": wav()})
+    manager = self.manager(payload)
+    verified = threading.Event()
+    release = threading.Event()
+    extract = manager._extract
+
+    def pause_after_extract(source, sounds):
+      extract(source, sounds)
+      verified.set()
+      release.wait(2)
+
+    manager._extract = pause_after_extract
+    job = manager.action("download", {"pack": "duck"})["job"]["id"]
+    self.assertTrue(verified.wait(1))
+    started = time.monotonic()
+    manager.action("cancel", {"job": job})
+    self.assertLess(time.monotonic() - started, .2)
+    release.set()
+    self.assertEqual(self.finish(manager)["job"]["state"], "cancelled")
+    self.assertFalse((self.root / "duck").exists())
+
+  def test_malformed_catalog_is_unavailable(self):
+    for content in ('[]', '{"version":1,"packs":{}}', 'x' * 65537):
+      with self.subTest(content=content[:20]):
+        self.catalog.write_text(content)
+        with self.assertRaises(SoundDownloadError) as caught:
+          SoundDownloads(lambda: True, self.root, self.catalog)
+        self.assertEqual(caught.exception.status, 503)
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/audio/tests/test_soundd_loop_dismissal.py

@@ -0,0 +1,86 @@
+"""Dismissal preserves the current loop's tail; urgent alert changes interrupt it."""
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from openpilot.selfdrive.ui import soundd
+
+
+def daemon_for(samples, *, builtin):
+  daemon = soundd.Soundd.__new__(soundd.Soundd)
+  daemon.current_alert = soundd.AudibleAlert.warningSoft
+  daemon.current_sound = daemon.current_alert
+  daemon.current_sound_frame = 0
+  daemon.current_volume = 1.0
+  daemon.pending_stop = False
+  daemon.loaded_sounds = {soundd.AudibleAlert.warningSoft: samples,
+                          soundd.AudibleAlert.warningImmediate: -samples}
+  daemon.pack_loader = SimpleNamespace(is_builtin=lambda _: builtin)
+  daemon.saved_volumes = {"WarningSoftVolume": 100, "WarningImmediateVolume": 100}
+  return daemon
+
+
+@pytest.mark.parametrize("builtin", [False, True], ids=["stock-or-custom", "builtin-starpilot"])
+def test_first_loop_dismissal_repeated_updates_finish_exact_tail_once(builtin):
+  # Continuous periodic tone with exact zero endpoints. The chosen dismissal is
+  # deliberately away from zero, exposing the old step to silence.
+  samples = np.sin(np.linspace(0., 8. * np.pi, 8193)).astype(np.float32)
+  samples[0] = samples[-1] = 0.
+  daemon = daemon_for(samples, builtin=builtin)
+  consumed = 401
+  np.testing.assert_array_equal(daemon.get_sound_data(consumed), samples[:consumed])
+  assert abs(samples[consumed - 1]) > .5
+  daemon.update_alert(soundd.AudibleAlert.none)
+  daemon.update_alert(soundd.AudibleAlert.none)
+  assert daemon.pending_stop and daemon.current_alert == soundd.AudibleAlert.warningSoft
+  remaining = len(samples) - consumed
+  output = daemon.get_sound_data(remaining + 32)
+  np.testing.assert_array_equal(output[:remaining], samples[consumed:])
+  np.testing.assert_array_equal(output[remaining:], np.zeros(32, dtype=np.float32))
+  assert daemon.current_sound_frame == len(samples)
+  assert daemon.current_alert == soundd.AudibleAlert.none and not daemon.pending_stop
+  assert not np.any(daemon.get_sound_data(len(samples)))
+
+
+@pytest.mark.parametrize("builtin", [False, True], ids=["stock-or-custom", "builtin-starpilot"])
+def test_urgent_replacement_interrupts_pending_tail_immediately(builtin):
+  samples = np.linspace(0., 1., 8193, dtype=np.float32)
+  daemon = daemon_for(samples, builtin=builtin)
+  daemon.get_sound_data(400)
+  daemon.update_alert(soundd.AudibleAlert.none)
+  assert daemon.pending_stop
+  daemon.update_alert(soundd.AudibleAlert.warningImmediate)
+  assert not daemon.pending_stop and daemon.current_sound_frame == 0
+  assert daemon.current_alert == soundd.AudibleAlert.warningImmediate
+  np.testing.assert_array_equal(daemon.get_sound_data(128), -samples[:128])
+
+
+@pytest.mark.parametrize("builtin", [False, True], ids=["stock-or-custom", "builtin-starpilot"])
+@pytest.mark.parametrize("loops", [0, 1, 2])
+def test_dismissal_at_completed_boundary_does_not_start_an_extra_loop(builtin, loops):
+  samples = np.linspace(0., 1., 1024, dtype=np.float32)
+  daemon = daemon_for(samples, builtin=builtin)
+  daemon.get_sound_data(len(samples) * loops)
+  assert daemon.current_sound_frame == len(samples) * loops
+  daemon.update_alert(soundd.AudibleAlert.none)
+  assert daemon.current_alert == soundd.AudibleAlert.none and not daemon.pending_stop
+  assert not np.any(daemon.get_sound_data(len(samples)))
+  assert daemon.current_sound_frame == len(samples) * loops
+
+
+@pytest.mark.parametrize("builtin", [False, True], ids=["stock-or-custom", "builtin-starpilot"])
+def test_same_alert_reassertion_cancels_pending_stop_without_restarting(builtin):
+  samples = np.sin(np.linspace(0., 8. * np.pi, 8193)).astype(np.float32)
+  daemon = daemon_for(samples, builtin=builtin)
+  consumed = 401
+  daemon.get_sound_data(consumed)
+  daemon.update_alert(soundd.AudibleAlert.none)
+  assert daemon.pending_stop
+  daemon.update_alert(soundd.AudibleAlert.warningSoft)
+  assert not daemon.pending_stop and daemon.current_sound_frame == consumed
+  remaining = len(samples) - consumed
+  output = daemon.get_sound_data(remaining + 128)
+  np.testing.assert_array_equal(output[:remaining], samples[consumed:])
+  np.testing.assert_array_equal(output[remaining:], samples[:128])
+  assert daemon.current_alert == soundd.AudibleAlert.warningSoft

openpilot/starpilot/audio/tests/test_soundd_output.py

@@ -0,0 +1,108 @@
+"""Actual soundd load, alert service, stream configuration and audio callback."""
+from pathlib import Path
+from types import SimpleNamespace as NS
+from unittest.mock import patch
+
+import numpy as np
+import pytest
+
+from openpilot.common.params import Params
+from openpilot.selfdrive.ui import soundd
+from openpilot.starpilot.audio.sound_pack import DEFAULT_PACK, SoundPackLoader, read_wav
+
+
+@pytest.mark.parametrize("raw", [None, b"", b"default", b"Frog", b"frogpilot", b"missing", b"\xff"])
+def test_default_repair_persists_and_flows_through_actual_stream(tmp_path, raw):
+  params = Params(str(tmp_path / "params"))
+  if raw is not None:
+    Path(params.get_param_path("SoundPack")).write_bytes(raw)
+  daemon = soundd.Soundd(params, pack_root=tmp_path / "packs")
+  assert params.get("SoundPack") == DEFAULT_PACK
+  root = Path(soundd.__file__).resolve().parents[1] / "assets/sounds_starpilot"
+  class Stream:
+    def __init__(self, **kw):
+      self.kw = kw
+  device = NS(_terminate=lambda: None, _initialize=lambda: None, OutputStream=Stream)
+  stream = daemon.get_stream(device)
+  assert stream.kw["samplerate"] == 48000 and stream.kw["channels"] == 1
+  assert stream.kw["blocksize"] == 4096
+  for alert in (soundd.AudibleAlert.engage, soundd.AudibleAlert.disengage,
+                soundd.AudibleAlert.prompt, soundd.AudibleAlert.warningSoft, soundd.AudibleAlert.warningImmediate):
+    daemon.current_alert = soundd.AudibleAlert.none
+    class State:
+      updated = {"selfdriveState": True}
+      logMonoTime = {}
+      valid = {"selfdriveState": True}
+      alive = {"selfdriveState": True}
+      def __init__(self, sound):
+        self.sound = sound
+      def __getitem__(self, service):
+        return NS(alertSound=NS(raw=self.sound))
+    daemon.get_audible_alert(State(alert))
+    daemon.current_volume = .5
+    output = np.zeros((4096, 1), np.float32)
+    with patch("pathlib.Path.open", side_effect=AssertionError("callback file IO")):
+      stream.kw["callback"](output, 4096, None, None)
+    assert np.isfinite(output).all() and np.max(np.abs(output)) <= .5 and np.any(output != 0)
+    logical = soundd.sound_list[alert][0]
+    from openpilot.starpilot.audio.sound_pack import BUILTIN_FILES
+    expected = read_wav(root / BUILTIN_FILES[logical])
+    np.testing.assert_array_equal(output[:, 0], expected[:4096] * .5)
+
+
+def test_warning_urgency_levels_and_loop_finish(tmp_path):
+  daemon = soundd.Soundd(Params(str(tmp_path / "params")), pack_root=tmp_path / "packs")
+  alerts = (soundd.AudibleAlert.promptRepeat, soundd.AudibleAlert.warningSoft, soundd.AudibleAlert.warningImmediate)
+  rms, peak = [], []
+  for alert in alerts:
+    samples = daemon.loaded_sounds[alert]
+    rms.append(float(np.sqrt(np.mean(samples ** 2))))
+    peak.append(float(np.max(np.abs(samples))))
+    assert np.isfinite(samples).all() and 0 < peak[-1] < 1
+    assert samples[0] == 0 and samples[-1] == 0
+    daemon.current_alert, daemon.current_sound_frame, daemon.current_volume = alert, 0, 1.
+    output = np.zeros((len(samples) * 2 + 4096, 1), np.float32)
+    daemon.callback(output, len(output), None, None)
+    np.testing.assert_array_equal(output[:len(samples), 0], samples)
+    np.testing.assert_array_equal(output[len(samples):2*len(samples), 0], samples)
+    daemon.update_alert(soundd.AudibleAlert.none)
+    tail = np.zeros((len(samples), 1), np.float32)
+    daemon.callback(tail, len(tail), None, None)
+    assert daemon.current_alert == soundd.AudibleAlert.none
+    output.fill(1)
+    daemon.callback(output, len(output), None, None)
+    assert not np.any(output)
+  assert rms[0] < rms[1] < rms[2] and peak[0] < peak[1] < peak[2]
+
+
+def test_builtin_loading_does_not_require_stock_and_damaged_clip_uses_stock(tmp_path):
+  from openpilot.starpilot.audio.sound_pack import BUILTIN_FILES
+  params = Params(str(tmp_path / "params"))
+  params.put("SoundPack", DEFAULT_PACK, block=True)
+  actual = Path(soundd.__file__).resolve().parents[1] / "assets"
+  stock = tmp_path / "assets/sounds"
+  builtin = stock.parent / "sounds_starpilot"
+  builtin.mkdir(parents=True)
+  for name in set(BUILTIN_FILES.values()):
+    (builtin / name).write_bytes((actual / "sounds_starpilot" / name).read_bytes())
+  loader = SoundPackLoader(params, stock, tmp_path / "packs")
+  loaded = loader.refresh(("engage.wav", "warning.wav"), 0)
+  assert np.any(loaded["engage.wav"])
+  stock.mkdir()
+  (stock / "engage.wav").write_bytes((actual / "sounds/engage.wav").read_bytes())
+  (builtin / "engage.wav").write_text("version https://git-lfs.github.com/spec/v1\n")
+  restored = loader.refresh(("engage.wav", "warning.wav"), 1)
+  np.testing.assert_array_equal(restored["engage.wav"], read_wav(stock / "engage.wav"))
+  np.testing.assert_array_equal(restored["warning.wav"], loaded["warning.wav"])
+
+
+@pytest.mark.parametrize("name", ["stock", "custom"])
+def test_valid_explicit_pack_survives_restart(tmp_path, name):
+  params = Params(str(tmp_path / "params"))
+  packs = tmp_path / "packs"
+  custom = packs / "custom/sounds"
+  custom.mkdir(parents=True)
+  params.put("SoundPack", name, block=True)
+  soundd.Soundd(params, pack_root=packs)
+  soundd.Soundd(params, pack_root=packs)
+  assert params.get("SoundPack") == name

openpilot/starpilot/audio/tests/test_stream_diagnostics.py

@@ -0,0 +1,32 @@
+"""Capture transport fault frequency without doing logging in PortAudio's callback."""
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import numpy as np
+
+from openpilot.selfdrive.ui import soundd
+
+
+def test_underflow_counters_preserve_signal_and_log_on_service_thread():
+  daemon = soundd.Soundd.__new__(soundd.Soundd)
+  daemon.pending_stream_status = None
+  daemon.stream_status_count = 0
+  daemon.output_underflow_count = 0
+  samples = np.array([0.5, -0.5, 0., 0.25], dtype=np.float32)
+  daemon.get_sound_data = lambda frames: samples[:frames]
+  output = np.empty((4, 1), dtype=np.float32)
+  underflow = SimpleNamespace(output_underflow=True)
+  other = SimpleNamespace(output_underflow=False)
+  with patch.object(soundd.cloudlog, "warning") as warning, patch.object(soundd.cloudlog, "info") as info:
+    for status in (underflow, underflow, other, None):
+      daemon.callback(output, 4, None, status)
+      np.testing.assert_array_equal(output[:, 0], samples)
+    warning.assert_not_called()
+    info.assert_not_called()
+    assert daemon.stream_status_count == 3
+    assert daemon.output_underflow_count == 2
+    daemon.log_pending_stream_status(SimpleNamespace(latency=0.2, cpu_load=0.01))
+    warning.assert_called_once()
+    info.assert_called_once_with("soundd stream diagnostics: status_callbacks=3 output_underflows=2 latency=0.2 cpu_load=0.01")
+    daemon.log_pending_stream_status(SimpleNamespace(latency=0.2, cpu_load=0.01))
+    assert info.call_count == 1