Wait for NetworkManager to manage a shared Wi-Fi interface and tell stream losses apart

Nineteen Bugsink joins were refused with 'device is strictly unmanaged'.
On a shared interface the join now waits up to 8 s for NetworkManager to
manage it (dedicated mode already waited) and otherwise reports that the
interface is unmanaged and for how long, which the diagnosis already maps
to wifi_not_ready.

Stream losses keep the projection_interrupted category but now carry a kind
(stalled, silent, disconnected) with matching next steps, reported as
aa.loss_kind, so a car shutting off is no longer read as link loss.

openpilot/starpilot/system/starport/bugsink_diagnosis.py

@@ -97,6 +97,24 @@ def attempt_records(events):
   return attempts
 
 
+STREAM_LOSS = {
+  "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."]),
+  "silent": ("Head unit stopped answering", "The head unit stopped answering protocol messages while the link looked up.",
+             ["Check whether the car's screen turned off or switched source at that time.",
+              "Compare with other drives on the same head unit before changing timings."]),
+  "disconnected": ("Head unit disconnected", "The head unit closed the connection, which is what a car shutdown or a source change looks like.",
+                   ["Check whether the drive ended or the driver changed source on the head unit.",
+                    "Only investigate further if it happened while the car stayed on and Android Auto was showing."]),
+}
+
+
+def stream_loss_kind(error: str) -> str:
+  """Why a stream ended, from the logged error text: stalled, silent, disconnected or other."""
+  return next((kind for kind, (marker, _, _) in STREAM_LOSS.items() if marker.lower() in error.lower()), "other")
+
+
 def diagnosis(session):
   stage, error = session.get("failed_stage", ""), session.get("error", "").lower()
   category, confidence = "unknown", "unknown"
@@ -156,6 +174,8 @@ def diagnosis(session):
     explanation = "Projection was interrupted. EOF or a timeout alone cannot distinguish a car shutdown from link loss."
     actions = ["Compare video acknowledgments, Wi-Fi signal, stalls, encoder recoveries and head-unit shutdown events.",
                "Retry while parked and verify sustained acknowledged frames, not just a streaming status."]
+    if (loss := stream_loss_kind(session.get("error", ""))) != "other":
+      explanation, actions = STREAM_LOSS[loss][1], STREAM_LOSS[loss][2]
   return {"category": category, "confidence": confidence, "summary": explanation,
           "evidence": session.get("error", ""), "recovered": session.get("outcome") == "recovered",
           "next_steps": actions, "log_file": session.get("name", ""),

openpilot/starpilot/system/starport/bugsink_report.py

@@ -14,7 +14,7 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from openpilot.starpilot.system.starport import identity as identity_store, reachability, system_slice
-from openpilot.starpilot.system.starport.bugsink_diagnosis import attempt_records, diagnosis, elapsed, sanitize
+from openpilot.starpilot.system.starport.bugsink_diagnosis import attempt_records, diagnosis, elapsed, sanitize, stream_loss_kind
 from openpilot.starpilot.system.starport.bugsink_summary import batch_summary, failure_summary, pairing_summary, session_timeline, software_tags
 
 STATE_PATH = identity_store.DATA_DIR / "bugsink_report.json"
@@ -405,6 +405,7 @@ def failure_detail(session: dict) -> dict:
     "session": session["seq"],
     "attempt": session.get("failed_attempt"),
     "cause": assessment["category"], "stage": stage, "error_kind": kind, "reach": session.get("reach", ""),
+    "loss": stream_loss_kind(session.get("error", "")) if assessment["category"] == "projection_interrupted" else "",
     "summary": sanitize(failure_summary(assessment), names),
     "diagnosis": sanitize(assessment, names),
   }
@@ -499,7 +500,8 @@ def drive_event(sessions: list[dict]) -> dict:
              **({"aa.phone_address_sent": "yes" if sent else "no"} if sent is not None else {}),
              **({"aa.attention": ",".join(reasons)[:190]} if reasons else {}),
              **({"aa.cause": latest["cause"], "aa.stage": latest["stage"], "aa.error_kind": latest["error_kind"]} if latest else {}),
-             **({"aa.reach_class": latest["reach"]} if latest and latest.get("reach") else {})},
+             **({"aa.reach_class": latest["reach"]} if latest and latest.get("reach") else {}),
+             **({"aa.loss_kind": latest["loss"]} if latest and latest.get("loss") else {})},
     "breadcrumbs": {"values": attempt_breadcrumbs(sessions)},
     "contexts": {"starport_drive": summary, "starport_head_unit": head_unit,
                  **{f"starport_session_{index + 1}": _public(session) for index, session in enumerate(sessions)},

openpilot/starpilot/system/starport/network.py

@@ -40,6 +40,10 @@ NM_AP_IFACE = "org.freedesktop.NetworkManager.AccessPoint"
 DEVICE_TYPE_WIFI = 2
 DEVICE_STATE_DISCONNECTED = 30
 DEVICE_READY_TIMEOUT_S = 5.0
+# A shared interface can be briefly unmanaged (for example while the Wi-Fi mode is being switched). Bugsink showed 19 joins
+# refused with "device is strictly unmanaged"; waiting a few seconds turns a transient state into a normal join, and a
+# device that stays unmanaged is reported as such instead of as an opaque D-Bus error.
+UNMANAGED_WAIT_S = 8.0
 ACTIVE_STATE_ACTIVATED = 2
 ACTIVE_STATE_DEACTIVATED = 4
 # The car sends its Wi-Fi details ~4 s after ignition, but a Honda's access point took ~55 s from ignition to appear
@@ -253,6 +257,7 @@ class NetworkLease:
         self._prepare_device(cancelled)
       else:
         self.device_path = self._wifi_device()
+        self._wait_managed(cancelled)
         self._pause_autoconnect()
       active = self._get(self.device_path, NM_DEVICE_IFACE, "ActiveConnection")
       if active and active != "/" and not self.previous_connection and not self.dedicated:
@@ -316,6 +321,22 @@ class NetworkLease:
       raise DedicatedUnavailable(failure)
     time.sleep(0.25)
 
+  def _wait_managed(self, cancelled: Callable[[], bool], timeout: float = UNMANAGED_WAIT_S) -> None:
+    """Give NetworkManager a few seconds to manage the shared interface before joining on it."""
+    deadline = time.monotonic() + timeout
+    waited = False
+    while (state := self._get(self.device_path, NM_DEVICE_IFACE, "State")) < DEVICE_STATE_DISCONNECTED:
+      if cancelled():
+        raise NetworkError("cancelled")
+      if time.monotonic() >= deadline:
+        raise NetworkError(f"{self.interface} is unmanaged by NetworkManager (state {state}) after waiting {timeout:.0f} s")
+      if not waited:
+        waited = True
+        self.log("wifi_waiting_managed", interface=self.interface, state=state)
+      time.sleep(0.25)
+    if waited:
+      self.log("wifi_became_managed", interface=self.interface)
+
   def _prepare_device(self, cancelled: Callable[[], bool]) -> None:
     if self.device_ready:
       return

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

@@ -0,0 +1,45 @@
+"""A shared Wi-Fi interface that NetworkManager has not taken yet is waited for, then reported as unmanaged."""
+import pytest
+
+from openpilot.starpilot.system.starport import network
+from openpilot.starpilot.system.starport.network import NetworkError, NetworkLease
+
+
+class FakeLease(NetworkLease):
+  def __init__(self, states):
+    super().__init__(lambda *a, **k: self.events.append((a, k)), "wlan0")
+    self.states, self.events, self.device_path = list(states), [], "/dev/0"
+
+  def _get(self, path, interface, name):
+    assert name == "State"
+    return self.states.pop(0) if len(self.states) > 1 else self.states[0]
+
+
+@pytest.fixture(autouse=True)
+def no_sleep(monkeypatch):
+  monkeypatch.setattr(network.time, "sleep", lambda seconds: None)
+
+
+def test_a_device_that_becomes_managed_is_joined_normally():
+  lease = FakeLease([10, 10, 20, 30])  # unmanaged, unmanaged, unavailable, disconnected
+  lease._wait_managed(lambda: False)
+  assert [event[0][0] for event in lease.events] == ["wifi_waiting_managed", "wifi_became_managed"]
+
+
+def test_a_managed_device_does_not_wait_or_log():
+  lease = FakeLease([30])
+  lease._wait_managed(lambda: False)
+  assert lease.events == []
+
+
+def test_a_device_that_stays_unmanaged_says_so(monkeypatch):
+  ticks = iter(range(0, 1000, 3))
+  monkeypatch.setattr(network.time, "monotonic", lambda: next(ticks))
+  lease = FakeLease([10])
+  with pytest.raises(NetworkError, match="unmanaged by NetworkManager"):
+    lease._wait_managed(lambda: False, timeout=8.0)
+
+
+def test_cancelling_stops_the_wait():
+  with pytest.raises(NetworkError, match="cancelled"):
+    FakeLease([10])._wait_managed(lambda: True)

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

@@ -52,3 +52,15 @@ def test_report_names_a_sleeping_car_instead_of_unknown(tmp_path):
   recorded = [(0, 0, "session_start", {}), (0, 9, "attempt_failed", {"stage": "rfcomm", "kind": "ConnectionRefusedError",
                                                                       "error": "refused", "reach": "not_ready"})]
   assert br.session_record(write_session(tmp_path, 2, recorded))["reach"] == "not_ready"
+
+
+def test_stream_losses_are_told_apart_by_what_the_log_says(tmp_path):
+  from openpilot.starpilot.system.starport.bugsink_diagnosis import stream_loss_kind
+  assert stream_loss_kind("projecting: Video acknowledgement older than 2.5 s") == "stalled"
+  assert stream_loss_kind("projecting: Head unit stopped answering for 6 s") == "silent"
+  assert stream_loss_kind("projecting: Head unit disconnected after 4/8 bytes") == "disconnected"
+  assert stream_loss_kind("something else") == "other"
+  for text, kind in (("Video acknowledgement older than 2.5 s", "stalled"), ("Head unit disconnected after 4/8 bytes", "disconnected")):
+    events = [(0, 0, "session_start", {}), (0, 1, "stage", {"state": "streaming"}), (0, 9, "attempt_failed", {"stage": "streaming", "kind": "EOFError", "error": text})]
+    event = br.drive_event([br.session_record(write_session(tmp_path, 3 if kind == "stalled" else 4, events))])
+    assert event["tags"]["aa.cause"] == "projection_interrupted" and event["tags"]["aa.loss_kind"] == kind