Add iOS Live Activity webhook handlers to mobile_app

homeassistant/components/mobile_app/__init__.py

@@ -41,6 +41,7 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DATA_PUSH_CHANNEL,
     DATA_STORE,
@@ -75,6 +76,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         DATA_CONFIG_ENTRIES: {},
         DATA_DELETED_IDS: app_config.get(DATA_DELETED_IDS, []),
         DATA_DEVICES: {},
+        DATA_LIVE_ACTIVITY_TOKENS: {},
         DATA_PUSH_CHANNEL: {},
         DATA_STORE: store,
         DATA_PENDING_UPDATES: {sensor_type: {} for sensor_type in SENSOR_TYPES},
@@ -225,6 +227,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     webhook_unregister(hass, webhook_id)
     del hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id]
     del hass.data[DOMAIN][DATA_DEVICES][webhook_id]
+    hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS].pop(webhook_id, None)
     await hass_notify.async_reload(hass, DOMAIN)
 
     return True

homeassistant/components/mobile_app/const.py

@@ -35,6 +35,26 @@
 ATTR_PUSH_WEBSOCKET_CHANNEL = "push_websocket_channel"
 ATTR_PUSH_TOKEN = "push_token"
 ATTR_PUSH_URL = "push_url"
+
+ATTR_SUPPORTS_LIVE_ACTIVITIES = "supports_live_activities"
+ATTR_SUPPORTS_LIVE_ACTIVITIES_FREQUENT_UPDATES = (
+    "supports_live_activities_frequent_updates"
+)
+ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN = "live_activity_push_to_start_token"
+ATTR_LIVE_ACTIVITY_PUSH_TO_START_APNS_ENVIRONMENT = (
+    "live_activity_push_to_start_apns_environment"
+)
+
+# Tag identifying a specific Live Activity instance — matches the `tag` field used by
+# the iOS companion app's ActivityKit integration.
+ATTR_LIVE_ACTIVITY_TAG = "tag"
+
+# In-memory store for per-device Live Activity push tokens, keyed by webhook_id → tag.
+# Populated by update_live_activity_token and cleared by live_activity_dismissed webhooks.
+DATA_LIVE_ACTIVITY_TOKENS = "live_activity_tokens"
+
+EVENT_LIVE_ACTIVITY_TOKEN_UPDATED = f"{DOMAIN}_live_activity_token_updated"
+EVENT_LIVE_ACTIVITY_DISMISSED = f"{DOMAIN}_live_activity_dismissed"
 ATTR_PUSH_RATE_LIMITS = "rateLimits"
 ATTR_PUSH_RATE_LIMITS_ERRORS = "errors"
 ATTR_PUSH_RATE_LIMITS_MAXIMUM = "maximum"
@@ -92,6 +112,20 @@
         # Set to True to indicate that this registration will connect via websocket channel
         # to receive push notifications.
         vol.Optional(ATTR_PUSH_WEBSOCKET_CHANNEL): cv.boolean,
+        # iOS Live Activities capability flags and push-to-start token (iOS 17.2+).
+        # push-to-start allows HA to remotely start a new Live Activity on the device
+        # without requiring one to already be running.
+        vol.Optional(ATTR_SUPPORTS_LIVE_ACTIVITIES): cv.boolean,
+        vol.Optional(ATTR_SUPPORTS_LIVE_ACTIVITIES_FREQUENT_UPDATES): cv.boolean,
+        # push-to-start token and environment must be provided together — a token
+        # without an environment is ambiguous (sandbox tokens fail on production).
+        vol.Inclusive(
+            ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN, "live_activity_push_to_start"
+        ): vol.All(cv.string, vol.Length(min=1)),
+        vol.Inclusive(
+            ATTR_LIVE_ACTIVITY_PUSH_TO_START_APNS_ENVIRONMENT,
+            "live_activity_push_to_start",
+        ): vol.In(["sandbox", "production"]),
     },
     extra=vol.ALLOW_EXTRA,
 )

homeassistant/components/mobile_app/notify.py

@@ -29,6 +29,8 @@
     ATTR_APP_ID,
     ATTR_APP_VERSION,
     ATTR_DEVICE_NAME,
+    ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN,
+    ATTR_LIVE_ACTIVITY_TAG,
     ATTR_OS_VERSION,
     ATTR_PUSH_RATE_LIMITS,
     ATTR_PUSH_RATE_LIMITS_ERRORS,
@@ -39,6 +41,7 @@
     ATTR_PUSH_URL,
     ATTR_WEBHOOK_ID,
     DATA_CONFIG_ENTRIES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_NOTIFY,
     DATA_PUSH_CHANNEL,
     DOMAIN,
@@ -145,6 +148,43 @@ async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
                 f"Device(s) with webhook id(s) {', '.join(failed_targets)} not connected to local push notifications"
             )
 
+    def _get_live_activity_token(
+        self, target: str, registration: dict[str, Any], data: dict[str, Any]
+    ) -> str | None:
+        """Return the Live Activity APNs token if this notification targets one.
+
+        Checks whether the notification payload contains live_activity: true
+        and a tag. If a per-activity APNs token is stored for that tag, it is
+        returned. Otherwise, if the device has a push-to-start token, that is
+        returned so the relay server can start a new activity remotely.
+
+        The token is sent alongside the FCM registration token to the same
+        relay URL. The relay server places it in the FCM message's
+        apns.liveActivityToken field, and FCM handles APNs delivery.
+
+        Returns None if this is a normal notification (not a Live Activity).
+        """
+        notification_data = data.get(ATTR_DATA) or {}
+        if notification_data.get("live_activity") is not True:
+            return None
+
+        tag = notification_data.get(ATTR_LIVE_ACTIVITY_TAG)
+        if not tag or not isinstance(tag, str):
+            return None
+
+        # Per-activity token — the activity is already running on the device.
+        live_activity_tokens = self.hass.data[DOMAIN].get(DATA_LIVE_ACTIVITY_TOKENS, {})
+        device_tokens = live_activity_tokens.get(target, {})
+        if tag in device_tokens:
+            return device_tokens[tag][ATTR_PUSH_TOKEN]
+
+        # Push-to-start token — start a new activity remotely (iOS 17.2+).
+        app_data = registration[ATTR_APP_DATA]
+        if ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN in app_data:
+            return app_data[ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN]
+
+        return None
+
     async def _async_send_remote_message_target(self, target, registration, data):
         """Send a message to a target."""
         app_data = registration[ATTR_APP_DATA]
@@ -154,6 +194,13 @@ async def _async_send_remote_message_target(self, target, registration, data):
         target_data = dict(data)
         target_data[ATTR_PUSH_TOKEN] = push_token
 
+        # If this is a Live Activity notification, include the APNs token so the
+        # relay server can set apns.liveActivityToken in the FCM payload. FCM then
+        # handles apns-push-type: liveactivity and APNs routing automatically.
+        live_activity_token = self._get_live_activity_token(target, registration, data)
+        if live_activity_token:
+            target_data["live_activity_token"] = live_activity_token
+
         reg_info = {
             ATTR_APP_ID: registration[ATTR_APP_ID],
             ATTR_APP_VERSION: registration[ATTR_APP_VERSION],

homeassistant/components/mobile_app/util.py

@@ -14,6 +14,7 @@
     ATTR_PUSH_TOKEN,
     ATTR_PUSH_URL,
     ATTR_PUSH_WEBSOCKET_CHANNEL,
+    ATTR_SUPPORTS_LIVE_ACTIVITIES,
     CONF_CLOUDHOOK_URL,
     DATA_CONFIG_ENTRIES,
     DATA_DEVICES,
@@ -48,6 +49,14 @@ def supports_push(hass: HomeAssistant, webhook_id: str) -> bool:
     ) or ATTR_PUSH_WEBSOCKET_CHANNEL in app_data
 
 
+@callback
+def supports_live_activities(hass: HomeAssistant, webhook_id: str) -> bool:
+    """Return if the device supports iOS Live Activities."""
+    config_entry = hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id]
+    app_data = config_entry.data.get(ATTR_APP_DATA, {})
+    return bool(app_data.get(ATTR_SUPPORTS_LIVE_ACTIVITIES))
+
+
 @callback
 def get_notify_service(hass: HomeAssistant, webhook_id: str) -> str | None:
     """Return the notify service for this webhook ID."""

homeassistant/components/mobile_app/webhook.py

@@ -66,10 +66,12 @@
     ATTR_DEVICE_NAME,
     ATTR_EVENT_DATA,
     ATTR_EVENT_TYPE,
+    ATTR_LIVE_ACTIVITY_TAG,
     ATTR_MANUFACTURER,
     ATTR_MODEL,
     ATTR_NO_LEGACY_ENCRYPTION,
     ATTR_OS_VERSION,
+    ATTR_PUSH_TOKEN,
     ATTR_SENSOR_ATTRIBUTES,
     ATTR_SENSOR_DEVICE_CLASS,
     ATTR_SENSOR_DISABLED,
@@ -90,6 +92,7 @@
     ATTR_WEBHOOK_DATA,
     ATTR_WEBHOOK_ENCRYPTED,
     ATTR_WEBHOOK_ENCRYPTED_DATA,
+    ATTR_WEBHOOK_ID,
     ATTR_WEBHOOK_TYPE,
     CONF_CLOUDHOOK_URL,
     CONF_REMOTE_UI_URL,
@@ -98,12 +101,15 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DOMAIN,
     ERR_ENCRYPTION_ALREADY_ENABLED,
     ERR_ENCRYPTION_REQUIRED,
     ERR_INVALID_FORMAT,
     ERR_SENSOR_NOT_REGISTERED,
+    EVENT_LIVE_ACTIVITY_DISMISSED,
+    EVENT_LIVE_ACTIVITY_TOKEN_UPDATED,
     SCHEMA_APP_DATA,
     SENSOR_TYPES,
     SIGNAL_LOCATION_UPDATE,
@@ -797,3 +803,84 @@ async def webhook_scan_tag(
         registration_context(config_entry.data),
     )
     return empty_okay_response()
+
+
+@WEBHOOK_COMMANDS.register("update_live_activity_token")
+@validate_schema(
+    {
+        vol.Required(ATTR_LIVE_ACTIVITY_TAG): vol.All(cv.string, vol.Length(min=1)),
+        vol.Required(ATTR_PUSH_TOKEN): vol.All(cv.string, vol.Length(min=1)),
+    }
+)
+async def webhook_update_live_activity_token(
+    hass: HomeAssistant, config_entry: ConfigEntry, data: dict[str, Any]
+) -> Response:
+    """Handle a Live Activity token update from the iOS companion app.
+
+    When the iOS app creates a Live Activity locally, ActivityKit provides
+    a per-activity APNs push token. The app sends this token so HA can
+    later include it as live_activity_token in the push relay request.
+    The relay server places it in the FCM message's apns.liveActivityToken
+    field, and FCM handles APNs delivery automatically.
+    """
+    webhook_id = config_entry.data[CONF_WEBHOOK_ID]
+    activity_tag = data[ATTR_LIVE_ACTIVITY_TAG]
+
+    live_activity_tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    live_activity_tokens.setdefault(webhook_id, {})[activity_tag] = {
+        ATTR_PUSH_TOKEN: data[ATTR_PUSH_TOKEN],
+    }
+
+    device: dr.DeviceEntry = hass.data[DOMAIN][DATA_DEVICES][webhook_id]
+    hass.bus.async_fire(
+        EVENT_LIVE_ACTIVITY_TOKEN_UPDATED,
+        {
+            ATTR_LIVE_ACTIVITY_TAG: activity_tag,
+            ATTR_DEVICE_ID: device.id,
+            ATTR_WEBHOOK_ID: webhook_id,
+        },
+        EventOrigin.remote,
+        context=registration_context(config_entry.data),
+    )
+
+    return empty_okay_response()
+
+
+@WEBHOOK_COMMANDS.register("live_activity_dismissed")
+@validate_schema(
+    {
+        vol.Required(ATTR_LIVE_ACTIVITY_TAG): vol.All(cv.string, vol.Length(min=1)),
+    }
+)
+async def webhook_live_activity_dismissed(
+    hass: HomeAssistant, config_entry: ConfigEntry, data: dict[str, str]
+) -> Response:
+    """Handle a Live Activity dismissal from the iOS companion app.
+
+    When a Live Activity ends on the device (user dismissal, expiration,
+    or an explicit end event), the app notifies HA so the stored push
+    token for that activity can be cleaned up.
+    """
+    webhook_id = config_entry.data[CONF_WEBHOOK_ID]
+    activity_tag = data[ATTR_LIVE_ACTIVITY_TAG]
+
+    live_activity_tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    if webhook_id in live_activity_tokens:
+        live_activity_tokens[webhook_id].pop(activity_tag, None)
+        # Clean up the device key if no activities remain.
+        if not live_activity_tokens[webhook_id]:
+            del live_activity_tokens[webhook_id]
+
+    device: dr.DeviceEntry = hass.data[DOMAIN][DATA_DEVICES][webhook_id]
+    hass.bus.async_fire(
+        EVENT_LIVE_ACTIVITY_DISMISSED,
+        {
+            ATTR_LIVE_ACTIVITY_TAG: activity_tag,
+            ATTR_DEVICE_ID: device.id,
+            ATTR_WEBHOOK_ID: webhook_id,
+        },
+        EventOrigin.remote,
+        context=registration_context(config_entry.data),
+    )
+
+    return empty_okay_response()

tests/components/mobile_app/test_init.py

@@ -1,9 +1,11 @@
 """Tests for the mobile app integration."""
 
 from collections.abc import Awaitable, Callable
+from http import HTTPStatus
 from typing import Any
 from unittest.mock import Mock, patch
 
+from aiohttp.test_utils import TestClient
 import pytest
 
 from homeassistant.components.cloud import CloudNotAvailable
@@ -12,6 +14,7 @@
     CONF_CLOUDHOOK_URL,
     CONF_USER_ID,
     DATA_DELETED_IDS,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DOMAIN,
 )
 from homeassistant.config_entries import ConfigEntry, ConfigEntryState
@@ -615,3 +618,33 @@ def mock_listen_cloudhook_change(hass_instance, wh_id: str, callback):
 
         # URL should remain the same
         assert config_entry.data[CONF_CLOUDHOOK_URL] == new_url
+
+
+@pytest.mark.usefixtures("create_registrations")
+async def test_unload_removes_live_activity_tokens(
+    hass: HomeAssistant, webhook_client: TestClient
+) -> None:
+    """Test that live activity tokens are removed from hass.data when entry is unloaded."""
+    # Use the cleartext (non-encrypted) entry
+    config_entry = hass.config_entries.async_entries("mobile_app")[1]
+    webhook_id = config_entry.data["webhook_id"]
+
+    # Store a live activity token via the webhook
+    resp = await webhook_client.post(
+        f"/api/webhook/{webhook_id}",
+        json={
+            "type": "update_live_activity_token",
+            "data": {
+                "tag": "washer_cycle",
+                "push_token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
+            },
+        },
+    )
+    assert resp.status == HTTPStatus.OK
+    assert webhook_id in hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+
+    # Unload the config entry
+    await hass.config_entries.async_unload(config_entry.entry_id)
+
+    # Verify the token is removed so stale tokens cannot be used after reloads/unloads
+    assert webhook_id not in hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]