Refactor handshake logic and improve comments for clarity

This commit is contained in:
Prabhaav Pillai
2026-11-05 14:40:57 -05:00
parent b9d6539aa7
commit 55eae09ad4
11 changed files with 121 additions and 43 deletions
+1 -1
View File
@@ -2439,7 +2439,7 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
if action == 'start' and set(payload) == {'action'}:
command, arguments = 'start', {}
elif action == 'reset_handshake' and set(payload) == {'action'}:
command, arguments = 'reset_handshake', {} # keeps the pairing; the next attempt starts at the standard handshake
command, arguments = 'reset_handshake', {}
elif action == 'auto_connect' and set(payload) == {'action', 'enabled'} and type(payload['enabled']) is bool:
if not configuration_allowed():
self.json(403, {'error': 'Park before changing automatic connection'})
@@ -95,8 +95,7 @@ assert.equal(guidedState({ setup: null, runtime: running("rfcomm") }).show, fals
assert.deepEqual(handshakeSteps({ steps: [null, { label: 3 }, { id: "a", label: "A", state: "worked" }] }).map((step) => step.id), ["a"])
}
// A handshake that moved off the standard one (2026-11-04: stuck on a reply the Civic hangs up on) can be started over
// without re-pairing; a fresh or learned-standard car offers nothing to reset.
// Fresh and learned-standard handshakes need no reset; alternatives and failed rungs do.
{
const steps = (...states) => states.map((state, index) => ({ id: ["standard", "other_reply"][index], label: "x", state }))
assert.equal(handshakeResettable(null), false)
@@ -65,7 +65,7 @@ export function handshakeSteps(ladder) {
return interesting ? steps.map((step) => ({ id: String(step.id), label: step.label, state: String(step.state) })) : []
}
// A car whose handshake has moved off the standard one, or has rungs marked failed, can be started over without re-pairing.
// Reset learned or failed handshakes without re-pairing.
export function handshakeResettable(ladder) {
if (!ladder || typeof ladder !== "object") return false
const steps = Array.isArray(ladder.steps) ? ladder.steps : []
@@ -287,6 +287,7 @@ class WirelessBootstrap:
self.version_major = 0
self.refusals: list[int] = [] # WifiStartResponse statuses that were not success, in order
self.replied_at: float | None = None # monotonic time our WifiVersionResponse went out
self._reply_accepted = False
self.flags = EXPERIMENTS[self.experiment][1] if self.experiment else set()
self.initial_kick_delay = initial_kick_delay
self.join_ping_interval = join_ping_interval
@@ -309,10 +310,11 @@ class WirelessBootstrap:
return bool(self.refusals) and all(status == STATUS_NOT_READY for status in self.refusals)
def hung_up_on_reply(self, error: BaseException) -> bool:
"""The car closed the link right after our version reply, before refusing anything: it rejected the reply."""
"""A close immediately after the version reply, before Wi-Fi setup progressed."""
closed = isinstance(error, (BootstrapClosed, ConnectionResetError, BrokenPipeError))
return (closed and self.version_major > 0 and self.replied_at is not None and not self.refusals
and time.monotonic() - self.replied_at <= REPLY_REJECT_SECONDS)
return (closed and self.stage == "wifi_start" and not self._reply_accepted and self.version_major > 0
and self.replied_at is not None and not self.refusals
and 0 <= time.monotonic() - self.replied_at <= REPLY_REJECT_SECONDS)
def join_is_cancelled(self) -> bool:
"""For join_wifi: stop on user cancellation or once the handshake has failed."""
@@ -326,7 +328,7 @@ class WirelessBootstrap:
self._log_hex("tx", message_id, payload)
def _log_hex(self, direction: str, message_id: int, payload: bytes) -> None:
if self.experiment and message_id not in HEX_LOG_SKIP:
if self.experiment and message_id not in HEX_LOG_SKIP and not (self.phone_identity and message_id == WIFI_VERSION_RESPONSE):
self.log("bootstrap_frame_hex", direction=direction, message=NAMES.get(message_id, message_id), hex=payload.hex())
def next_frame(self, timeout: float) -> tuple[int, bytes]:
@@ -377,7 +379,7 @@ class WirelessBootstrap:
if self.phone_address:
identity += field(4, self.phone_address)
reply = field(1, major) + field(2, minor) + field(3, self.lifetime_id[:8]) + field(4, 0) + field(6, identity)
self.log("bootstrap_version_reply", style="phone_identity", has_address=bool(self.phone_address), hex=reply.hex())
self.log("bootstrap_version_reply", style="phone_identity", has_address=bool(self.phone_address))
return reply
def service(self, message_id: int, payload: bytes) -> bool:
@@ -446,6 +448,7 @@ class WirelessBootstrap:
endpoint=version_endpoint.__dict__ if version_endpoint else None, message=describe_version_request(payload))
self.send(WIFI_VERSION_RESPONSE, self._version_response(major, minor, payload))
self.replied_at = time.monotonic()
self._reply_accepted = False
hinted = version_endpoint or hinted
if hinted is not None and hint_ready_at is None:
hint_ready_at = time.monotonic() + 3.0
@@ -456,6 +459,7 @@ class WirelessBootstrap:
if parsed is None:
raise BootstrapError(self.stage, "WifiStartRequest has no valid IPv4 endpoint")
endpoint = parsed
self._reply_accepted = True
if "start_response_first" in self.flags:
self.send(WIFI_START_RESPONSE, field(3, STATUS_SUCCESS))
elif message_id == WIFI_START_RESPONSE:
@@ -465,6 +469,7 @@ class WirelessBootstrap:
parsed = parse_endpoint(payload)
if parsed is not None and status == STATUS_SUCCESS:
endpoint = parsed
self._reply_accepted = True
else:
self.log("bootstrap_start_refused", status=status, endpoint=parsed is not None)
self.refusals.append(status)
@@ -472,6 +477,8 @@ class WirelessBootstrap:
start_request_at = time.monotonic() + self.start_request_delay
elif message_id == WIFI_SETUP_INFO:
setup_endpoint, setup_credentials = parse_setup_info(payload)
if setup_endpoint is not None or setup_credentials is not None:
self._reply_accepted = True
hinted = setup_endpoint or hinted
if hinted is not None and hint_ready_at is None:
hint_ready_at = time.monotonic() + 3.0
@@ -483,6 +490,7 @@ class WirelessBootstrap:
# One that cannot be read is ignored as before, and the network is asked for after the endpoint.
try:
credentials = parse_info_response(payload)
self._reply_accepted = True
except ValueError as error:
self.log("bootstrap_early_info_unreadable", error=str(error))
else:
@@ -220,9 +220,7 @@ def session_setup(events: list[dict]) -> dict:
def session_ladder(events: list[dict]) -> list[dict]:
"""The handshake strategies tried in a session, in order, and how each ended (advanced, wrapped, exhausted, counted,
dropped, reverted, or worked), why (a timeout, or the car hanging up on the reply), and how many waits ended with the
car only saying "not ready". Kept for working pairings too: it is what shows which strategy a head unit likes."""
"""Summarize attempted handshakes, outcomes, failure reasons and busy waits."""
tried: list[dict] = []
for event in events:
name = event["event"]
@@ -286,7 +286,7 @@ def _tail(path: Path, limit: int) -> bytes | None:
def shareable_setup_state(path: Path, chosen: str = "") -> dict | None:
"""The handshake ladder's saved state with car addresses replaced by whether each is the chosen car."""
try:
data = json.loads(path.read_text())
data = json.loads(read_log(path, MAX_LOG_BYTES))
except (OSError, ValueError, UnicodeError):
return None
cars = data.get("cars") if isinstance(data, dict) else None
@@ -1,10 +1,7 @@
"""Try wireless handshakes in order and remember a strategy after sustained video.
"""Learn wireless handshakes from sustained video and qualified bootstrap failures.
Only bootstrap timeouts after a version exchange advance the ladder, and a car that hangs up the instant it gets our
version reply rejects that rung outright; failures before the car answers, a car that only ever says "not ready", and
anything after acknowledged video leave its strategy unchanged.
The last strategy that streamed is kept until the car is forgotten or the ladder is reset; a rejected rung and each new drive go back to it.
Busy cars and failures after video do not advance the ladder. Preserve the last
working strategy across drives until pairing is forgotten or the ladder is reset.
"""
from __future__ import annotations
@@ -185,7 +182,7 @@ class SetupLadder:
return True
def reset(self, address: str) -> bool:
"""Start this car's handshake over from the standard one, keeping its pairing (the setup screen's reset)."""
"""Reset learned handshakes without deleting the Bluetooth pairing."""
return self.forget(address)
def forget(self, address: str) -> bool:
@@ -211,10 +208,7 @@ class SetupLadder:
def failed(self, address: str, strategy: Strategy, *, ladder_failure: bool, video_seen: bool, version_major: int,
adapter: bool, rejected: bool = False, unofficial: bool = False) -> str | None:
"""Record a failed attempt; returns what changed ("advanced", "wrapped", "exhausted", "counted", "dropped",
"reverted") or None.
Only a handshake the car answered, before any video this session, counts: see the module docstring."""
"""Record a qualified failure and return the ladder transition, or None."""
if not (ladder_failure or rejected) or video_seen:
return None
entry = self._entry(address)
@@ -624,6 +624,7 @@ class Supervisor:
replies = dict(self.config.get("modern_reply_cache", {}))
reset = bool(address) and (self._ladder.reset(address) | (replies.pop(address.upper(), None) is not None))
if reset:
self._strategy = None
self.config["modern_reply_cache"] = replies
self._modern_reply.pop(address.upper(), None)
identity_store.save_config(self.config)
@@ -1494,9 +1495,10 @@ class Supervisor:
unofficial = self._unofficial_dongle()
# The ladder decides how to answer this car: what worked before, else the first rung (today's behavior), moving on
# only when a car that answered the handshake keeps refusing it before any video.
base_modern = self._modern_reply.get(address, config.get("modern_reply_cache", {}).get(address, False))
strategy = self._strategy = self._ladder.strategy(address, base_modern)
learned = self._ladder.snapshot(address, adapter, unofficial)["learned"]
with self._lock:
base_modern = self._modern_reply.get(address, config.get("modern_reply_cache", {}).get(address, False))
strategy = self._strategy = self._ladder.strategy(address, base_modern)
learned = self._ladder.snapshot(address, adapter, unofficial)["learned"]
self.log("ladder_strategy", rung=strategy.rung, label=strategy.label, learned=learned, modern_reply=strategy.modern_reply,
channel_type=strategy.channel_type, patience=strategy.patience, experiments=strategy.experiments,
phone_identity=strategy.phone_identity)
@@ -1521,15 +1523,19 @@ class Supervisor:
# Only a silent or refusing car after the version exchange, or one that hangs up on our reply, counts against the handshake.
stopped = self._stop.is_set()
ladder_failure = (boot is not None and boot.version_major > 0 and isinstance(error, BootstrapTimeout)
and error.stage in ("wifi_start", "wifi_info") and not boot.only_not_ready() and not stopped)
and error.stage in ("wifi_start", "wifi_info")
and not (error.stage == "wifi_start" and boot.only_not_ready()) and not stopped)
rejected = boot is not None and not stopped and boot.hung_up_on_reply(error)
outcome = self._ladder.failed(address, strategy, ladder_failure=ladder_failure, rejected=rejected, video_seen=self._video_seen,
version_major=boot.version_major if boot is not None else 0, adapter=adapter, unofficial=unofficial)
if outcome:
self.log("ladder_failed", outcome=outcome, rung=strategy.rung, next=self._ladder.strategy(address, base_modern).rung,
version_major=boot.version_major, reason="rejected_reply" if rejected else "timeout")
elif boot is not None and isinstance(error, BootstrapTimeout) and boot.only_not_ready():
self.log("ladder_not_ready", rung=strategy.rung, refusals=len(boot.refusals))
with self._lock:
current = self._strategy is strategy
outcome = self._ladder.failed(address, strategy, ladder_failure=ladder_failure and current, rejected=rejected and current,
video_seen=self._video_seen, version_major=boot.version_major if boot is not None else 0,
adapter=adapter, unofficial=unofficial)
if outcome:
self.log("ladder_failed", outcome=outcome, rung=strategy.rung, next=self._ladder.strategy(address, base_modern).rung,
version_major=boot.version_major, reason="rejected_reply" if rejected else "timeout")
elif boot is not None and isinstance(error, BootstrapTimeout) and error.stage == "wifi_start" and boot.only_not_ready():
self.log("ladder_not_ready", rung=strategy.rung, refusals=len(boot.refusals))
if source == "cache" and not self._stop.is_set() and drops_cached_channel(self._status["last_stage"], error):
self._remember_channel(address, None) # the car never answered there; ask it over SDP next time
self.log("rfcomm_cache_dropped", channel=channel)
@@ -1545,8 +1551,10 @@ class Supervisor:
self._project(result, lease, ident)
finally:
keepalive_stop.set()
if boot.version_major >= MODERN_MAJOR and self._ladder.snapshot(address, adapter, unofficial)["learned"]:
self._remember_reply(address, boot.modern_reply)
with self._lock:
if (self._strategy is strategy and boot.version_major >= MODERN_MAJOR and
self._ladder.snapshot(address, adapter, unofficial)["learned"]):
self._remember_reply(address, boot.modern_reply)
def _connect_rfcomm(self, address: str, channel: int):
from openpilot.starpilot.system.starport import bt_sockets
@@ -1885,13 +1893,14 @@ class Supervisor:
if not session.video_confirmed:
return
self._video_seen = True
strategy, address = self._strategy, self.config["receiver_address"]
streamed = time.monotonic() - session.video_confirmed_at
if strategy is None or not address or streamed < setup_ladder.CLEAN_SECONDS:
return
self._ladder.worked(address, strategy, t_ready_s=session.video_confirmed_at - self._run_started)
self.log("ladder_worked", rung=strategy.rung, streamed_s=round(streamed, 1),
t_ready_s=round(session.video_confirmed_at - self._run_started, 1))
with self._lock:
strategy, address = self._strategy, self.config["receiver_address"]
if strategy is None or not address or streamed < setup_ladder.CLEAN_SECONDS:
return
self._ladder.worked(address, strategy, t_ready_s=session.video_confirmed_at - self._run_started)
self.log("ladder_worked", rung=strategy.rung, streamed_s=round(streamed, 1),
t_ready_s=round(session.video_confirmed_at - self._run_started, 1))
# ANDROID AUTO VIDEO PERFORMANCE GUARDRAIL: The bounded ACK/input drain and conditional waits keep video and touches responsive.
# Altering this can degrade video smoothness or introduce lag on Android Auto.
@@ -214,3 +214,23 @@ def test_the_phone_identity_reply_needs_protocol_four_and_leaves_out_an_unknown_
boot.modern_reply, boot.phone_identity = True, True
reply = bs.parse_fields(boot._version_response(4, 0, MAVLOU_V40))
assert 4 not in bs.parse_fields(reply[6][0])
def test_phone_identity_is_sent_but_never_hex_logged_even_with_experiments():
records = []
address = "AA:BB:CC:DD:EE:FF"
boot = bs.WirelessBootstrap(Mock(), lambda event, **values: records.append((event, values)),
modern_reply=True, phone_identity=True, phone_address=address, experiment="all")
reply = boot._version_response(4, 0, MAVLOU_V40)
boot.send(bs.WIFI_VERSION_RESPONSE, reply)
assert address.encode() in reply
assert records and not any("hex" in values for _, values in records)
boot.sock.sendall.assert_called_once()
def test_early_wifi_credentials_show_the_version_reply_was_accepted():
boot, _ = scripted([(bs.WIFI_VERSION_REQUEST, MAVLOU_V40), (bs.WIFI_INFO_RESPONSE, INFO),
bs.BootstrapClosed("wifi_start", "closed after credentials")])
with pytest.raises(bs.BootstrapClosed) as error:
boot.run(slow_join(0, []))
assert boot.stage == "wifi_start" and not boot.hung_up_on_reply(error.value)
@@ -488,3 +488,14 @@ def test_bundle_without_a_ladder_state_has_no_setup_state_file(tmp_path):
write_log(logs, "session-000025-20260928-141351.jsonl", CIVIC_SESSION)
with zipfile.ZipFile(io.BytesIO(compat_report.bundle(logs, tmp_path / "config.json", snapshot=None))) as archive:
assert "setup_state.json" not in archive.namelist()
def test_setup_state_export_uses_bounded_regular_file_reads(tmp_path, monkeypatch):
path = tmp_path / "setup_state.json"
path.write_text(json.dumps({"version": 1, "cars": {}}))
assert compat_report.shareable_setup_state(path) == {"version": 1, "cars": []}
link = tmp_path / "state-link.json"
link.symlink_to(path)
assert compat_report.shareable_setup_state(link) is None
monkeypatch.setattr(compat_report, "MAX_LOG_BYTES", 8)
assert compat_report.shareable_setup_state(path) is None
@@ -381,6 +381,7 @@ def test_bootstrap_tells_a_busy_car_from_one_that_hangs_up_on_the_reply():
assert boot.only_not_ready()
boot.refusals.append(-4)
assert not boot.only_not_ready()
boot.stage = "wifi_start"
boot.refusals, boot.version_major, boot.replied_at = [], 4, _time.monotonic()
assert boot.hung_up_on_reply(ConnectionResetError(104, "Connection reset by peer"))
assert boot.hung_up_on_reply(bs.BootstrapClosed("rfcomm", "head unit closed the RFCOMM connection"))
@@ -391,6 +392,14 @@ def test_bootstrap_tells_a_busy_car_from_one_that_hangs_up_on_the_reply():
assert not boot.hung_up_on_reply(ConnectionResetError()) # it was already talking Wi-Fi setup
@pytest.mark.parametrize("stage", ["wifi_info", "joining_wifi", "connecting_tcp"])
def test_a_close_after_startup_was_accepted_does_not_reject_the_reply(stage):
from openpilot.starpilot.system.starport import bootstrap as bs
boot = bs.WirelessBootstrap(None, lambda *a, **k: None)
boot.stage, boot.version_major, boot.replied_at = stage, 4, _time.monotonic()
assert not boot.hung_up_on_reply(bs.BootstrapClosed(stage, "closed"))
def _failing_bootstrap(monkeypatch, sup, error, refusals=()):
from openpilot.starpilot.system.starport import bootstrap as bs
from openpilot.starpilot.system.starport import supervisor as module
@@ -402,6 +411,7 @@ def _failing_bootstrap(monkeypatch, sup, error, refusals=()):
replies.append(kwargs["modern_reply"])
def run(self, *args, **kwargs):
self.stage = "wifi_start"
self.version_major, self.replied_at, self.refusals = 4, _time.monotonic(), list(refusals)
raise error(self)
@@ -423,6 +433,35 @@ def test_a_car_that_only_says_not_ready_keeps_its_learned_handshake(sup, monkeyp
assert entry["working"]["rung"] == "standard" and entry["working_failures"] == 0
def test_a_later_wifi_info_timeout_counts_despite_an_earlier_busy_response(sup, monkeypatch):
from openpilot.starpilot.system.starport.bootstrap import STATUS_NOT_READY, BootstrapTimeout
_failing_bootstrap(monkeypatch, sup, lambda boot: BootstrapTimeout("wifi_info", "no credentials"), refusals=[STATUS_NOT_READY])
with pytest.raises(BootstrapTimeout):
sup._attempt(SimpleNamespace())
assert sup._ladder.strategy(RECEIVER, False).rung == "other_reply"
assert not any(record["event"] == "ladder_not_ready" for record in sup.log.recent)
def test_a_reset_during_an_attempt_cannot_be_undone_by_its_failure(sup, monkeypatch):
from openpilot.starpilot.system.starport.bootstrap import BootstrapTimeout
def reset_then_fail(boot):
assert sup.reset_handshake()
return BootstrapTimeout("wifi_start", "timeout after reset")
_failing_bootstrap(monkeypatch, sup, reset_then_fail)
with pytest.raises(BootstrapTimeout):
sup._attempt(SimpleNamespace())
assert RECEIVER not in sup._ladder.cars
def test_a_reset_during_projection_cannot_relearn_the_old_strategy(sup):
strategy = sl.resolve("other_reply", False)
sup._ladder.worked(RECEIVER, strategy)
sup._strategy, sup._run_started = strategy, _time.monotonic() - 150
assert sup.reset_handshake()
sup._note_streamed(streamed(90))
assert RECEIVER not in sup._ladder.cars
def test_the_supervisor_reverts_when_the_car_hangs_up_on_the_other_reply(sup, monkeypatch):
sup._ladder.worked(RECEIVER, sl.resolve("standard", False), 18.0)
for _ in range(2): # a silent car really did drop it