From d159081893cfc9be8bb9f187b7b297fb76615115 Mon Sep 17 00:00:00 2001 From: Ken Sanislo Date: Wed, 13 May 2026 08:30:26 -0700 Subject: [PATCH] device: Fix operator precedence bug and end-of-configuration timing in device.changed() (#3173) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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). --- lib/logitech_receiver/device.py | 68 ++++++++++++++++++- lib/logitech_receiver/hidpp20.py | 28 ++++++++ lib/logitech_receiver/notifications.py | 9 ++- .../logitech_receiver/test_hidpp20_simple.py | 41 +++++++++++ 4 files changed, 142 insertions(+), 4 deletions(-) diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index ff3565b7..f5e8a0f4 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -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) diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index d000b409..992b3433 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -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) diff --git a/lib/logitech_receiver/notifications.py b/lib/logitech_receiver/notifications.py index 693e3afd..21ede6c5 100644 --- a/lib/logitech_receiver/notifications.py +++ b/lib/logitech_receiver/notifications.py @@ -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) diff --git a/tests/logitech_receiver/test_hidpp20_simple.py b/tests/logitech_receiver/test_hidpp20_simple.py index f68c0992..b8f88896 100644 --- a/tests/logitech_receiver/test_hidpp20_simple.py +++ b/tests/logitech_receiver/test_hidpp20_simple.py @@ -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"