Provider State Management

Provider State Management

This guide explains how Lock Code Manager providers expose user/credential state to the coordinator and the sync seam: the domain model providers speak in, the three update modes (push / poll / drift), and the exception-routing contract.

Overview

The LockUsercodeUpdateCoordinator manages state through three update modes:

Mode	Mechanism	When to Use
Poll for updates	Periodic async_get_usercodes()	Integrations without real-time events (Akuvox, Schlage, Virtual)
Push for updates	Real-time event subscription	Integrations with push (Z-Wave JS, Matter, ZHA, Zigbee2MQTT)
Poll for drift	Periodic async_hard_refresh_codes()	Catch out-of-band changes (keypad, other controllers)

All modes include an initial poll to populate coordinator data.

Important: even with push enabled, you must implement async_get_users() (or the legacy async_get_usercodes()). The coordinator calls it for the initial load and any manual refresh request.

Coordinator Data Model

Coordinator data is dict[int, SlotCredential]. Each entry maps an LCM slot number to a SlotCredential:

• SlotCredential.empty() — slot has no credential
• SlotCredential.unreadable() — slot is occupied but the value is masked or write-only
• SlotCredential.known(pin) — slot has this readable PIN value

Providers expose this in two ways:

1. async_get_users() -> list[User] (the new path, post-4.0.0) — return a list of User objects each carrying credentials: list[Credential]. The base class's async_get_usercodes() projects these to dict[int, SlotCredential] via user.pin_credentials. Native-user providers map the lock's actual users; slot-only providers synthesize a single-credential user per occupied slot using user_from_slot(slot, state).
2. async_get_usercodes() -> dict[int, SlotCredential] (default-implemented) — the projection layer. Override only if you have a faster path to slot data than walking users.

Provider Architecture

┌─────────────────────────────────────────────────────────────────┐
│                LockUsercodeUpdateCoordinator                    │
├─────────────────────────────────────────────────────────────────┤
│  data: dict[int, SlotCredential]       # slot -> credential     │
├─────────────────────────────────────────────────────────────────┤
│  async_get_usercodes()                 # poll method            │
│  push_update({slot: SlotCredential})   # push entry point       │
│  _async_drift_check()                  # hard refresh timer     │
│  _async_connection_check()             # connection poll timer  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              │ calls (via BaseLock seam)
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                         BaseLock                                │
├─────────────────────────────────────────────────────────────────┤
│  Properties:                                                    │
│    supports_push: bool                  # opt in to push mode   │
│    supports_native_users: bool          # user lifecycle path   │
│    usercode_scan_interval: timedelta    # polling interval      │
│    hard_refresh_interval: timedelta     # drift detection       │
│    connection_check_interval: timedelta # connection polling    │
├─────────────────────────────────────────────────────────────────┤
│  Always implement:                                              │
│    async_get_users() -> list[User]                              │
│    async_set_credential(user_id, credential, pin, ...) -> bool  │
│    async_delete_credential(ref) -> bool                         │
│    is_integration_connected() -> bool                           │
├─────────────────────────────────────────────────────────────────┤
│  Native-user providers also implement:                          │
│    async_set_user(user) -> SetUserResult                        │
│    async_delete_user(user_id) -> None                           │
│    async_get_capabilities() -> LockCapabilities                 │
├─────────────────────────────────────────────────────────────────┤
│  Optional:                                                      │
│    async_hard_refresh_codes()           # re-fetch from device  │
│    subscribe_push_updates()             # set up listeners      │
│    unsubscribe_push_updates()           # clean up listeners    │
│    async_release_managed_slot(slot)     # slot teardown hook    │
└─────────────────────────────────────────────────────────────────┘
                              │
                              │ implements
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      YourLockProvider                           │
└─────────────────────────────────────────────────────────────────┘

Slot-Only vs Native-User Providers

Providers split into two flavors based on the lock's user model:

• Slot-only — the lock indexes credentials by slot directly with no user concept (Z-Wave User Code CC, ZHA, Zigbee2MQTT, Virtual) OR the user is the credential's friendly name (Akuvox, Schlage). Set supports_native_users = False (the default). LCM addresses credentials by slot; user_id in async_set_credential is ignored.
• Native-user — the lock has a real user record that owns one or more credentials (Matter, Z-Wave User Credential CC). Set supports_native_users = True. LCM follows a user-first lifecycle: async_set_user then async_set_credential. Implement async_get_capabilities() so the seam can decide whether to write a user record before the credential.

The seam (_set_credential / _delete_credential in BaseLock) routes the orchestration automatically; you just implement the primitives.

Update Modes

Poll Mode (Default)

The coordinator periodically calls async_get_usercodes() at usercode_scan_interval.

Flow:

1. Timer fires at usercode_scan_interval
2. Coordinator calls async_internal_get_usercodes()
3. Base class applies rate limiting, then calls async_get_usercodes() (which by default projects async_get_users())
4. Provider returns current credential state
5. Coordinator updates data and notifies listening entities

When to use: integrations without push event support.

Example:

class MyLock(BaseLock):
    @property
    def usercode_scan_interval(self) -> timedelta:
        return timedelta(minutes=1)

    @property
    def hard_refresh_interval(self) -> timedelta | None:
        return None  # No separate drift check needed

Push Mode

For integrations that emit real-time events (Z-Wave value updates, Matter LockUserChange, ZHA programming events, Zigbee2MQTT MQTT messages).

Flow:

1. Provider subscribes to integration events in subscribe_push_updates()
2. An event fires
3. Provider's handler resolves the LCM slot (for native-user providers under the tag identity model, this means walking userIndex → user.name → lcm:<slot>: tag) and the SlotCredential value
4. Provider calls self.coordinator.push_update({slot: SlotCredential})
5. Coordinator merges and notifies entities immediately

When to use: any integration with push. Cuts polling overhead and gives the UI real-time feedback.

Example:

class MyLock(BaseLock):
    @property
    def supports_push(self) -> bool:
        return True

    @property
    def hard_refresh_interval(self) -> timedelta | None:
        return timedelta(hours=1)  # Still recommended for drift

Drift Detection

Drift detection re-reads the lock to catch changes the push path missed (keypad-enrolled codes, races, other controllers).

Flow:

1. Timer fires at hard_refresh_interval
2. Coordinator calls async_internal_hard_refresh_codes()
3. Provider queries the device directly, bypassing any cache (async_hard_refresh_codes())
4. Coordinator updates and notifies entities if anything changed

When to use:

• Codes can be changed outside Home Assistant
• The integration caches values that can go stale
• You want a backstop even with push mode

Interval Properties

Property	Default	Purpose
supports_push	False	Opt-in to push mode (disables polling for updates; polling for drift still applies)
supports_native_users	False	Opt-in to the user lifecycle path
usercode_scan_interval	1 minute	How often to poll for updates (ignored if supports_push=True)
hard_refresh_interval	None	How often to hard refresh for drift detection (None = disabled)
connection_check_interval	30 seconds	How often to check connection state (None = disabled, used when the integration emits config-entry state changes)

Implementing the Provider Primitives

async_get_users()

Return the lock's users with their credentials.

from custom_components.lock_code_manager.domain.credentials import (
    Credential,
    CredentialType,
    User,
    user_from_slot,
)
from custom_components.lock_code_manager.domain.models import SlotCredential


# Slot-only: synthesize one user per occupied slot.
async def async_get_users(self) -> list[User]:
    """Read current credential state from the device."""
    try:
        raw = await self._device.get_all_codes()
    except SomeDeviceError as err:
        raise LockDisconnected(f"get codes failed: {err}") from err

    return [
        user_from_slot(
            slot,
            SlotCredential.known(pin) if pin else SlotCredential.empty(),
        )
        for slot, pin in raw.items()
    ]


# Native-user: return the lock's actual users.
async def async_get_users(self) -> list[User]:
    """Read users + credentials from the lock."""
    raw_users = await self._driver.list_users()
    raw_creds = await self._driver.list_credentials()
    users_by_id = {
        u.user_id: User(user_id=u.user_id, name=u.name, active=u.active)
        for u in raw_users
    }
    for cred in raw_creds:
        owner = users_by_id.get(cred.user_id)
        if owner is None:
            continue
        owner.credentials.append(
            Credential(
                type=CredentialType.PIN,
                slot=cred.slot,
                state=(
                    SlotCredential.known(cred.value)
                    if cred.value
                    else SlotCredential.unreadable()
                ),
            )
        )
    return list(users_by_id.values())

async_set_credential()

Write a single credential. The base seam has already resolved user_id (for native-user providers, via the find-or-create-by-tag in async_set_user).

async def async_set_credential(
    self,
    user_id: int,
    credential: Credential,
    pin: str,
    *,
    name: str | None,
    source: Literal["sync", "direct"],
) -> bool:
    """Write a PIN to a slot; return whether the lock changed."""
    try:
        await self._driver.set_pin(
            user_id=user_id,
            slot=credential.slot,
            pin=pin,
        )
    except DuplicateError as err:
        raise DuplicateCodeError(
            code_slot=credential.slot,
            lock_entity_id=self.lock.entity_id,
        ) from err
    except DeviceError as err:
        # Transient: route to retry, not slot suspension.
        raise LockDisconnected(
            f"set credential slot {credential.slot} failed: {err}"
        ) from err
    return True

Return True if the value changed, False if the lock confirmed it was already set. When you can't tell (write-only locks like Matter), return True — the seam uses this to drive a coordinator refresh.

async_delete_credential()

async def async_delete_credential(self, ref: CredentialRef) -> bool:
    """Delete the credential addressed by ref."""
    try:
        await self._driver.clear_credential(
            user_id=ref.user_id,
            slot=ref.slot,
        )
    except DeviceError as err:
        raise LockDisconnected(
            f"delete credential slot {ref.slot} failed: {err}"
        ) from err
    return True

Note: the seam does not delete the owning user after async_delete_credential returns. The user is a persistent slot anchor (see Architecture for the lifecycle rationale). User deletion happens via async_release_managed_slot(slot).

async_set_user() (Native-User Only)

Find-or-create the lock user for the LCM slot encoded in user.name.

from custom_components.lock_code_manager.providers._util import parse_tag


async def async_set_user(self, user: User) -> SetUserResult:
    """Find-or-create the lock user for the LCM slot in user.name."""
    slot = self._slot_from_seam_user(user)
    existing_user_id = await self._find_user_index_for_slot(slot)
    try:
        result = await self._driver.set_user(
            user_id=existing_user_id,  # None on CREATE
            user_name=user.name,
            active=user.active,
        )
    except DeviceError as err:
        raise LockDisconnected(f"set user for slot {slot} failed: {err}") from err
    return SetUserResult(
        user_id=result["user_id"], created=existing_user_id is None
    )


def _slot_from_seam_user(self, user: User) -> int:
    """Recover LCM slot from the tagged user name; fall back to user_id."""
    if user.name:
        slot, _ = parse_tag(user.name)
        if slot is not None:
            return slot
    return user.user_id


async def _find_user_index_for_slot(self, slot: int) -> int | None:
    """Return the lock user_id LCM owns for slot, if any.

    1. Canonical pass: a user whose name parses to this LCM slot.
    2. Legacy adoption: an untagged user at the pre-PR-C invariant
       (user_id == slot for U3C; credential_index == slot for Matter)
       that also owns a PIN at the slot. Skip users already tagged for
       a different slot so we don't re-bind cross-slot.
    """
    users = await self.async_get_users()
    try:
        return next(
            u.user_id
            for u in users
            if u.name and parse_tag(u.name)[0] == slot
        )
    except StopIteration:
        return next(
            (
                u.user_id
                for u in users
                if u.user_id == slot
                and parse_tag(u.name or "")[0] is None
                for cred in u.pin_credentials
                if cred.slot == slot
            ),
            None,
        )

The legacy adoption pass is what makes pre-4.0.0 installs upgrade seamlessly without orphaning existing PINs.

async_delete_user() (Native-User Only)

async def async_delete_user(self, user_id: int) -> None:
    """Delete a lock user (cascades its credentials per spec)."""
    try:
        await self._driver.delete_user(user_id)
    except DeviceError as err:
        raise LockDisconnected(f"delete user {user_id} failed: {err}") from err

async_get_capabilities() (Native-User Only)

async def async_get_capabilities(self) -> LockCapabilities:
    """Report lock capabilities used by the seam to gate the user path."""
    caps = await self._driver.get_capabilities()
    return LockCapabilities(
        supports_user_management=caps["supports_user_management"],
        max_users=caps["max_users"],
        credential_types={
            CredentialType.PIN: CredentialTypeCapability(
                num_slots=caps["pin_num_slots"],
                min_length=caps["pin_min_length"],
                max_length=caps["pin_max_length"],
                supports_learn=False,
            )
        },
        max_user_name_length=caps.get("max_user_name_length", 0),
    )

The seam reads supports_user_management and max_user_name_length to decide whether the user-lifecycle path applies, and to size the tagged user name.

is_integration_connected()

def is_integration_connected(self) -> bool:
    """Return whether the integration is connected."""
    return self.lock_config_entry.state == ConfigEntryState.LOADED

Implementing Push Updates

1. Enable push mode

@property
def supports_push(self) -> bool:
    return True

2. Subscribe to events

@callback
def subscribe_push_updates(self) -> None:
    """Subscribe to real-time updates. Must be idempotent."""
    if self._event_unsub is not None:
        return

    @callback
    def on_code_changed(event_data) -> None:
        slot = event_data["slot"]
        state = (
            SlotCredential.known(event_data["pin"])
            if event_data.get("pin")
            else SlotCredential.empty()
        )
        if self.coordinator:
            self.coordinator.push_update({slot: state})

    self._event_unsub = self._device.subscribe(on_code_changed)

3. Tear down on unload

@callback
def unsubscribe_push_updates(self) -> None:
    """Unsubscribe. Must be idempotent."""
    if self._event_unsub:
        self._event_unsub()
        self._event_unsub = None

Push under the tag identity model

For native-user providers using lcm:<slot>: tags, the push handler must resolve the LCM slot via the owning user's tag, not via lock-side indices that may differ from LCM slots. See the Matter provider's _dispatch_lock_user_change for the canonical pattern: extract userIndex from the event, fetch the raw user list, parse its name's tag, and push to the resolved LCM slot.

Exception Routing

Providers must raise LockCodeManagerProviderError subclasses for lock-side failures. The seam routes each subclass differently:

Exception	When to raise	Seam behavior
LockDisconnected	Transient communication failure (lock asleep, integration not loaded, network glitch)	Retry with backoff
LockOperationFailed	Lock reachable but the operation didn't take (rejected input, response timeout)	Log + propagate
CodeRejectedError	Lock will not accept this PIN (any reason)	Disable slot, persistent notification
DuplicateCodeError	PIN duplicates another slot's value	Disable slot, persistent notification (subclass of CodeRejectedError)
ProviderNotImplementedError	A primitive is missing on a subclass (raised by BaseLock defaults)	Programmer error; do not raise from real provider code

from ..domain.exceptions import (
    CodeRejectedError,
    DuplicateCodeError,
    LockDisconnected,
    LockOperationFailed,
)


async def async_set_credential(self, user_id, credential, pin, *, name, source):
    try:
        await self._driver.set_pin(user_id, credential.slot, pin)
    except TransientError as err:
        raise LockDisconnected(f"set slot {credential.slot}: {err}") from err
    except DuplicatePinError as err:
        raise DuplicateCodeError(
            code_slot=credential.slot,
            lock_entity_id=self.lock.entity_id,
        ) from err
    except RejectedError as err:
        raise CodeRejectedError(
            code_slot=credential.slot,
            lock_entity_id=self.lock.entity_id,
            reason=str(err),
        ) from err
    return True

Do not raise bare Exception or HomeAssistantError from providers — those bypass the routing.

Rate Limiting

The base class provides automatic rate limiting through _execute_rate_limited(). The @final async_internal_* methods apply it:

• async_internal_get_usercodes() — rate-limited get
• async_set_usercode() — rate-limited set + refresh
• async_clear_usercode() — rate-limited clear + refresh
• async_internal_hard_refresh_codes() — rate-limited hard refresh

Default delay between operations is 2 seconds (MIN_OPERATION_DELAY).

For read-modify-write sequences (e.g., Akuvox's list-then-modify pattern), there is an outer _sequence_lock that providers can acquire to make the whole sequence atomic without deadlocking on the inner per-call lock.

Connection State Management

The base class handles connection state transitions:

1. Reconnection detection: when is_integration_connected() transitions from False to True:
   • Coordinator refresh is triggered
   • Push subscriptions are re-established (if supports_push=True)
2. Disconnection handling: when it transitions from True to False:
   • Push subscriptions are cleaned up
3. Config entry state changes: for integrations that expose config entry state (Z-Wave JS, Matter):
   • The base class listens for state changes
   • Automatically re-runs setup when the integration reloads

For these "rich-state" integrations, set connection_check_interval = None since the connection event drives the transitions.

Slot Teardown Hook

When a slot is removed from LCM config, the base seam calls async_release_managed_slot(slot) on each provider that manages that slot. The default implementation is a no-op. Native-user providers override to remove the persistent user anchor for the slot (and any remaining credentials cascade per spec).

async def async_release_managed_slot(self, slot: int) -> None:
    """Remove the lock user that anchors this LCM slot."""
    user_id = await self._find_user_index_for_slot(slot)
    if user_id is not None:
        await self.async_delete_user(user_id)

Wrap in try/except LockDisconnected — slot teardown can't block on lock connectivity.

Best Practices

1. Prefer push when the integration supports events — more responsive, less device traffic, less mesh load.
2. Enable drift detection when codes can change outside Home Assistant; pair it with push.
3. Cache appropriately — async_get_users() may return cached data; async_hard_refresh_codes() must bypass cache.
4. Use the typed exceptions — LockDisconnected for transient, CodeRejectedError for permanent rejections, LockOperationFailed for everything in between.
5. Make subscriptions idempotent — both subscribe and unsubscribe may be called multiple times.
6. Return True when uncertain — for write-only locks, returning True from set/delete primitives drives a coordinator refresh; returning False skips it.
7. Use the orchestration helpers — call the @final seam methods (async_set_usercode / async_clear_usercode), not the primitives directly. The seam owns rate limiting, connection checks, and coordinator routing.
8. Resolve LCM slots, not lock indices, in push handlers — for native-user providers under the tag identity model, the lock's user_id / credential_index may differ from the LCM slot. Always resolve via the owning user's lcm:<slot>: tag.