StarPort: wait out the comma's own Wi-Fi switching networks, and say so on the car screen

- When the comma's Wi-Fi joins or roams while the car is on aa0, wait up to 8 s for video acks instead of 2.5 s; reconnect at once if the car takes its screen back.
- After a drop it causes, the car screen shows "Reconnected: comma Wi-Fi Joined <network>" (the network's SSID) for 8 s; Galaxy and bug reports name it too.
- Logs ack_grace, ack_grace_recovered and other_wifi_changed events.

openpilot/starpilot/system/starport/bugsink_diagnosis.py

@@ -98,6 +98,9 @@ def attempt_records(events):
 
 
 STREAM_LOSS = {
+  "shared_radio": ("comma's own Wi-Fi", "The car stopped acknowledging video while the comma's own Wi-Fi joined or roamed; both share one radio.",
+                   ["Check the other_wifi_changed and ack_grace events for the network and how long acks stopped.",
+                    "Repeated drops at home or on a hotspot mean the 8 s grace is too short or the car takes its screen back first."]),
   "stalled": ("Video acknowledgement older than", "The car kept the link but stopped acknowledging video for too long, so the stream was ended.",
               ["Compare the Wi-Fi signal, frame age and encoder recoveries around the stall.",
                "If it repeats at the same place or speed, look for the comma stalling (CPU, thermal) rather than the car."]),

openpilot/starpilot/system/starport/current_car_ui.py

@@ -188,12 +188,14 @@ def run(frames_path: str, touch_path: str | None = None, control_path: str | Non
     gui_app._width, gui_app._height = geometry.logical_width, geometry.logical_height
     gui_app._scale = 1.0
     gui_app._render_texture = None
+    from openpilot.starpilot.system.starport import reconnect_notice
     from openpilot.starpilot.system.starport.identity import certificate_days_left
     from openpilot.starpilot.system.starport.projection_layout_runtime import ProjectionLayoutWatcher, comma_display_seed
     viewport = (geometry.logical_width, geometry.logical_height)
     watcher = ProjectionLayoutWatcher(viewport, display_seed=comma_display_seed)
     layout = ProjectionOnroad(viewport=viewport, customization=watcher.load(),
-                              certificate_days=certificate_days_left(), layout_watcher=watcher)
+                              certificate_days=certificate_days_left(), layout_watcher=watcher,
+                              reconnect_notice=reconnect_notice.take())
     resources.callback(layout.close)
     watcher.start()
     resources.callback(watcher.close)

openpilot/starpilot/system/starport/network.py

@@ -109,6 +109,37 @@ def recover_autoconnect(log: Callable[..., None], marker: Path = AUTOCONNECT_MAR
   return True
 
 
+SETTINGS_PROFILE_PREFIX = "openpilot connection "  # system/ui/lib/wifi_manager.py names saved networks this way
+
+
+def profile_network_name(profile: str) -> str:
+  """A saved connection's network name: "openpilot connection Home" -> "Home"; other profiles as they are."""
+  return profile[len(SETTINGS_PROFILE_PREFIX):] if profile.startswith(SETTINGS_PROFILE_PREFIX) else profile
+
+
+def wifi_network_name(interface: str = "wlan0") -> str:
+  """The network the comma's own Wi-Fi is on or joining, for telling the driver; "" if none or unknown.
+
+  Its own D-Bus connection: never the car lease's, whose lock the link check and reconnect wait on."""
+  lease = NetworkLease(lambda *args, **kwargs: None, interface)
+  try:
+    for path, name in lease._wifi_devices():
+      if name != interface:
+        continue
+      access_point = lease._get(path, NM_WIRELESS_IFACE, "ActiveAccessPoint")
+      if access_point and access_point != "/":
+        ssid = bytes(lease._get(access_point, NM_AP_IFACE, "Ssid")).decode("utf-8", "replace").strip()
+        if ssid:
+          return ssid
+      # Still joining (no access point yet): the profile's name, which the comma's Wi-Fi settings prefix.
+      active = lease._get(path, NM_DEVICE_IFACE, "ActiveConnection")
+      return profile_network_name(str(lease._get(active, NM_ACTIVE_IFACE, "Id"))) if active and active != "/" else ""
+    return ""
+  finally:
+    if lease.router is not None:
+      lease.router.close()
+
+
 def connection_settings(credentials: WifiCredentials, interface: str) -> dict:
   wireless = {
     "ssid": ("ay", credentials.ssid.encode("utf-8")),

openpilot/starpilot/system/starport/projection_onroad.py

@@ -14,6 +14,7 @@ from openpilot.starpilot.system.starport.projection_map_lease import MapLeaseWri
 CERTIFICATE_NOTICE_NS = 10_000_000_000  # how long the expiry heads-up stays at the start of a drive
 BOOKMARK_NOTICE_NS = 2_000_000_000  # how long "Bookmark Saved" stays after the drive is bookmarked
 MODE_NOTICE_NS = 2_000_000_000  # how long the driving-mode banner stays after the mode changes
+RECONNECT_NOTICE_NS = 8_000_000_000  # how long "why it reconnected" stays once the car screen is back
 DRIVER_MONITOR_HOLD_NS = 350_000_000
 STANDBY_LOGO = Path(__file__).parents[3] / 'selfdrive/assets/images/starpilot_logo.png'
 BUBBLE_REDUNDANT_ALERTS = frozenset(('preLaneChangeLeft', 'preLaneChangeRight', 'laneChange',
@@ -150,12 +151,14 @@ class ProjectionOnroad:
   """Separate renderer with projected actions; native UIState stays read-only."""
 
   def __init__(self, *, dependencies=None, viewport=None, customization=None, certificate_days=None, native_focus=None,
-               layout_watcher=None):
+               layout_watcher=None, reconnect_notice=""):
     viewport = FALLBACK_VIEWPORT if viewport is None else viewport
     self.width, self.height = viewport
     self.customization = customization
     self._certificate_notice = certificate_notice(certificate_days)
     self._certificate_notice_until_ns: int | None = None
+    self._reconnect_notice = reconnect_notice
+    self._reconnect_notice_until_ns: int | None = None
     self._bookmark_notice_until_ns: int | None = None
     self._mode_label: str | None = None  # the drive's last known driving mode
     self._mode_engaged: bool | None = None
@@ -424,6 +427,18 @@ class ProjectionOnroad:
     return replace(state, alert=self.native.alert(size=sizes.SMALL, text1=self._certificate_notice,
                                                   alert_type='androidAutoCertificate/warning'))
 
+  def _with_reconnect_notice(self, state, now_ns):
+    """Say why the car screen just came back (the comma's Wi-Fi switched networks); a real alert always wins."""
+    if not self._reconnect_notice:
+      return state
+    if self._reconnect_notice_until_ns is None:
+      self._reconnect_notice_until_ns = now_ns + RECONNECT_NOTICE_NS
+    sizes = self.native.alert_size
+    if now_ns >= self._reconnect_notice_until_ns or state.alert.size != sizes.NONE:
+      return state
+    return replace(state, alert=self.native.alert(size=sizes.SMALL, text1=self._reconnect_notice,
+                                                  alert_type='androidAutoReconnected/info'))
+
   def bookmark_saved(self, now_ns=None):
     """A bookmark was published for this drive (from any button): confirm it on the car screen."""
     self._bookmark_notice_until_ns = (time.monotonic_ns() if now_ns is None else now_ns) + BOOKMARK_NOTICE_NS
@@ -516,7 +531,8 @@ class ProjectionOnroad:
       # Touch hit-tests read viewport_width; the view draws at the car canvas width (1920 for a 1280x720 head unit).
       if state.viewport_width != self.width:
         state = replace(state, viewport_width=self.width)
-      state = self._with_bookmark_notice(self._with_certificate_notice(state, now_ns), now_ns)
+      state = self._with_certificate_notice(self._with_reconnect_notice(state, now_ns), now_ns)
+      state = self._with_bookmark_notice(state, now_ns)
       self._state = self._with_mode_notice(state, now_ns)
       if self._slc_input is not None:
         self._state = self._slc_input.present(self._state)

openpilot/starpilot/system/starport/reconnect_notice.py

@@ -0,0 +1,51 @@
+"""Why the last projection reconnected, handed from the supervisor to the next car-screen renderer."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+from openpilot.starpilot.system.starport import identity as identity_store
+
+NOTICE_PATH = identity_store.DATA_DIR / "reconnect_notice.json"
+NOTICE_FRESH_S = 120.0  # a notice older than this is about an earlier drive or reconnect (monotonic: same boot only)
+NAME_MAX = 24
+
+
+def shared_radio_text(network: str) -> str:
+  """The car-screen line for a drop caused by the comma's own Wi-Fi joining or roaming."""
+  name = network.strip()
+  if len(name) > NAME_MAX:
+    name = name[:NAME_MAX - 3].rstrip() + "..."  # plain dots: the car-screen fonts may lack an ellipsis glyph
+  return f"Reconnected: comma Wi-Fi Joined {name}" if name else "Reconnected: comma Wi-Fi Was Switching Networks"
+
+
+def write(text: str, path: Path = NOTICE_PATH, now: float | None = None) -> None:
+  try:
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temporary = path.with_suffix(".tmp")
+    temporary.write_text(json.dumps({"text": text, "written": time.monotonic() if now is None else now}))
+    os.replace(temporary, path)
+  except OSError:
+    pass  # the notice is a courtesy; projection never depends on it
+
+
+def take(path: Path = NOTICE_PATH, now: float | None = None) -> str:
+  """Read and remove the pending notice; "" if there is none or it is stale."""
+  try:
+    record = json.loads(path.read_text())
+  except (OSError, ValueError):
+    return ""
+  try:
+    path.unlink()
+  except OSError:
+    pass
+  now = time.monotonic() if now is None else now
+  if not isinstance(record, dict) or not isinstance(record.get("text"), str):
+    return ""
+  written = record.get("written")
+  if not isinstance(written, (int, float)) or not 0 <= now - written <= NOTICE_FRESH_S:
+    return ""
+  return record["text"]

openpilot/starpilot/system/starport/session.py

@@ -116,6 +116,14 @@ class PeerRequestedStop(EOFError):
   pass
 
 
+class SharedRadioStall(TimeoutError):
+  """The car stopped acknowledging video while the comma's other Wi-Fi network was joining or roaming."""
+
+  def __init__(self, message: str, network: str):
+    super().__init__(message)
+    self.network = network
+
+
 START_CODE = re.compile(b"\x00\x00(?:\x00)?\x01")
 
 
@@ -644,9 +652,22 @@ class ProjectionSession(Session):
   # held it ~1 s on 2026-10-01 while the car kept acking in 27-219 ms. Stays under the 3 s after which a Honda takes
   # its screen back without video.
   ACK_TIMEOUT = 2.5
+  # The radio is shared with the comma's own Wi-Fi (aa0 + wlan0): joining or roaming there stalled the car's acks past
+  # 2.5 s on 2026-11-02 (hotspot join, home join, mesh roam). While that happens, wait this long before reconnecting.
+  ACK_GRACE_TIMEOUT = 8.0
   ACK_DRAIN_SECONDS = 0.1  # before calling an ACK late, read what already arrived, for at most this long
   MAX_WINDOW = 2
   nav = None
+  grace_frame: float | None = None  # send time of the frame whose late ack is being waited out
+  grace_network = ""
+  grace_focus_lost = False
+
+  @staticmethod
+  def radio_change() -> str | None:
+    """Replaced by the supervisor: the comma's other Wi-Fi network's name if it joined or roamed in the last few
+    seconds ("" when unnamed), else None."""
+    return None
+
   # HeadunitPad acknowledges every fragment (codec config included) rather than every frame.
   fragment_acks = False
 
@@ -885,6 +906,8 @@ class ProjectionSession(Session):
     granted = focus in (FOCUS_PROJECTED, FOCUS_PROJECTED_NO_INPUT)
     self.focused = granted and self.allow_projection
     self.event("video_focus", focus=focus, unsolicited=unsolicited, focused=self.focused)
+    if was_focused and not self.focused and self.grace_frame is not None:
+      self.grace_focus_lost = True  # the car gave up on the stalled video: reconnect rather than wait unfocused
     if was_focused and not self.focused and self.touch is not None:
       self.touch_events.extend(self.touch.reset())
     if self.focused and not was_focused:
@@ -943,6 +966,9 @@ class ProjectionSession(Session):
     self.unacked -= count
     self.acked += count
     self.epoch_acked += count
+    if self.grace_frame is not None and (not self.pending or self.pending[0] != self.grace_frame):
+      self.event("ack_grace_recovered", network=self.grace_network, gap_ms=round((now - self.grace_frame) * 1000))
+      self.grace_frame = None
     self._confirm_video()
 
   def _confirm_video(self) -> None:
@@ -966,6 +992,8 @@ class ProjectionSession(Session):
   def check_progress(self, now: float | None = None) -> None:
     measured = now is None
     now = time.monotonic() if measured else now
+    if self.grace_focus_lost:
+      raise SharedRadioStall("The car took its screen back while the comma's Wi-Fi was switching networks", self.grace_network)
     if self._ack_overdue(now):
       # ACKs that arrived while this loop was busy (a slow encode, a CPU burst) still sit in the socket; the
       # streaming loop drains only a bounded batch per pass, so read what is there before calling the car silent.
@@ -975,12 +1003,28 @@ class ProjectionSession(Session):
       if measured:
         now = time.monotonic()
       if self._ack_overdue(now):
-        raise TimeoutError(f"Video acknowledgement older than {self.ACK_TIMEOUT:.1f} s")
+        self._wait_out_radio_change(now)
     # Only a receiver known to answer pings can be declared dead by silence; for
     # others a lost link surfaces as a socket error or a dropped Wi-Fi lease.
     if self.ping_responses and now - self.last_rx > 15.0:
       raise TimeoutError("Head unit stopped answering for 15 s")
 
+  def _wait_out_radio_change(self, now: float) -> None:
+    """The oldest frame is past ACK_TIMEOUT: give up, unless the comma's other Wi-Fi just changed networks."""
+    oldest = self.pending[0]
+    if self.grace_frame == oldest:
+      network = self.grace_network
+    else:
+      changed = self.radio_change()
+      if changed is None:
+        raise TimeoutError(f"Video acknowledgement older than {self.ACK_TIMEOUT:.1f} s")
+      self.grace_frame, self.grace_network = oldest, changed
+      self.event("ack_grace", network=changed, after_ms=round((now - oldest) * 1000))
+      network = changed
+    if now - oldest > self.ACK_GRACE_TIMEOUT:
+      message = f"Video acknowledgement older than {self.ACK_GRACE_TIMEOUT:.1f} s while the comma's Wi-Fi was switching networks"
+      raise SharedRadioStall(message, network)
+
   def send_frame(self, data: bytes, timestamp_us: int, *, keyframe: bool) -> None:
     if not self.can_send():
       raise RuntimeError("Cannot send without video focus and window credit")

openpilot/starpilot/system/starport/supervisor.py

@@ -34,6 +34,7 @@ from typing import Any
 
 from openpilot.common.hardware import COMMA_HARDWARE
 from openpilot.starpilot.system.starport import car_profiles, device_health, identity as identity_store, new_setup, reachability, setup_ladder, system_slice
+from openpilot.starpilot.system.starport import reconnect_notice
 from openpilot.starpilot.system.starport.auto_connect import AutoConnectPolicy
 from openpilot.starpilot.system.starport.bootstrap import MODERN_MAJOR, NAMES, BootstrapError, BootstrapTimeout, WirelessBootstrap
 from openpilot.starpilot.system.starport.bugsink_report import PairingAttempt
@@ -42,7 +43,7 @@ from openpilot.starpilot.system.starport.frame_source import (DEFAULT_PATH as DE
 from openpilot.starpilot.system.starport.projection_control import DEFAULT_CONTROL_SOCKET, NATIVE_FOCUS
 from openpilot.starpilot.system.starport.touch import DEFAULT_TOUCH_SOCKET
 from openpilot.starpilot.system.starport.view import CAR_FRAME_PATH, ViewSource, renderer_available
-from openpilot.starpilot.system.starport.session import CODEC_H265, AuthenticationRejected, PeerRequestedStop, ProjectionSession
+from openpilot.starpilot.system.starport.session import CODEC_H265, AuthenticationRejected, PeerRequestedStop, ProjectionSession, SharedRadioStall
 
 BACKOFF_SECONDS = (2.0, 4.0, 8.0, 15.0, 30.0)
 STABLE_SESSION_SECONDS = 30.0
@@ -67,6 +68,8 @@ USB_CONFIGURE_WAIT = 15.0    # slower receivers need time to enumerate after the
 LINK_CHECK_INTERVAL = 1.0    # how often the link watcher asks whether the car's Wi-Fi is still up
 LINK_CHECK_SLOW = 1.0        # a link check slower than this is logged (NetworkManager answers over D-Bus)
 STREAM_STALL_LOG = 1.0       # a streaming loop pass slower than this is logged
+RADIO_CHECK_INTERVAL = 0.5   # how often the comma's own Wi-Fi is checked for a join or roam (a sysfs read)
+RADIO_CHANGE_WINDOW = 15.0   # a join or roam this recent can still stall the car (11-02: stall came 7.5 s after a roam)
 REPORTED_SETTINGS = ("connection", "usb_mode", "view", "encoder", "fps", "bitrate_kbps", "rate_control", "gpu_nv12",
                      "async_readback", "wifi_interface", "dedicated_wifi_interface", "verify_head_unit", "version_status",
                      "rfcomm_channel", "phone_class", "auto_connect", "allow_hevc")
@@ -250,6 +253,60 @@ class LinkWatch:
       self._stop.wait(self.interval)
 
 
+class SharedRadioWatch:
+  """Notices the comma's own Wi-Fi joining or roaming while the car is on aa0; both share one radio.
+
+  The link going down or up bumps carrier_changes, a plain sysfs read. The network's name comes from
+  NetworkManager on this thread, never the streaming one.
+  """
+
+  def __init__(self, log: Callable[..., None], interface: str = "wlan0", interval: float = RADIO_CHECK_INTERVAL,
+               counter: Path | None = None, name: Callable[[str], str] | None = None):
+    self.log, self.interface, self.interval = log, interface, interval
+    if name is None:
+      from openpilot.starpilot.system.starport.network import wifi_network_name as name
+    self.name = name
+    self.counter = counter or Path(f"/sys/class/net/{interface}/carrier_changes")
+    self.changed_at: float | None = None
+    self.network = ""
+    self._stop = threading.Event()
+    self._thread = threading.Thread(target=self._run, name="aa_radio_watch", daemon=True)
+
+  def start(self) -> SharedRadioWatch:
+    self._thread.start()
+    return self
+
+  def stop(self) -> None:
+    self._stop.set()
+
+  def recent(self, now: float | None = None) -> str | None:
+    """The other network's name if it changed within RADIO_CHANGE_WINDOW ("" if unnamed), else None."""
+    changed_at = self.changed_at
+    now = time.monotonic() if now is None else now
+    return self.network if changed_at is not None and now - changed_at <= RADIO_CHANGE_WINDOW else None
+
+  def _read(self) -> str | None:
+    try:
+      return self.counter.read_text().strip()
+    except OSError:
+      return None
+
+  def _run(self) -> None:
+    last = self._read()
+    while not self._stop.wait(self.interval):
+      current = self._read()
+      if current is None or current == last:
+        continue
+      last = current
+      self.network = ""  # named below; D-Bus can be slow, and the change must count from now
+      self.changed_at = time.monotonic()
+      try:
+        self.network = self.name(self.interface)
+      except Exception:
+        self.network = ""
+      self.log("other_wifi_changed", interface=self.interface, network=self.network)
+
+
 class EventLog:
   """Sanitized JSONL session log under /data/starport/logs, plus the recent tail in memory.
 
@@ -1263,6 +1320,8 @@ class Supervisor:
           self._set(error=message)
           reach = reachability.classify(stage, type(error).__name__, f"{message} {error}")
           self.log("attempt_failed", stage=stage, error=message, kind=type(error).__name__, where=_where(error), reach=reach)
+          if isinstance(error, SharedRadioStall):
+            reconnect_notice.write(reconnect_notice.shared_radio_text(error.network))
           if isinstance(error, StrayLinkError):
             self._clear_stray_link()
             car_first = True  # the car pages right back once its link drops; let it, rather than crossing it again
@@ -1305,6 +1364,9 @@ class Supervisor:
   def _describe_error(stage: str, error: Exception) -> str:
     if isinstance(error, AuthenticationRejected):
       return "The car rejected the Android Auto identity; it may have expired"
+    if isinstance(error, SharedRadioStall):
+      joined = f"joined {error.network}" if error.network else "was switching networks"
+      return f"The car's Wi-Fi stalled while the comma's own Wi-Fi {joined}; reconnecting"
     text = str(error) or type(error).__name__
     if stage == "authenticating" and "certificate" in text.lower():
       text += " (set verify_head_unit false in config.json to test without verifying the car)"
@@ -1667,6 +1729,10 @@ class Supervisor:
                         renderer_log=identity_store.LOG_DIR / "car_ui.log")
     self._set(view=source.label, encoder=getattr(encoder, "backend", "libx264"), target_fps=fps)
     watch = LinkWatch(lease, self.log).start()
+    radio = None
+    if getattr(lease, "dedicated", False) and getattr(lease, "interface", "wlan0") != "wlan0":
+      radio = SharedRadioWatch(self.log).start()
+      session.radio_change = radio.recent
     try:
       from openpilot.starpilot.system.starport.media_audio import MediaAudio, music_control
       from openpilot.starpilot.system.starport.media_status import MediaStatus
@@ -1690,6 +1756,8 @@ class Supervisor:
       if getattr(session, "nav", None) is not None:
         session.nav.close()
       watch.stop()
+      if radio is not None:
+        radio.stop()
       source.close()
       encoder.close()
       if self._stop.is_set():

openpilot/starpilot/system/starport/tests/test_projection_onroad_boundary.py

@@ -401,6 +401,40 @@ class TestProjectionOnroad(unittest.TestCase):
     self.assertEqual(shown, [notice, notice, 'Lane Departure Detected', '', notice])
     view.close()
 
+  def test_reconnect_notice_shows_eight_seconds_under_real_alerts_and_before_the_certificate(self):
+    native, _ = self.dependencies()
+
+    @dataclass(frozen=True)
+    class Alert:
+      size: str = 'none'
+      text1: str = ''
+      alert_type: str = ''
+
+    real = {'alert': Alert()}
+    native.alert, native.alert_size = Alert, SimpleNamespace(NONE='none', SMALL='small')
+    native.adapter = lambda *_: SimpleNamespace(build=lambda *_a, **_k: SimpleNamespace(
+      onroad=OnroadState(False, False, None, None, SpeedLimitObservation(), alert=cast(Any, real['alert']))))
+    reconnected = 'Reconnected: comma Wi-Fi Joined Home'
+    view = projection.ProjectionOnroad(dependencies=native, certificate_days=3, reconnect_notice=reconnected)
+    shown = []
+    view.onroad.render = lambda state: shown.append(state.alert.text1)
+    clock = [0]
+    original = projection.time
+    projection.time = SimpleNamespace(monotonic_ns=lambda: clock[0])
+    try:
+      native.ui_state.started = True
+      for clock[0] in (100, 7_000_000_000):
+        view.render()
+      real['alert'] = Alert('mid', 'Lane Departure Detected')
+      view.render()  # a real alert always wins
+      real['alert'] = Alert()
+      clock[0] = 8_000_000_200
+      view.render()  # past eight seconds: the certificate heads-up gets its remaining time
+    finally:
+      projection.time = original
+    self.assertEqual(shown, [reconnected, reconnected, 'Lane Departure Detected', 'Heads Up: AA Certificate Expires in 3 Days'])
+    view.close()
+
   def test_no_certificate_notice_when_far_from_expiry(self):
     native, _ = self.dependencies()
     view = projection.ProjectionOnroad(dependencies=native, certificate_days=200)

openpilot/starpilot/system/starport/tests/test_shared_radio_grace.py

@@ -0,0 +1,166 @@
+"""2026-11-02: the comma's own Wi-Fi joining or roaming stalled the car's video acks (one shared radio)."""
+import time
+
+import pytest
+
+from openpilot.starpilot.system.starport import reconnect_notice
+from openpilot.starpilot.system.starport.bugsink_diagnosis import stream_loss_kind
+from openpilot.starpilot.system.starport import supervisor as supervisor_module
+from openpilot.starpilot.system.starport.session import ProjectionSession, SharedRadioStall
+from openpilot.starpilot.system.starport.supervisor import SharedRadioWatch, Supervisor
+from openpilot.starpilot.system.starport.tests.fake_head_unit import FakeHeadUnit, make_identity
+from openpilot.starpilot.system.starport.tests.test_starport import end_session, pump_until, sent_frame, streaming_session
+
+
+@pytest.fixture(scope="module")
+def identity(tmp_path_factory):
+  return make_identity(tmp_path_factory.mktemp("identity"))
+
+
+def stalled(identity, network: str | None):
+  hu = FakeHeadUnit(identity)
+  session = streaming_session(hu, identity)
+  session.radio_change = lambda: network
+  hu.hold_acks.set()
+  return hu, session, sent_frame(hu, session)
+
+
+def test_no_radio_change_keeps_the_standard_limit(identity):
+  hu, session, sent = stalled(identity, None)
+  with pytest.raises(TimeoutError, match="2.5 s") as raised:
+    session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
+  assert not isinstance(raised.value, SharedRadioStall)
+  end_session(session, hu)
+
+
+def test_radio_change_waits_up_to_the_grace_limit_then_names_the_network(identity):
+  hu, session, sent = stalled(identity, "Huckleberrys House")
+  session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
+  session.check_progress(now=sent + ProjectionSession.ACK_GRACE_TIMEOUT - 0.1)
+  with pytest.raises(SharedRadioStall) as raised:
+    session.check_progress(now=sent + ProjectionSession.ACK_GRACE_TIMEOUT + 0.05)
+  assert raised.value.network == "Huckleberrys House"
+  end_session(session, hu)
+
+
+def test_grace_started_by_a_change_survives_the_change_ageing_out(identity):
+  hu, session, sent = stalled(identity, "Home")
+  session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
+  session.radio_change = lambda: None  # same stall: the decision was made when it began
+  session.check_progress(now=sent + ProjectionSession.ACK_GRACE_TIMEOUT - 0.1)
+  end_session(session, hu)
+
+
+def test_acks_resuming_during_the_grace_keep_the_session_and_log_the_gap(identity):
+  hu, session, sent = stalled(identity, "Home")
+  events = []
+  session.event = lambda name, **fields: events.append((name, fields))
+  session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
+  hu.hold_acks.clear()
+  hu._ack(session.session_id)
+  pump_until(session, lambda: not session.pending)
+  assert session.grace_frame is None
+  assert [name for name, _ in events] == ["ack_grace", "ack_grace_recovered"]
+  session.check_progress(now=sent + ProjectionSession.ACK_GRACE_TIMEOUT + 1)  # nothing pending: no timeout
+  end_session(session, hu)
+
+
+def test_car_taking_its_screen_back_during_the_grace_reconnects(identity):
+  hu, session, sent = stalled(identity, "Home")
+  session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
+  hu.set_focus(False)
+  pump_until(session, lambda: not session.focused)
+  with pytest.raises(SharedRadioStall, match="took its screen back"):
+    session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.5)
+  end_session(session, hu)
+
+
+def test_unfocus_outside_a_grace_is_the_normal_reverse_camera(identity):
+  hu = FakeHeadUnit(identity)
+  session = streaming_session(hu, identity)
+  hu.set_focus(False)
+  pump_until(session, lambda: not session.focused)
+  session.check_progress()
+  end_session(session, hu)
+
+
+def home(interface):
+  return "Huckleberrys House"
+
+
+def test_watch_reports_a_carrier_change_with_the_network_name(tmp_path, monkeypatch):
+  counter = tmp_path / "carrier_changes"
+  counter.write_text("4\n")
+  logged = []
+  watch = SharedRadioWatch(lambda name, **fields: logged.append((name, fields)), interval=0.01, counter=counter, name=home).start()
+  try:
+    time.sleep(0.05)
+    assert watch.recent() is None  # the starting value is not a change
+    counter.write_text("5\n")
+    deadline = time.monotonic() + 2
+    while watch.recent() != "Huckleberrys House":
+      assert time.monotonic() < deadline
+      time.sleep(0.01)
+    assert logged == [("other_wifi_changed", {"interface": "wlan0", "network": "Huckleberrys House"})]
+    assert watch.recent(now=watch.changed_at + supervisor_module.RADIO_CHANGE_WINDOW + 0.1) is None
+  finally:
+    watch.stop()
+
+
+def test_watch_without_the_counter_never_reports(tmp_path):
+  watch = SharedRadioWatch(lambda *a, **k: None, interval=0.01, counter=tmp_path / "missing", name=home).start()
+  try:
+    time.sleep(0.05)
+    assert watch.recent() is None
+  finally:
+    watch.stop()
+
+
+def test_name_lookup_failure_still_counts_the_change(tmp_path):
+  def broken(interface):
+    raise RuntimeError("NetworkManager timed out")
+
+  counter = tmp_path / "carrier_changes"
+  counter.write_text("1")
+  watch = SharedRadioWatch(lambda *a, **k: None, interval=0.01, counter=counter, name=broken).start()
+  try:
+    counter.write_text("2")
+    deadline = time.monotonic() + 2
+    while watch.recent() is None:
+      assert time.monotonic() < deadline
+      time.sleep(0.01)
+    assert watch.recent() == ""
+  finally:
+    watch.stop()
+
+
+def test_notice_text_is_short_and_plain():
+  assert reconnect_notice.shared_radio_text("Huckleberrys House") == "Reconnected: comma Wi-Fi Joined Huckleberrys House"
+  assert reconnect_notice.shared_radio_text("") == "Reconnected: comma Wi-Fi Was Switching Networks"
+  long = reconnect_notice.shared_radio_text("A" * 60)
+  assert long.endswith("...") and len(long) <= len("Reconnected: comma Wi-Fi Joined ") + reconnect_notice.NAME_MAX
+
+
+def test_notice_is_shown_once_and_only_while_fresh(tmp_path):
+  path = tmp_path / "notice.json"
+  reconnect_notice.write("Reconnected", path, now=1000.0)
+  assert reconnect_notice.take(path, now=1010.0) == "Reconnected"
+  assert reconnect_notice.take(path, now=1010.0) == ""  # taken
+  reconnect_notice.write("Reconnected", path, now=1000.0)
+  assert reconnect_notice.take(path, now=1000.0 + reconnect_notice.NOTICE_FRESH_S + 1) == ""
+  assert not path.exists()
+
+
+def test_galaxy_status_says_what_happened():
+  text = Supervisor._describe_error("streaming", SharedRadioStall("stalled", "Huckleberrys House"))
+  assert text == "The car's Wi-Fi stalled while the comma's own Wi-Fi joined Huckleberrys House; reconnecting"
+  assert "switching networks" in Supervisor._describe_error("streaming", SharedRadioStall("stalled", ""))
+  assert stream_loss_kind(text) == "shared_radio"  # bug reports file it apart from ordinary stalls
+  assert stream_loss_kind("projecting: Video acknowledgement older than 2.5 s") == "stalled"
+
+
+
+def test_saved_profile_names_become_the_network_name():
+  from openpilot.starpilot.system.starport.network import profile_network_name
+  assert profile_network_name("openpilot connection Huckleberrys House") == "Huckleberrys House"
+  assert profile_network_name("iPhone") == "iPhone"