Preserve button events on camera endpoints and avoid silence-only push faults

custom_components/aqara_lanlink/__init__.py

@@ -11,11 +11,6 @@
 import asyncio
 import logging
 import uuid
-# Import `monotonic` by name, NOT `import time`: this package has a sibling
-# `time.py` (the HA time platform), and once that submodule is imported it
-# binds as `aqara_lanlink.time`, shadowing a module-level `import time` in this
-# __init__ (they share the package namespace). Importing the function avoids it.
-from time import monotonic
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from datetime import timedelta
@@ -45,7 +40,6 @@
     DOMAIN,
     PHONE_ID_NAMESPACE,
     PLATFORMS,
-    PUSH_STALL_TTL_SECONDS,
 )
 from .device import registry
 from .device import catalog as device_catalog
@@ -105,16 +99,8 @@ class AqaraLanLinkRuntimeData:
     subscription_targets: list[tuple[Device, Mapping[str, Any], str, str]] = field(
         default_factory=list,
     )
-    # Monotonic timestamp of the last watchdog-triggered re-arm; used to
-    # rate-limit re-arms to at most once per stall TTL.
-    watchdog_last_rearm: float = 0.0
-    # True while a re-arm pass is running, so overlapping triggers (session-up +
-    # topology-growth on a fresh connect, or the watchdog) coalesce into one.
+    # Coalesce concurrent session-up and topology-growth subscription passes.
     rearm_in_flight: bool = False
-    # Consecutive watchdog re-arms that have not yet produced a report; once it
-    # reaches the threshold a "push_stalled" Repair is raised. Reset to 0 (and
-    # the Repair cleared) as soon as reports resume.
-    stall_rearm_count: int = 0
     # Snapshot of entry.options taken at setup. The update listener compares
     # against it so a data-only write (e.g. persisting the rediscovered LANLink
     # endpoint) does not trigger a full reload -- only an actual options change
@@ -350,81 +336,6 @@ async def _rearm_subscriptions(
         data.rearm_in_flight = False
 
 
-# Number of consecutive failed watchdog re-arms before raising a Repair telling
-# the user the hub likely needs attention (reboot / LAN Control). At one re-arm
-# per TTL this is ~15 minutes of sustained silence.
-STALL_REARM_REPAIR_THRESHOLD = 3
-
-
-def _push_appears_stalled(
-    *,
-    connected: bool,
-    topology_size: int,
-    seconds_since_report: float,
-    ttl: float,
-) -> bool:
-    """True when the tunnel is up and the hub claims a ready topology, yet no
-    report has arrived within ``ttl`` -- i.e. forwarding is silently wedged.
-    Disconnected / empty-topology states are owned by the reconnect and
-    topology-growth re-arm paths, so they are not treated as stalled here.
-    """
-    return bool(connected) and topology_size > 0 and seconds_since_report > ttl
-
-
-async def _watchdog_tick(
-    hass: HomeAssistant, entry: "AqaraLanLinkConfigEntry",
-) -> None:
-    """One watchdog pass: re-arm the subscription if pushes appear stalled."""
-    data = getattr(entry, "runtime_data", None)
-    if data is None:
-        return
-    coordinator = data.hub
-    issue_id = f"push_stalled_{entry.entry_id}"
-    seconds_since = coordinator.seconds_since_last_report()
-    if not _push_appears_stalled(
-        connected=coordinator.connected,
-        topology_size=len(coordinator.lanlink_topology_dids),
-        seconds_since_report=seconds_since,
-        ttl=PUSH_STALL_TTL_SECONDS,
-    ):
-        # Healthy (or recovered): clear any stall escalation.
-        if data.stall_rearm_count:
-            data.stall_rearm_count = 0
-            ir.async_delete_issue(hass, DOMAIN, issue_id)
-        return
-    # Cap: once we have escalated to a Repair, stop re-arming. A wedged hub does
-    # not honor re-subscribes -- its relay table is persistent and only a factory
-    # reset clears it (docs/dev/tunnel-resilience-report.md section 8) -- so
-    # further re-arms are pointless cloud load. The escalation self-clears (above)
-    # the moment reports resume, which restores the re-arm budget.
-    if data.stall_rearm_count >= STALL_REARM_REPAIR_THRESHOLD:
-        return
-    # Cooldown: a single re-arm needs time to take effect (and produce a report);
-    # don't hammer the cloud every keepalive while a hub stays wedged.
-    now = monotonic()
-    if now - data.watchdog_last_rearm < PUSH_STALL_TTL_SECONDS:
-        return
-    data.watchdog_last_rearm = now
-    data.stall_rearm_count += 1
-    _LOGGER.warning(
-        "LANLink pushes appear stalled (no report in %.0fs, topology=%d DIDs); "
-        "re-arming push subscription (attempt %d)",
-        seconds_since, len(coordinator.lanlink_topology_dids),
-        data.stall_rearm_count,
-    )
-    await _rearm_subscriptions(hass, entry)
-    # Re-arming repeatedly without recovery means the hub itself needs attention
-    # (reboot / LAN Control off) -- escalate to a Repair.
-    if data.stall_rearm_count >= STALL_REARM_REPAIR_THRESHOLD:
-        ir.async_create_issue(
-            hass, DOMAIN, issue_id,
-            is_fixable=False,
-            is_persistent=False,
-            severity=ir.IssueSeverity.WARNING,
-            translation_key="push_stalled",
-        )
-
-
 async def _build_device(
     hass: HomeAssistant,
     entry: AqaraLanLinkConfigEntry,
@@ -768,14 +679,10 @@ def _on_session_up() -> None:
 
         coordinator.on_session_up = _on_session_up
 
-        # Push-liveness watchdog, driven off the keepalive cycle (every ~10s):
-        # re-arm if a connected, topology-ready hub stops forwarding past the
-        # stall TTL -- a case neither session-up nor topology-growth re-arm would
-        # catch. _watchdog_tick self-rate-limits via the re-arm cooldown.
-        def _on_keepalive() -> None:
-            hass.async_create_task(_watchdog_tick(hass, entry))
-
-        coordinator.on_keepalive = _on_keepalive
+        # Reports are event-driven; silence is not proof of failed forwarding.
+        # Retire a previous silence-only Repair instead of telling users to
+        # reboot or factory-reset a hub that still forwards physical events.
+        ir.async_delete_issue(hass, DOMAIN, f"push_stalled_{entry.entry_id}")
 
         # Step 3f: Build the self-device for the tunnel host's own traits.
         # Mirrors the per-subentry setup (steps 3a-3e); keep the two in sync.

custom_components/aqara_lanlink/binary_sensor.py

@@ -20,10 +20,11 @@
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
 
-from .const import DOMAIN, PUSH_STALL_TTL_SECONDS
+from .const import DOMAIN, PUSH_REPORT_FRESHNESS_SECONDS
 from .device.base import _should_replay_seed
 from .device.descriptors import BinarySensorDescriptor
 from .entity import AqaraEntity, build_descriptor_entities, setup_descriptor_platform
+from .push_health import forwarding_health
 
 
 async def async_setup_entry(
@@ -128,10 +129,10 @@ async def async_will_remove_from_hass(self) -> None:
 class AqaraHubForwardingHealth(BinarySensorEntity):
     """Diagnostic connectivity sensor: on when the hub is actually forwarding.
 
-    'On' means the tunnel is up, the hub reports a populated LAN-direct
-    topology, and a report (incl. the hub's own heartbeat) has arrived within
-    the push-stall TTL. It flips off exactly in the wedged-forwarding case the
-    watchdog acts on, giving users and automations a signal for it.
+    Recent reports prove forwarding. Silence from event-driven devices does
+    not prove a fault: expose unknown until fresh evidence arrives. Tunnel
+    disconnection is an observed failure and reports off. The diagnostic
+    remains available independently of the monitored connection.
     """
 
     _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
@@ -145,15 +146,13 @@ def __init__(self, hub) -> None:
         self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, hub.did)})
 
     @property
-    def available(self) -> bool:
-        return bool(self._hub.connected)
-
-    @property
-    def is_on(self) -> bool:
-        return (
-            bool(self._hub.connected)
-            and len(self._hub.lanlink_topology_dids) > 0
-            and self._hub.seconds_since_last_report() < PUSH_STALL_TTL_SECONDS
+    def is_on(self) -> bool | None:
+        return forwarding_health(
+            connected=bool(self._hub.connected),
+            topology_size=len(self._hub.lanlink_topology_dids),
+            report_age=(self._hub.seconds_since_last_report()
+                        if self._hub.has_received_report else None),
+            freshness_window=PUSH_REPORT_FRESHNESS_SECONDS,
         )
 
 

custom_components/aqara_lanlink/const.py

@@ -68,11 +68,9 @@
 # explicit model. Matches the G3 base hub firmware id.
 DEFAULT_HUB_MODEL = "lumi.gateway.agl004"
 
-# Push-liveness watchdog TTL (seconds). A connected hub with a populated
-# topology that goes silent past this is treated as wedged-forwarding and its
-# subscription is re-armed. Also surfaced by the tunnel-forwarding diagnostic
-# binary sensor as its staleness threshold.
-PUSH_STALL_TTL_SECONDS = 300.0
+# Window for positive evidence in the forwarding diagnostic. Outside this
+# window an event-driven stream is unknown, not failed.
+PUSH_REPORT_FRESHNESS_SECONDS = 300.0
 
 # Aqara cloud regions, in the order the Aqara apps display them. The
 # legacy RPC `cloud_client.AREAS` table is the authoritative mapping;

custom_components/aqara_lanlink/device/device_types/_fallback.py

@@ -10,7 +10,9 @@
 """
 from __future__ import annotations
 
-from custom_components.aqara_lanlink.device.descriptors import AnyDescriptor
+from custom_components.aqara_lanlink.device.descriptors import (
+    AnyDescriptor,
+)
 from custom_components.aqara_lanlink.device.trait_policy import BUTTON_TRAITS
 from custom_components.aqara_lanlink.device.traits import TraitSpec
 
@@ -33,6 +35,21 @@ def compose(
 
 
 def _descriptor_for_trait(wp: str, spec: TraitSpec) -> AnyDescriptor | None:
+    # ButtonEvent is an event even when Aqara places it on a Camera endpoint
+    # rather than a Doorbell endpoint (e.g. several G4 models). The semantic
+    # trait identity takes precedence over the endpoint's data_type dispatch.
+    if spec.trait_code == "ButtonEvent":
+        if spec.function_code == "Doorbell":
+            from .doorbell import button_event_descriptor
+
+            return button_event_descriptor(spec)
+        if spec.function_code == "Button":
+            # Button.ButtonEvent also appears on Switch endpoints. It keeps
+            # event semantics there even though the endpoint composer is a
+            # switch-specific handler.
+            from .button import button_event_descriptor
+
+            return button_event_descriptor(spec)
     # Press-to-trigger traits (e.g. Identify.IdentifyTime) render as a
     # stateless Button rather than the data_type-dispatched defaults
     # (which would otherwise turn this writable trait into a Switch or

custom_components/aqara_lanlink/device/device_types/button.py

@@ -13,7 +13,8 @@
 from ._base import _ec, make_single_trait_composer
 
 
-def _button_event(spec: TraitSpec) -> EventDescriptor:
+def button_event_descriptor(spec: TraitSpec) -> EventDescriptor:
+    """Preserve button event semantics independently of endpoint type."""
     event_types = (
         tuple(spec.enum_values.values()) if spec.enum_values else ("press",)
     )
@@ -30,5 +31,5 @@ def _button_event(spec: TraitSpec) -> EventDescriptor:
 compose = make_single_trait_composer(
     function_code="Button",
     trait_code="ButtonEvent",
-    descriptor_factory=_button_event,
+    descriptor_factory=button_event_descriptor,
 )

custom_components/aqara_lanlink/device/device_types/doorbell.py

@@ -30,38 +30,47 @@ def compose(
     out: list[AnyDescriptor] = []
     others: dict[str, TraitSpec] = {}
     if button_event is not None:
-        # HA deprecates DOORBELL event entities that don't support the "ring"
-        # event_type (removal in 2027.4), so "ring" must always be present.
-        ev = button_event.enum_values or {}
-        if len(ev) > 1:
-            # Multi-press doorbell (e.g. Single/Double press): the primary
-            # (first) press IS the ring; additional presses keep their
-            # humanized labels so automations can distinguish them.
-            items = list(ev.items())
-            normalised_enum = {items[0][0]: "ring"}
-            extra = []
-            for wire, label in items[1:]:
-                normalised_enum[wire] = label
-                extra.append(label)
-            trigger_trait = dataclasses.replace(button_event, enum_values=normalised_enum)
-            event_types = ("ring", *extra)
-        else:
-            # Single press (or a bare signal with no enum) -> just "ring".
-            normalised_enum = {wire: "ring" for wire in ev} or None
-            trigger_trait = dataclasses.replace(button_event, enum_values=normalised_enum)
-            event_types = ("ring",)
-        out.append(EventDescriptor(
-            key=f"auto_{button_event.id.replace('.', '_')}",
-            name=button_event.name or "Doorbell",
-            trigger_trait=trigger_trait,
-            event_types=event_types,
-            device_class=EventDeviceClass.DOORBELL,
-            entity_category=_ec(button_event),
-            entity_registry_enabled_default=button_event.default_enabled,
-        ))
+        out.append(button_event_descriptor(button_event))
     for wp, spec in traits.items():
         if spec.function_code == "Doorbell" and spec.trait_code == "ButtonEvent":
             continue
         others[wp] = spec
     out.extend(_fallback.compose(endpoint_id, others, context))
     return out
+
+
+def button_event_descriptor(button_event: TraitSpec) -> EventDescriptor:
+    """Build the canonical doorbell event regardless of endpoint type.
+
+    Aqara can place Doorbell.ButtonEvent on a Camera endpoint. Event identity
+    belongs to the trait, not the endpoint's deviceType, so the fallback
+    composer uses this same factory for those models.
+    """
+    # HA deprecates DOORBELL event entities that don't support the "ring"
+    # event_type (removal in 2027.4), so "ring" must always be present.
+    ev = button_event.enum_values or {}
+    if len(ev) > 1:
+        # Multi-press doorbell: the first code is the ring; additional presses
+        # keep their labels so automations can distinguish them.
+        items = list(ev.items())
+        normalised_enum = {items[0][0]: "ring"}
+        extra = []
+        for wire, label in items[1:]:
+            normalised_enum[wire] = label
+            extra.append(label)
+        trigger_trait = dataclasses.replace(button_event, enum_values=normalised_enum)
+        event_types = ("ring", *extra)
+    else:
+        # Single press (including wire code "0") or bare signal -> ring.
+        normalised_enum = {wire: "ring" for wire in ev} or None
+        trigger_trait = dataclasses.replace(button_event, enum_values=normalised_enum)
+        event_types = ("ring",)
+    return EventDescriptor(
+        key=f"auto_{button_event.id.replace('.', '_')}",
+        name=button_event.name or "Doorbell",
+        trigger_trait=trigger_trait,
+        event_types=event_types,
+        device_class=EventDeviceClass.DOORBELL,
+        entity_category=_ec(button_event),
+        entity_registry_enabled_default=button_event.default_enabled,
+    )

custom_components/aqara_lanlink/hub/coordinator.py

@@ -149,9 +149,8 @@ class HubCoordinator:
     # after a successful checkin. The hub's push subscription is per-connection,
     # so the integration uses this to re-arm subscribe+seed on every reconnect.
     on_session_up: "Callable[[], None] | None" = None
-    # Invoked (no args) on each keepalive ack from the hub. The integration uses
-    # this ~10s cadence to run the push-liveness watchdog without a wall-clock
-    # timer. A None value means no callback is installed.
+    # Invoked (no args) on each keepalive ack from the hub. This confirms the
+    # session, not the event-report forwarding path.
     on_keepalive: "Callable[[], None] | None" = None
 
     def __init__(
@@ -200,11 +199,11 @@ def __init__(
         # ``async_get_topology`` for cloud-vs-LANLink cross-checking.
         self._lanlink_topology_dids: frozenset[str] = frozenset()
         # Monotonic timestamp of the most recent inbound report (any report,
-        # including the hub's own 1.176.20009.1 heartbeat). The push-liveness
-        # watchdog uses ``seconds_since_last_report`` to detect a hub that is
-        # connected with a populated topology yet has silently stopped
-        # forwarding. Seeded to construction time so we don't fire immediately.
+        # including the hub's own 1.176.20009.1 heartbeat). This is activity
+        # evidence, not a guaranteed heartbeat schedule. Construction time
+        # supplies an age for diagnostics but does not count as a report.
         self._last_report_monotonic = time.monotonic()
+        self.has_received_report = False
         # Union of every DID ever seen in a topology push. Lets us tell a
         # LAN-direct device that has gone offline (was present, now dropped)
         # apart from a Zigbee sub-device that is never in the LAN-direct
@@ -593,6 +592,8 @@ async def _run_session_once(self) -> None:
         client.on_report = self._dispatch_report
         client.on_async = self._dispatch_async
         self._client = client
+        # A previous session's report does not prove this session forwards.
+        self.has_received_report = False
 
         listen_task = asyncio.create_task(
             client.listen(), name="lanlink-listen",
@@ -720,6 +721,7 @@ def _dispatch_report(self, report: LANReport) -> None:
         rid-keyed settings, which receive no reports at all.
         """
         self._last_report_monotonic = time.monotonic()
+        self.has_received_report = True
         _LOGGER.debug(
             "LANLink report did=%s sdid=%s values=%s",
             report.did, report.sdid, report.values,

custom_components/aqara_lanlink/push_health.py

@@ -0,0 +1,21 @@
+"""Evidence-based health for an event-driven LANLink report stream."""
+from __future__ import annotations
+
+
+def forwarding_health(
+    *, connected: bool, topology_size: int,
+    report_age: float | None, freshness_window: float,
+) -> bool | None:
+    """Return recent push evidence, a disconnected tunnel, or unknown.
+
+    No supported protocol contract guarantees a periodic report. A quiet
+    doorbell can therefore provide no current evidence without being broken.
+    Keepalive replies prove the tunnel, not the device-report forwarding path.
+    """
+    if not connected:
+        return False
+    if topology_size <= 0 or report_age is None:
+        return None
+    if report_age < freshness_window:
+        return True
+    return None