device: Fix operator precedence bug and end-of-configuration timing in device.changed() (#3173)

* Fix WIRELESS_DEVICE_STATUS reconfiguration: push settings and ack with proper cookie

When a device sends WIRELESS_DEVICE_STATUS with config_needed=1, the host
should re-push settings and acknowledge via ConfigChange SetComplete. The
existing code relied on the push flag in changed(), but operator precedence
caused push=True to be ignored for devices that have the WIRELESS_DEVICE_STATUS
feature — the exact devices that send this notification.

Handle the reconfiguration entirely in the notification handler: explicitly
push settings and ack, rather than trying to overload the changed() condition.

Also replace hardcoded cookie 0x11 with proper GetCookie/SetComplete protocol:
read the device's current configuration cookie and echo it back, per the
ConfigChange (0x0020) specification.

* WIRELESS_DEVICE_STATUS reconfig: gate apply on ConfigChange cookie

Add a cookie-comparison gate so devices that emit multiple reconfig
notifications (PRO X 2 sends two on power-on) don't get their settings
re-applied for every one. The gate also benefits any path that calls
apply_settings_if_needed — devices that retain config through
power-save now skip the redundant apply.

New flow:

- Device.apply_settings_if_needed() reads ConfigChange (0x0020) cookie
  via GetCookie. If the live value matches the cookie we stored after
  the last successful sync (persister key `_config_cookie`), the
  device hasn't drifted and we skip apply_all_settings entirely. On a
  real apply, SetComplete echoes the cookie back and we persist it.
  Devices without CONFIG_CHANGE bypass the gate (always apply).

- device.changed(active=True) calls apply_settings_if_needed in place
  of the old apply_all_settings + signal_configuration_complete pair.

- WIRELESS_DEVICE_STATUS reconfig handler drops the redundant explicit
  apply + ack (changed() already handles it on transition to active)
  and instead calls apply_settings_if_needed to cover the case of
  follow-up reconfig notifications on an already-active device — the
  cookie gate makes the second/third call essentially free.

Protocol cleanup:

- set_configuration_complete no longer auto-increments the cookie
  before echoing. The device owns the cookie and bumps it when its
  config drifts; the host's job is to confirm which value it synced
  with, not to advance the value. Host-side increment introduced a
  needless race with device-side bumps.

- signal_configuration_complete gains an optional cookie= arg so the
  caller can pass a value it already read (saves a redundant GetCookie
  round-trip from inside the gated apply path).

Tests:

- get_configuration_cookie returns the two-byte cookie from fn 0.
- set_configuration_complete echoes a provided cookie unchanged.
- set_configuration_complete with cookie=None reads the live cookie
  and echoes it (no increment).
This commit is contained in:
Ken Sanislo 2026-05-13 08:30:26 -07:00 committed by GitHub
parent ad96dd395b
commit d159081893
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 142 additions and 4 deletions

View File

@ -437,6 +437,65 @@ class Device:
if self.online and self.protocol >= 2.0:
_hidpp20.config_change(self, configuration_, no_reply=no_reply)
def signal_configuration_complete(self, cookie=None):
"""SetComplete on ConfigChange to ack end of configuration.
With no cookie, sends the host's session counter (see
Hidpp20.set_configuration_complete)."""
if self.online and self.protocol >= 2.0:
_hidpp20.set_configuration_complete(self, cookie=cookie)
def _record_config_cookie(self):
"""After a successful apply, SetComplete with the next session cookie
and persist it so the dedup gate in apply_settings_if_needed can
detect drift on follow-up reconfig notifications within this session."""
if self.protocol < 2.0:
return
if not (self.features and SupportedFeature.CONFIG_CHANGE in self.features):
return
cookie = _hidpp20.next_session_cookie()
self.signal_configuration_complete(cookie=cookie)
if self.persister is not None:
self.persister["_config_cookie"] = [cookie[0], cookie[1]]
def apply_settings_if_needed(self):
"""Cookie-gated dedup helper for repeat WIRELESS_DEVICE_STATUS
reconfig notifications on an already-active device. Skips when the
live ConfigChange cookie matches the value stored by the most
recent apply, otherwise applies and re-records. Must NOT be used as
the initial-activation apply path across power cycles, devices
whose firmware resets the cookie to a fixed value would falsely
match a stored cookie from a prior session and skip the apply the
device actually needs.
Returns True if apply ran, False if it was skipped."""
if not self.online:
return False
if self.protocol >= 2.0 and self.features and SupportedFeature.CONFIG_CHANGE in self.features:
live = _hidpp20.get_configuration_cookie(self)
if live and len(live) >= 2:
stored = self.persister.get("_config_cookie") if self.persister else None
live_pair = [live[0], live[1]]
if stored == live_pair:
if logger.isEnabledFor(logging.INFO):
logger.info(
"%s: config cookie %02X%02X matches stored — skip apply_all_settings",
self,
live[0],
live[1],
)
return False
if logger.isEnabledFor(logging.INFO):
logger.info(
"%s: config cookie live=%02X%02X stored=%s — apply_all_settings",
self,
live[0],
live[1],
"%02X%02X" % (stored[0], stored[1]) if stored else "None",
)
settings.apply_all_settings(self)
self._record_config_cookie()
return True
def reset(self, no_reply=False):
self.set_configuration(0, no_reply)
@ -536,12 +595,17 @@ class Device:
):
if logger.isEnabledFor(logging.INFO):
logger.info("%s pushing device settings %s", self, self.settings)
# Activation apply must be unconditional — across power
# cycles, the device may have lost state while its cookie
# reset to a value that happens to match what we stored
# last session. Cookie comparison is only a valid dedup
# signal for repeat reconfig notifications within an
# already-active session (see apply_settings_if_needed).
settings.apply_all_settings(self)
self._record_config_cookie()
if not was_active:
if self.protocol < 2.0: # Make sure to set notification flags on the device
self.notification_flags = self.enable_connection_notifications()
else:
self.set_configuration(0x11) # signal end of configuration
self.read_battery() # battery information may have changed so try to read it now
elif was_active and self.receiver and not isinstance(self.receiver, CenturionReceiver):
hidpp10.set_configuration_pending_flags(self.receiver, 0xFF)

View File

@ -24,6 +24,7 @@ import threading
from collections import UserDict
from enum import Flag
from enum import IntEnum
from random import getrandbits
from typing import Any
from typing import Dict
from typing import Generator
@ -1666,6 +1667,11 @@ def feature_request(device, feature, function=0x00, *params, no_reply=False):
class Hidpp20:
# Host-side counter for SetComplete cookies (see set_configuration_complete).
# Seeded to a non-zero random 16-bit value at import so successive sessions
# don't trivially collide; we just need to never send 0x0000.
_session_cookie = getrandbits(16) or 1
def get_firmware(self, device) -> tuple[common.FirmwareInfo] | None:
"""Reads a device's firmware info.
@ -2051,7 +2057,29 @@ class Hidpp20:
return struct.unpack("!B", result[:1])[0]
return None
def get_configuration_cookie(self, device: Device):
"""ConfigChange (0x0020) GetCookie — read the device's current configuration cookie."""
response = device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x00)
return response[:2] if response else None
def next_session_cookie(self):
"""Bump and return the host-side counter used as the SetComplete cookie."""
Hidpp20._session_cookie = (Hidpp20._session_cookie + 1) & 0xFFFF or 1
return bytes([Hidpp20._session_cookie >> 8, Hidpp20._session_cookie & 0xFF])
def set_configuration_complete(self, device: Device, cookie=None, no_reply=False):
"""ConfigChange (0x0020) SetComplete — acknowledge host has synced with device configuration.
Sends a host-side monotonic counter, incremented per call and
always non-zero. Cookie 0x0000 has been observed to release the
SW effect-engine claim on at least the G515 LS TKL; we avoid it."""
if cookie is None:
cookie = self.next_session_cookie()
if cookie and len(cookie) >= 2:
return device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x10, cookie[0], cookie[1], no_reply=no_reply)
def config_change(self, device: Device, configuration, no_reply=False):
"""Deprecated — use set_configuration_complete() instead."""
return device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x10, configuration, no_reply=no_reply)

View File

@ -322,9 +322,14 @@ def _process_feature_notification(device: Device, notification: HIDPPNotificatio
if logger.isEnabledFor(logging.DEBUG):
logger.debug("wireless status: %s", notification)
reason = "powered on" if notification.data[2] == 1 else None
if notification.data[1] == 1: # device is asking for software reconfiguration so need to change status
if notification.data[1] == 1: # device is asking for software reconfiguration
alert = Alert.NONE
device.changed(active=True, alert=alert, reason=reason, push=True)
device.changed(active=True, alert=alert, reason=reason)
# changed(active=True) already runs apply_settings_if_needed on
# the first transition; for follow-up reconfig notifications
# on an already-active device, fire the gate here so the
# cookie comparison decides whether to re-push.
device.apply_settings_if_needed()
else:
logger.warning("%s: unknown WIRELESS %s", device, notification)

View File

@ -372,6 +372,47 @@ def test_config_change():
assert result == bytes.fromhex("03FFFF")
def test_get_configuration_cookie():
responses = [fake_hidpp.Response("12345678", 0x0400)]
device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE)
result = _hidpp20.get_configuration_cookie(device)
assert result == bytes.fromhex("1234")
def test_set_configuration_complete_explicit_cookie():
# An explicit cookie is sent unchanged.
responses = [fake_hidpp.Response("00", 0x0410, "1234")]
device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE)
result = _hidpp20.set_configuration_complete(device, cookie=bytes.fromhex("1234"))
assert result == bytes.fromhex("00")
def test_set_configuration_complete_monotonic_counter():
# With no cookie, sends a host-side monotonic counter, +1 per call.
hidpp20.Hidpp20._session_cookie = 0x1233
responses = [
fake_hidpp.Response("00", 0x0410, "1234"),
fake_hidpp.Response("00", 0x0410, "1235"),
]
device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE)
assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00")
assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00")
def test_set_configuration_complete_skips_zero():
# Counter at 0xFFFF wraps to 1, never 0.
hidpp20.Hidpp20._session_cookie = 0xFFFF
responses = [fake_hidpp.Response("00", 0x0410, "0001")]
device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE)
assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00")
def test_decipher_battery_status():
report = b"\x50\x20\x00\xff\xff"