StarPort: Checklist cleanup for car fixes, logs and AAWireless

openpilot/starpilot/galaxy/tests/test_starport_guided.mjs

@@ -199,6 +199,8 @@ console.log("StarPort guided setup: question, steps, candidates, live checklist,
   const done = guidedState({ setup, runtime: refused, setupDone: true })
   assert.equal(done.phase, "bluetooth")   // even a finished setup shows the thing to fix
   assert.equal(guidedState({ setup, runtime: { running: false, state: "error", error: "something else" }, carType: "adapter" }).hint, null)
+  const waiting = { running: true, state: "backoff", error: "Connecting to car: Waiting for the adapter to finish starting up" }
+  assert.equal(guidedState({ setup, runtime: waiting, carType: "adapter" }).hint, null)  // AAWireless waits; it never asks for a car
   assert.match(StarPortGuidedSetup.template, /state\.hint\.href/)
 }
 // A car-specific fix is only ever advice on the setup screen: shown when the daemon names one, never otherwise.

openpilot/starpilot/galaxy/web/js/settings.js

@@ -311,7 +311,7 @@ export const SettingsPage = {
       const owner = (this.state.data?.rows || []).map((row) => (row.heading ? heads.push({ title: row.label, collapsible: !!row.collapsed }) : heads.length))
       const cards = [{ title: "", entries: [], collapsible: false, open: true }, ...heads.map(({ title, collapsible }) => ({ title, entries: [], collapsible, open: !collapsible }))]
       for (const entry of this.visibleRows) if (!entry.row.heading) cards[owner[entry.index] || 0].entries.push(entry)
-      for (const card of cards) if (card.collapsible) card.open = !!this.state.query || (this.state.openCards || []).includes(card.title)
+      for (const card of cards) if (card.collapsible) card.open = !!this.state.query || this.state.openCards.includes(card.title)
       return cards.filter((card, index) => !index || card.entries.length)
     },
   },
@@ -366,7 +366,7 @@ export const SettingsPage = {
       }
     },
     toggleCard(title) {
-      const open = this.state.openCards || (this.state.openCards = [])
+      const open = this.state.openCards
       const at = open.indexOf(title)
       if (at < 0) open.push(title)
       else open.splice(at, 1)

openpilot/starpilot/galaxy/web/js/starport-guided.js

@@ -121,8 +121,9 @@ export function guidedState({ setup = null, runtime = null, selected = null, rec
     const restarting = state === "dongle_restart"
     const hangups = Number(runtime?.hangups) || 0
     // Only an unofficial dongle waits for a restart; for every other receiver repeated hang-ups are said, not acted on.
+    const adapter = carType === "adapter"
     const hangupNotice = wireless && !restarting && hangups >= HANGUP_NOTICE_AT ?
-      `${carType === "adapter" ? "The adapter" : "The car"} has hung up ${hangups} times before answering. If this keeps happening, ${carType === "adapter" ? "unplug the adapter for 10 seconds" : "turn the car off and on again"}.` : ""
+      `${adapter ? "The adapter" : "The car"} has hung up ${hangups} times before answering. If this keeps happening, ${adapter ? "unplug the adapter for 10 seconds" : "turn the car off and on again"}.` : ""
     return { ...base, show: true, phase: restarting ? "restart" : "connecting", title: restarting ? RESTART_TITLE : runtime?.label || "Connecting to your car",
       detail: restarting ? RESTART_DETAIL : runtime?.detail || "Keep your car’s screen on. We’ll move to the next step automatically.",
       progress: connectionStages(runtime), handshake: wireless ? handshakeSteps(ladder) : [], exhausted: !!ladder?.exhausted,

openpilot/starpilot/system/starport/bugsink_report.py

@@ -320,7 +320,7 @@ def session_record(path: Path) -> dict:
     "setup": setup,
     "offered_modes": offered_modes(events),
     # True when the comma's address went to the car in the pairing request; None when the car never opened that channel.
-    "phone_address_sent": bool(opened.get("phone_address")) if opened is not None else None,
+    "phone_address_sent": bool(opened.get("phone_address_sent", opened.get("phone_address"))) if opened is not None else None,
     "strategy": session_strategy(events, setup),
     "ladder": session_ladder(events),
     "fixes": session_fixes(events),

openpilot/starpilot/system/starport/compat_report.py

@@ -142,14 +142,15 @@ def summarize(events: list[dict]) -> dict:
       frames = 0
     elif name == "bootstrap_rx":
       frames += 1
-    elif name == "ladder_strategy" and record.get("rung"):
-      report["ladder"].setdefault(record["rung"], {"tries": 0, "failed": 0, "why": ""})["tries"] += 1
-    elif name == "ladder_failed" and record.get("rung"):
+    elif name in ("ladder_strategy", "ladder_failed", "ladder_worked") and record.get("rung"):
       rung = report["ladder"].setdefault(record["rung"], {"tries": 0, "failed": 0, "why": ""})
-      rung["failed"] += 1
-      rung["why"] = record.get("why") or record.get("reason", "")
-    elif name == "ladder_worked" and record.get("rung"):
-      report["ladder"].setdefault(record["rung"], {"tries": 0, "failed": 0, "why": ""})["why"] = "worked"
+      if name == "ladder_strategy":
+        rung["tries"] += 1
+      elif name == "ladder_failed":
+        rung["failed"] += 1
+        rung["why"] = record.get("why") or record.get("reason", "")
+      else:
+        rung["why"] = "worked"
     elif name == "dongle_restart_requested":
       report["restarts"].append({"t": record.get("t", ""), "hangups": record.get("hangups"), "how": ""})
     elif name == "dongle_restart_done" and report["restarts"]:

openpilot/starpilot/system/starport/head_unit_fixes.py

@@ -1,17 +1,11 @@
-"""Car-specific fixes: rules matched on one exact head-unit build, applied through small reusable actions.
-
-A rule names the build it was seen on, the failure it targets and an ordered ladder of actions. A trial runs one more action
-per level (see quirk_state), so the smallest change that helps is the one that stays. Rules are either
-``experimental`` (they run only when the owner switches them on) or ``validated`` (a tested sequence, applied automatically
-to the matching build). "Try experimental fixes" is the master switch: off, no rule runs, whatever has been learned.
-"""
+"""Car-specific fixes: rules for one exact head-unit build, each a ladder of small actions tried one level at a time."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass
 from typing import Any
 
-from openpilot.starpilot.system.starport.quirk_state import QuirkState
+from openpilot.starpilot.system.starport.quirk_state import MATCH_FIELDS, QuirkState
 
 EXPERIMENTAL, VALIDATED = "experimental", "validated"
 
@@ -26,10 +20,7 @@ ACTIONS = (OPEN_BLUETOOTH, OPEN_PHONE_STATUS, OPEN_SENSORS, OPEN_VENDOR, REQUEST
 # ServiceDiscoveryResponse field numbers (JSON turns them into strings).
 SDR_LABELS = {"2": "car_make", "3": "car_model", "4": "car_year", "7": "head_unit_make", "8": "head_unit_model",
               "9": "head_unit_software_build", "10": "head_unit_software_version"}
-MATCH_FIELDS = ("car_make", "car_model", "car_year", "head_unit_make", "head_unit_model", "head_unit_software_version")
 
-# Automatic experiments (an experimental rule that starts without its own toggle) are not enabled in this release.
-AUTO_EXPERIMENTS = False
 WIRELESS, WIRED = "wireless", "wired"
 
 
@@ -37,7 +28,6 @@ WIRELESS, WIRED = "wireless", "wired"
 class Evidence:
   failure: str                    # what the car does
   sessions: tuple[str, ...]       # where it was seen
-  early_shutdown_s: float | None  # the known-bad baseline: seconds of video before the car ends the session
 
 
 @dataclass(frozen=True)
@@ -60,7 +50,7 @@ RULES: tuple[Rule, ...] = (
               "head_unit_model": "myChevrolet", "head_unit_software_version": "W28E-V155.3.8-MIHS21A-107"},
        transports=(WIRELESS, WIRED), status=EXPERIMENTAL,
        evidence=Evidence("The car ends every session 10-25 s after video starts (shutdown reason 1), wired and wireless.",
-                         ("2026-10-12 sessions 5-8", "2026-11-08 sessions 1, 2, 4"), 25.0),
+                         ("2026-10-12 sessions 5-8", "2026-11-08 sessions 1, 2, 4")),
        ladder=(OPEN_BLUETOOTH, OPEN_PHONE_STATUS, OPEN_SENSORS, OPEN_VENDOR, REQUEST_BT_PAIRING),
        summary="This car's screen keeps ending after about 25 seconds."),
 )
@@ -79,7 +69,7 @@ class BootstrapFix:
 BOOTSTRAP_FIXES: tuple[BootstrapFix, ...] = (
   BootstrapFix(id="phone_ping", label="Dongle Wi-Fi setup: ping like a phone", experiment="phone_ping",
                evidence=Evidence("A Mavlou dongle never sends its Wi-Fi start request to the comma. A phone pings it once a second after its version reply.",
-                                 ("2026-11-03 phone capture", "2026-11-07 sessions 5, 19, 24"), None)),
+                                 ("2026-11-03 phone capture", "2026-11-07 sessions 5, 19, 24"))),
 )
 
 
@@ -163,7 +153,7 @@ def _enabled(config: dict, rule: Rule) -> bool:
   """Try experimental fixes is the master switch: off, no rule of either status runs, whatever has been learned."""
   if not config.get("head_unit_experiments"):
     return False
-  return rule.status == VALIDATED or AUTO_EXPERIMENTS or rule.id in (config.get("head_unit_experiment_rules") or [])
+  return rule.status == VALIDATED or rule.id in (config.get("head_unit_experiment_rules") or [])
 
 
 def eligible(car_type: str, profile: str) -> bool:

openpilot/starpilot/system/starport/network.py

@@ -127,7 +127,7 @@ def connection_settings(credentials: WifiCredentials, interface: str, static_ip:
   }
   if credentials.bssid:
     wireless["bssid"] = ("ay", bytes.fromhex(credentials.bssid.replace(":", "")))
-  settings = {
+  settings: dict[str, dict] = {
     "connection": {
       "type": ("s", "802-11-wireless"),
       "uuid": ("s", str(uuid.uuid4())),

openpilot/starpilot/system/starport/projection_layout.py

@@ -377,7 +377,7 @@ def _salvage_layout(value, viewport):
     raise ValueError('Projection layout does not match saved screen')
   defaults = {key: dict(widget['default']) for key, widget in metadata['widgets'].items()}
   saved = {key: placement for key, placement in value['widgets'].items() if key in defaults}
-  result = {'version': value.get('version'), 'canvas': value['canvas'], 'widgets': dict(defaults)}
+  result: dict = {'version': value.get('version'), 'canvas': value['canvas'], 'widgets': dict(defaults)}
 
   def accepts(candidate):
     try:

openpilot/starpilot/system/starport/quirk_state.py

@@ -1,26 +1,21 @@
-"""What StarPort has learned about how each car reacts to car-specific fixes, kept apart from the setup ladder's state.
-
-One file, ``quirk_state.json``, keyed by the chosen car's Bluetooth address. Each car stores the head-unit fingerprint it was
-learned with (a different head unit on the same address wipes the entry), an observation-only baseline, and one trial per rule.
-A trial ends each session as ``success``, ``failed``, ``inconclusive`` or ``regression``; only a recurrence of the targeted
-failure moves the ladder up, and a regression on any measured axis reverts the trial. Wired cars have no reliable identity, so
-their entries live in memory for the drive and are never written.
-"""
+"""Per-car record of how each car-specific fix worked, kept in quirk_state.json apart from the setup ladder."""
 
 from __future__ import annotations
 
 import json
-import os
 import statistics
-import tempfile
 import threading
 from dataclasses import dataclass
 from pathlib import Path
 
+from openpilot.common.utils import atomic_write
 from openpilot.starpilot.system.starport import identity as identity_store
+from openpilot.starpilot.system.starport.compat_report import read_log
 
 STATE_VERSION = 1
 MAX_CARS = 16
+MAX_STATE_BYTES = 256 << 10
+MATCH_FIELDS = ("car_make", "car_model", "car_year", "head_unit_make", "head_unit_model", "head_unit_software_version")
 EARLY_SECONDS = 60.0        # a car that ends the session sooner than this after video started has hung up early
 CONFIRM_SECONDS = 120.0     # streaming this long with no early shutdown is a success
 CONFIRM_SESSIONS = 2        # successes needed to confirm a trial
@@ -43,7 +38,6 @@ class SessionOutcome:
   first_frame_s: float | None = None
   ended_by_car: bool = False        # the car sent a shutdown request
   user_stopped: bool = False        # the owner stopped projection, or the car or comma went away
-  car_off: bool = False             # the car stopped answering because it powered down
   hangup: bool = False              # the link dropped with no video, or the car hung up on a reply
   pairing_prompt: bool = False      # the car reported the comma as not yet paired after a pairing request
   waiting_for_driver: bool = False  # focus was never granted and the car may be waiting for the driver to choose
@@ -55,7 +49,7 @@ class SessionOutcome:
 
 def classify(outcome: SessionOutcome, frozen: dict) -> str:
   """The verdict for one session run with a fix on, judged against the baseline frozen when the trial started."""
-  if outcome.user_stopped or outcome.car_off or outcome.waiting_for_driver:
+  if outcome.user_stopped or outcome.waiting_for_driver:
     return INCONCLUSIVE
   if outcome.pairing_prompt:
     return REGRESSION
@@ -127,7 +121,7 @@ class QuirkState:
 
   def _load(self) -> None:
     try:
-      raw = json.loads(self.path.read_text())
+      raw = json.loads(read_log(self.path, MAX_STATE_BYTES))
     except (OSError, ValueError):
       return
     if not isinstance(raw, dict) or raw.get("version") != STATE_VERSION or not isinstance(raw.get("cars"), dict):
@@ -151,17 +145,8 @@ class QuirkState:
     with self._lock:
       data = json.dumps({"version": STATE_VERSION, "generation": self.generation, "cars": self.cars}, indent=1) + "\n"
       self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-      fd, temp = tempfile.mkstemp(prefix=".quirk-", dir=self.path.parent)
-      try:
-        with os.fdopen(fd, "w") as handle:
-          handle.write(data)
-        os.chmod(temp, 0o600)
-        os.replace(temp, self.path)
-      except OSError:
-        try:
-          os.unlink(temp)
-        except OSError:
-          pass
+      with atomic_write(str(self.path), overwrite=True) as handle:
+        handle.write(data)
 
   # --------------------------------------------------------------------- keys
 
@@ -172,8 +157,7 @@ class QuirkState:
   @staticmethod
   def wired_key(fingerprint: dict) -> str:
     """A head-unit key for one drive. Only a full fingerprint identifies a build; anything less gets no key."""
-    needed = ("car_make", "car_model", "car_year", "head_unit_make", "head_unit_model", "head_unit_software_version")
-    return "usb:" + "|".join(fingerprint.get(k, "") for k in needed) if all(fingerprint.get(k) for k in needed) else ""
+    return "usb:" + "|".join(fingerprint.get(k, "") for k in MATCH_FIELDS) if all(fingerprint.get(k) for k in MATCH_FIELDS) else ""
 
   def _store(self, key: str) -> dict:
     return self.memory if key.startswith("usb:") else self.cars
@@ -297,6 +281,9 @@ class QuirkState:
       if key:
         self.cars.pop(key, None)
         self.memory.pop(key, None)
+        for entry in (*self.cars.values(), *self.memory.values()):
+          for trial in entry["trials"].values():
+            trial["generation"] = self.generation
       else:
         self.cars.clear()
         self.memory.clear()

openpilot/starpilot/system/starport/session.py

@@ -811,8 +811,8 @@ class ProjectionSession(Session):
     methods = bluetooth["bluetooth"].get("pairing_methods") or []
     if pair:
       self._send_pairing_request(methods)
-    self.event("bluetooth_opened", channel=bluetooth["id"], phone_address=self.phone_address if pair else "",
-               car_address=bluetooth["bluetooth"].get("car_address", ""), pairing_methods=methods)
+    self.event("bluetooth_opened", channel=bluetooth["id"], phone_address_sent=bool(pair and self.phone_address),
+               has_car_address=bool(bluetooth["bluetooth"].get("car_address")), pairing_methods=methods)
     return True
 
   def _send_pairing_request(self, methods: list[int]) -> None:
@@ -961,10 +961,11 @@ class ProjectionSession(Session):
     was_focused = self.focused
     granted = focus in (FOCUS_PROJECTED, FOCUS_PROJECTED_NO_INPUT)
     self.focused = granted and self.allow_projection
-    first_grant = self.focused and self.focus_granted_at is None
+    wait = {}
     if self.focused and self.focus_granted_at is None:
       self.focus_granted_at = time.monotonic()
-    wait = {"wait_s": round(self.focus_granted_at - self.focus_requested_at, 2)} if first_grant and self.focus_requested_at is not None else {}
+      if self.focus_requested_at is not None:
+        wait = {"wait_s": round(self.focus_granted_at - self.focus_requested_at, 2)}
     self.event("video_focus", focus=focus, unsolicited=unsolicited, focused=self.focused, **wait)
     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

openpilot/starpilot/system/starport/supervisor.py

@@ -2138,7 +2138,7 @@ class Supervisor:
       if selection:
         self.log("fix_outcome", rule=selection.rule.id, level=selection.level, verdict=verdict, streamed_s=round(outcome.streamed_s, 1),
                  state=after["state"] if after else "", reason=after.get("reason", "") if after else "")
-    except (OSError, ValueError, KeyError, TypeError) as error:
+    except OSError as error:
       self.log("fix_outcome_error", error=str(error)[:120])
 
   def _note_streamed(self, session: ProjectionSession) -> None:

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

@@ -482,6 +482,9 @@ def test_report_records_that_the_address_was_not_sent_and_old_logs_still_report(
   events: list = [list(item) for item in STREAMED]
   events.insert(6, (0, 10, "bluetooth_opened", {"channel": 6, "phone_address": ""}))
   assert br.drive_event([br.session_record(write_session(tmp_path, 1, events))])["tags"]["aa.phone_address_sent"] == "no"
+  current: list = [list(item) for item in STREAMED]
+  current.insert(6, (0, 10, "bluetooth_opened", {"channel": 6, "phone_address_sent": True, "has_car_address": True}))
+  assert br.session_record(write_session(tmp_path, 3, current))["phone_address_sent"] is True
   old = br.session_record(write_session(tmp_path, 2, STREAMED))  # a log written before these fields existed
   assert old["setup"] == {} and old["offered_modes"] == [] and old["phone_address_sent"] is None and old["strategy"] == {}
   tags = br.drive_event([old])["tags"]

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

@@ -230,3 +230,16 @@ def test_unofficial_dongle_turns_on_the_handshake_experiments_only_when_chosen()
   assert car_profiles.tuning(config(car_profile_override="unofficial"))["bootstrap_experiments"] is True
   for other in (config(), config(car_profile_override="default"), config(car_profile_override="aawireless")):
     assert car_profiles.tuning(other)["bootstrap_experiments"] is False
+
+
+@pytest.mark.parametrize("profile, expected", [("aawireless", "Waiting for the adapter to finish starting up"),
+                                               ("default", "Connect your paired car in Galaxy Bluetooth first")])
+def test_an_adapter_without_hands_free_yet_waits_or_asks_for_the_car(sup, profile, expected):
+  bluez = Mock()
+  bluez.device.return_value = {"address": ADAPTER, "paired": True, "starport": True, "uuids": []}
+  sup._phone = lambda: bluez
+  sup.config.update(connection="wireless", receiver_address=ADAPTER, receiver_name="AAWireless-4F2A", companion_address="",
+                    car_type="adapter", car_profile_override=profile)
+  with pytest.raises(RuntimeError, match=expected):
+    sup._attempt(None)
+  bluez.connect_device.assert_not_called()

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

@@ -1,3 +1,4 @@
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -21,12 +22,12 @@ def test_shipped_rules_validate():
 
 @pytest.mark.parametrize("change", [{"head_unit_software_version": "W28E-*"}, {"car_year": ""}, {"extra": "x"}])
 def test_validate_rejects_wide_rules(change):
-  bad = hf.Rule(**{**RULE.__dict__, "match": {**RULE.match, **change}})
+  bad = replace(RULE, match={**RULE.match, **change})
   assert hf.validate_rules((bad,))
 
 
 def test_validate_rejects_missing_evidence():
-  bad = hf.Rule(**{**RULE.__dict__, "evidence": hf.Evidence("", (), None)})
+  bad = replace(RULE, evidence=hf.Evidence("", ()))
   assert hf.validate_rules((bad,))
 
 
@@ -83,7 +84,9 @@ def test_wired_selection_with_key():
 
 def test_reverted_trial_stays_off(tmp_path):
   q = QuirkState(tmp_path / "q.json")
-  q.begin("bt:AA", BOLT, RULE.id)["state"] = "off"
+  trial = q.begin("bt:AA", BOLT, RULE.id)
+  assert trial is not None
+  trial["state"] = "off"
   assert pick(CFG_ON, q=q) is None
 
 

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

@@ -27,7 +27,9 @@ def run(identity, tmp_path, config, level=None, **hu_options):
   def choose(info):
     labels = hf.head_unit_labels(info)
     if level is not None:
-      quirks.begin("bt:AA", labels, RULE.id)["level"] = level
+      trial = quirks.begin("bt:AA", labels, RULE.id)
+      assert trial is not None
+      trial["level"] = level
     selection = hf.select(config, labels, "wireless", quirks, "bt:AA", car_type="builtin", profile="default")
     return selection.actions if selection else ()
 

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

@@ -266,7 +266,7 @@ class TestProjectionOnroad(unittest.TestCase):
       def deactivate(self): pass
       def close(self): pass
 
-    class Warning:
+    class FakeWarning:
       def sample(self, **kwargs):
         calls.append(kwargs)
         return True, False
@@ -279,7 +279,7 @@ class TestProjectionOnroad(unittest.TestCase):
                           canValid=True, gearShifter=drive)
     native.ui_state.params = object()
     native.display_message = lambda _sm, name, *_a, **_k: car if name == 'carState' else None
-    native.pip_renderer, native.pip_warning = Renderer, Warning
+    native.pip_renderer, native.pip_warning = Renderer, FakeWarning
     native.read_pip, native.read_vasm = (lambda _p: saved), (lambda _p: vasm)
     native.clock_pair_ns = lambda: (1, 2)
     native.car_schema = SimpleNamespace(CarState=SimpleNamespace(GearShifter=SimpleNamespace(drive=drive)))

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

@@ -21,7 +21,6 @@ def test_classify_cases():
   assert qs.classify(EARLY, frozen) == qs.FAILED
   assert qs.classify(GOOD, frozen) == qs.SUCCESS
   assert qs.classify(SessionOutcome(video_seen=True, streamed_s=30, user_stopped=True), frozen) == qs.INCONCLUSIVE
-  assert qs.classify(SessionOutcome(video_seen=True, streamed_s=30, car_off=True, ended_by_car=True), frozen) == qs.INCONCLUSIVE
   assert qs.classify(SessionOutcome(waiting_for_driver=True), frozen) == qs.INCONCLUSIVE
   assert qs.classify(SessionOutcome(video_seen=True, streamed_s=30), frozen) == qs.INCONCLUSIVE
   assert qs.classify(SessionOutcome(pairing_prompt=True), frozen) == qs.REGRESSION
@@ -122,3 +121,27 @@ def test_forget_offline_removes_entry(tmp_path):
   assert qs.forget_offline("aa", tmp_path / "quirk_state.json")
   assert state(tmp_path).cars == {}
   assert not qs.forget_offline("aa", tmp_path / "quirk_state.json")
+
+
+def test_resetting_one_car_keeps_the_other_cars_trials(tmp_path):
+  q = state(tmp_path)
+  q.begin("bt:AA", FP, "r")
+  q.begin("bt:BB", FP, "r")
+  run(q, "bt:BB", EARLY)
+  run(q, "bt:BB", EARLY)
+  q.reset("bt:AA")
+  assert q.trial("bt:AA", "r") is None
+  assert q.trial("bt:BB", "r")["level"] == 2
+
+
+def test_oversized_or_linked_state_file_is_ignored(tmp_path):
+  path = tmp_path / "quirk_state.json"
+  path.write_text(json.dumps({"version": qs.STATE_VERSION, "cars": {}, "pad": "x" * qs.MAX_STATE_BYTES}))
+  assert state(tmp_path).cars == {}
+  real = tmp_path / "real.json"
+  q = QuirkState(real)
+  q.begin("bt:AA", FP, "r")
+  q.save()
+  path.unlink()
+  path.symlink_to(real)
+  assert state(tmp_path).cars == {}

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

@@ -613,6 +613,8 @@ def test_session_opens_car_data_and_bluetooth_like_a_phone(identity):
   session = connect(hu, identity)
   session.phone_services = True
   session.phone_address = "39:90:21:64:07:00"
+  events = []
+  session._log = lambda name, **values: events.append((name, values))
   session.authenticate()
   session.start("StarPilot", "comma.ai")
   pump_until(session, lambda: session.focused and session.sensor_events)
@@ -624,6 +626,9 @@ def test_session_opens_car_data_and_bluetooth_like_a_phone(identity):
   assert hu.sensor_requests == [13, 10]  # driving status and night only; never location
   assert hu.pairing_requests == [("39:90:21:64:07:00", 2)]
   assert session.stats()["sensor_events"] == 1
+  opened = next(values for name, values in events if name == "bluetooth_opened")
+  assert opened["phone_address_sent"] is True
+  assert "39:90:21:64:07:00" not in repr(opened)
 
 
 def test_session_leaves_car_data_alone_by_default(identity):

openpilot/starpilot/ui/tests/test_layout_widget_art.py

@@ -182,7 +182,7 @@ def test_starport_widgets_get_sprites_and_empty_ones_keep_the_sketch(monkeypatch
   monkeypatch.setattr(canvas, 'fonts', NS(profile=Profile.LARGE), raising=False)
   canvas.target = NS(texture=object())
   drawn = []
-  canvas.view = NS(_ordered_widgets=lambda content, state, only_widget=None: drawn.append(only_widget))
+  monkeypatch.setattr(canvas, 'view', NS(_ordered_widgets=lambda content, state, only_widget=None: drawn.append(only_widget)), raising=False)
   monkeypatch.setattr(canvas, '_settle', lambda state: None)
   monkeypatch.setitem(renderer.RUNTIME_PREVIEW_WIDGETS, 'large', ())
   monkeypatch.setattr(rl, 'rl_set_blend_factors_separate', Mock())

openpilot/tools/starport/cars_md.py

@@ -23,7 +23,7 @@ def fetch(base: str, token: str, path: str) -> dict:
     return json.load(response)
 
 
-def pages(base: str, token: str, path: str):
+def pages(base: str, token: str, path: str | None):
   while path:
     page = fetch(base, token, path)
     yield from page["results"]