diff --git a/homeassistant/components/mobile_app/notify.py b/homeassistant/components/mobile_app/notify.py
index a52854bd992aaf..78e22fb677c135 100644
--- a/homeassistant/components/mobile_app/notify.py
+++ b/homeassistant/components/mobile_app/notify.py
@@ -98,20 +98,26 @@ async def async_send_message(self, message: str, title: str | None = None) -> No
         if title is not None:
             data[ATTR_TITLE] = title
 
+        webhook_id = self._config_entry.data[ATTR_WEBHOOK_ID]
+        push_channel: PushChannel | None = self.hass.data[DOMAIN][
+            DATA_PUSH_CHANNEL
+        ].get(webhook_id)
+        cloud_capable = ATTR_PUSH_URL in self._config_entry.data[ATTR_APP_DATA]
+
         # Sends notification via local push if available
-        # and fallback to cloud push if fails
-        if (webhook_id := self._config_entry.data[ATTR_WEBHOOK_ID]) in self.hass.data[
-            DOMAIN
-        ][DATA_PUSH_CHANNEL]:
-            push_channel: PushChannel = self.hass.data[DOMAIN][DATA_PUSH_CHANNEL][
-                webhook_id
-            ]
+        # and fallback to cloud push if fails; a degraded channel is
+        # bypassed for cloud-capable targets except a single probe send
+        if push_channel is not None and (
+            not cloud_capable or push_channel.async_should_send_local()
+        ):
             push_channel.async_send_notification(
                 data,
-                partial(_send_message, self._session, self._config_entry),
+                partial(_send_message, self._session, self._config_entry)
+                if cloud_capable
+                else None,
             )
         # Sends notification via cloud push notification service
-        elif ATTR_PUSH_URL in self._config_entry.data[ATTR_APP_DATA]:
+        elif cloud_capable:
             await _send_message(self._session, self._config_entry, data)
         else:
             raise HomeAssistantError(
@@ -217,16 +223,25 @@ async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
         for target in targets:
             entry: ConfigEntry = self.hass.data[DOMAIN][DATA_CONFIG_ENTRIES][target]
 
-            if target in local_push_channels:
-                local_push_channels[target].async_send_notification(
+            push_channel = local_push_channels.get(target)
+            cloud_capable = ATTR_PUSH_URL in entry.data[ATTR_APP_DATA]
+
+            # A degraded channel is bypassed for cloud-capable targets
+            # except a single probe send
+            if push_channel is not None and (
+                not cloud_capable or push_channel.async_should_send_local()
+            ):
+                push_channel.async_send_notification(
                     data,
-                    partial(self._async_send_remote_message_target, entry),
+                    partial(self._async_send_remote_message_target, entry)
+                    if cloud_capable
+                    else None,
                 )
                 async_dispatcher_send(self.hass, SIGNAL_RECORD_NOTIFICATION, target)
                 continue
 
             # Test if local push only.
-            if ATTR_PUSH_URL not in entry.data[ATTR_APP_DATA]:
+            if not cloud_capable:
                 failed_targets.append(target)
                 continue
 
diff --git a/homeassistant/components/mobile_app/push_notification.py b/homeassistant/components/mobile_app/push_notification.py
index e7b467f235d995..09ef54d58fad0e 100644
--- a/homeassistant/components/mobile_app/push_notification.py
+++ b/homeassistant/components/mobile_app/push_notification.py
@@ -2,13 +2,26 @@
 
 import asyncio
 from collections.abc import Callable
+from datetime import datetime
+import logging
 
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
 from homeassistant.helpers.event import async_call_later
 from homeassistant.util.uuid import random_uuid_hex
 
+_LOGGER = logging.getLogger(__name__)
+
 PUSH_CONFIRM_TIMEOUT = 10  # seconds
 
+# Consecutive confirm timeouts after which the channel is considered degraded
+# and cloud-capable targets are routed straight to cloud instead of waiting
+# out the confirm timeout for every message.
+PUSH_DEGRADED_AFTER_TIMEOUTS = 2
+
+# How long a degraded channel routes via cloud before the next send probes
+# local delivery again.
+PUSH_DEGRADED_PROBE_INTERVAL = 300  # seconds
+
 
 class PushChannel:
     """Class that represents a push channel."""
@@ -28,10 +41,31 @@ def __init__(
         self._send_message = send_message
         self.on_teardown = on_teardown
         self.pending_confirms: dict[str, dict] = {}
+        self._consecutive_timeouts = 0
+        self._degraded = False
+        self._probe_permit = False
+        self._unsub_degraded_probe: CALLBACK_TYPE | None = None
 
     @callback
-    def async_send_notification(self, data, fallback_send):
-        """Send a push notification."""
+    def async_should_send_local(self) -> bool:
+        """Return if the next send should be delivered locally.
+
+        Consumes the single probe permit of a degraded channel.
+        """
+        if not self._degraded:
+            return True
+        if self._probe_permit:
+            self._probe_permit = False
+            return True
+        return False
+
+    @callback
+    def async_send_notification(self, data, fallback_send: Callable | None):
+        """Send a push notification.
+
+        fallback_send is None for local-push-only registrations, which have no
+        cloud delivery to fall back to.
+        """
         if not self.support_confirm:
             self._send_message(data)
             return
@@ -40,16 +74,19 @@ def async_send_notification(self, data, fallback_send):
         data["hass_confirm_id"] = confirm_id
 
         async def handle_push_failed(_=None):
-            """Handle a failed local push notification."""
-            # Remove this handler from the pending dict
-            # If it didn't exist we hit a race condition between call_later and another
-            # push failing and tearing down the connection.
+            """Fall back to cloud for a local push left unconfirmed in time."""
+            # Already popped by a confirm or a teardown flush; nothing to fall back.
             if self.pending_confirms.pop(confirm_id, None) is None:
                 return
 
-            # Drop local channel if it's still open
+            # A teardown flush is not a delivery failure of a live channel
             if self.on_teardown is not None:
-                await self.async_teardown()
+                self._consecutive_timeouts += 1
+                if self._consecutive_timeouts >= PUSH_DEGRADED_AFTER_TIMEOUTS:
+                    self._async_mark_degraded()
+
+            if fallback_send is None:
+                return
 
             await fallback_send(data)
 
@@ -71,8 +108,48 @@ def async_confirm_notification(self, confirm_id) -> bool:
             return False
 
         self.pending_confirms.pop(confirm_id)["unsub_scheduled_push_failed"]()
+        # A timely confirm proves the channel delivers
+        if self._degraded:
+            _LOGGER.debug("Push channel %s restored to local delivery", self.webhook_id)
+        self._consecutive_timeouts = 0
+        self._async_clear_degraded()
         return True
 
+    @callback
+    def _async_mark_degraded(self) -> None:
+        """Route cloud-capable sends via cloud until a probe is confirmed."""
+        if not self._degraded:
+            _LOGGER.debug(
+                "Push channel %s degraded after %d consecutive confirm timeouts;"
+                " routing cloud-capable sends via cloud",
+                self.webhook_id,
+                self._consecutive_timeouts,
+            )
+        self._degraded = True
+        self._probe_permit = False
+        if self._unsub_degraded_probe is None:
+            self._unsub_degraded_probe = async_call_later(
+                self.hass, PUSH_DEGRADED_PROBE_INTERVAL, self._async_allow_probe
+            )
+
+    @callback
+    def _async_allow_probe(self, _now: datetime) -> None:
+        """Let a single next send probe local delivery again."""
+        _LOGGER.debug(
+            "Allowing a local delivery probe for push channel %s", self.webhook_id
+        )
+        self._unsub_degraded_probe = None
+        self._probe_permit = True
+
+    @callback
+    def _async_clear_degraded(self) -> None:
+        """Restore local delivery for all sends."""
+        if self._unsub_degraded_probe is not None:
+            self._unsub_degraded_probe()
+            self._unsub_degraded_probe = None
+        self._probe_permit = False
+        self._degraded = False
+
     async def async_teardown(self):
         """Tear down this channel."""
         # Tear down is in progress
@@ -81,6 +158,7 @@ async def async_teardown(self):
 
         self.on_teardown()
         self.on_teardown = None
+        self._async_clear_degraded()
 
         cancel_pending_local_tasks = [
             actions["handle_push_failed"]()
diff --git a/tests/components/mobile_app/test_notify.py b/tests/components/mobile_app/test_notify.py
index 301a6c2e6af3b9..5c91e7e05087ed 100644
--- a/tests/components/mobile_app/test_notify.py
+++ b/tests/components/mobile_app/test_notify.py
@@ -11,6 +11,9 @@
 from syrupy.assertion import SnapshotAssertion
 
 from homeassistant.components.mobile_app.const import DATA_LIVE_ACTIVITY_TOKENS, DOMAIN
+from homeassistant.components.mobile_app.push_notification import (
+    PUSH_DEGRADED_PROBE_INTERVAL,
+)
 from homeassistant.components.notify import (
     ATTR_MESSAGE,
     ATTR_TITLE,
@@ -25,10 +28,19 @@
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 
-from tests.common import MockConfigEntry, MockUser, snapshot_platform
+from tests.common import (
+    MockConfigEntry,
+    MockUser,
+    async_fire_time_changed,
+    snapshot_platform,
+)
 from tests.test_util.aiohttp import AiohttpClientMocker
 from tests.typing import WebSocketGenerator
 
+CONFIRM_TIMEOUT_PATCH = (
+    "homeassistant.components.mobile_app.push_notification.PUSH_CONFIRM_TIMEOUT"
+)
+
 
 @pytest.fixture
 async def setup_push_receiver(
@@ -412,6 +424,12 @@ async def test_notify_ws_confirming_works(
     result = await client.receive_json()
     assert result["success"]
 
+    # The timely confirm cancelled the fallback timer: advancing past the
+    # confirm timeout must not send anything via cloud
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=15))
+    await hass.async_block_till_done()
+    assert len(aioclient_mock.mock_calls) == 0
+
     # Drop local push channel and try to confirm another message
     await client.send_json_auto_id(
         {
@@ -444,7 +462,7 @@ async def test_notify_ws_not_confirming(
     setup_push_receiver,
     hass_ws_client: WebSocketGenerator,
 ) -> None:
-    """Test we go via cloud when failed to confirm."""
+    """Test a late confirm falls back via cloud without dropping the channel."""
     client = await hass_ws_client(hass)
 
     await client.send_json_auto_id(
@@ -457,31 +475,478 @@ async def test_notify_ws_not_confirming(
 
     sub_result = await client.receive_json()
     assert sub_result["success"]
+    sub_id = sub_result["id"]
 
     await hass.services.async_call(
         "notify", "mobile_app_test", {"message": "Hello world 1"}, blocking=True
     )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
 
-    with patch(
-        "homeassistant.components.mobile_app.push_notification.PUSH_CONFIRM_TIMEOUT", 0
-    ):
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
         await hass.services.async_call(
             "notify", "mobile_app_test", {"message": "Hello world 2"}, blocking=True
         )
         await hass.async_block_till_done()
         await hass.async_block_till_done()
 
-    # When we fail, all unconfirmed ones and failed one are sent via cloud
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+
+    # Only the message that missed its confirmation falls back via cloud;
+    # the channel stays registered instead of being torn down
+    assert len(aioclient_mock.mock_calls) == 1
+
+    # Later messages keep being delivered locally
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 3"}, blocking=True
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 3"
+    assert len(aioclient_mock.mock_calls) == 1
+
+    # Dropping the channel still flushes the unconfirmed messages via cloud once
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
+
+    assert len(aioclient_mock.mock_calls) == 3
+
+
+@pytest.mark.usefixtures("setup_push_receiver")
+async def test_notify_ws_confirm_resets_timeout_count(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test a timely confirm between two timeouts keeps the channel local."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_channel",
+            "webhook_id": "mock-webhook_id",
+            "support_confirm": True,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    sub_id = sub_result["id"]
+
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            "notify", "mobile_app_test", {"message": "Hello world 1"}, blocking=True
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
+    assert len(aioclient_mock.mock_calls) == 1
+
+    # A timely confirm resets the consecutive timeout count
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 2"}, blocking=True
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_confirm",
+            "webhook_id": "mock-webhook_id",
+            "confirm_id": msg_result["event"]["hass_confirm_id"],
+        }
+    )
+    result = await client.receive_json()
+    assert result["success"]
+
+    # The next timeout is an isolated one again and must not open the breaker
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            "notify", "mobile_app_test", {"message": "Hello world 3"}, blocking=True
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 3"
     assert len(aioclient_mock.mock_calls) == 2
 
-    # All future ones also go via cloud
+    # Local delivery continues
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 4"}, blocking=True
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 4"
+    assert len(aioclient_mock.mock_calls) == 2
+
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
+
+    assert len(aioclient_mock.mock_calls) == 3
+
+
+@pytest.mark.usefixtures("setup_push_receiver")
+async def test_notify_ws_degraded_channel(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test consecutive confirm timeouts degrade the channel to cloud routing."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_channel",
+            "webhook_id": "mock-webhook_id",
+            "support_confirm": True,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    sub_id = sub_result["id"]
+
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            "notify", "mobile_app_test", {"message": "Hello world 1"}, blocking=True
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        await hass.services.async_call(
+            "notify", "mobile_app_test", {"message": "Hello world 2"}, blocking=True
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+
+    # Both messages were delivered locally and fell back via cloud unconfirmed
+    assert len(aioclient_mock.mock_calls) == 2
+
+    # Two consecutive timeouts degraded the channel: this send goes straight
+    # via cloud without waiting out the confirm timeout
     await hass.services.async_call(
         "notify", "mobile_app_test", {"message": "Hello world 3"}, blocking=True
     )
+    assert len(aioclient_mock.mock_calls) == 3
+
+    # After the probe interval the next send tries local delivery again
+    async_fire_time_changed(
+        hass, dt_util.utcnow() + timedelta(seconds=PUSH_DEGRADED_PROBE_INTERVAL + 1)
+    )
+    await hass.async_block_till_done()
+
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            "notify", "mobile_app_test", {"message": "Hello world 4"}, blocking=True
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    # The probe was delivered locally and fell back via cloud unconfirmed;
+    # the bypassed message never reached the websocket
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 4"
+    assert len(aioclient_mock.mock_calls) == 4
+
+    # The unconfirmed probe kept the channel degraded
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 5"}, blocking=True
+    )
+    assert len(aioclient_mock.mock_calls) == 5
+
+    # Allow another probe and confirm it in time
+    async_fire_time_changed(
+        hass, dt_util.utcnow() + timedelta(seconds=PUSH_DEGRADED_PROBE_INTERVAL + 1)
+    )
+    await hass.async_block_till_done()
+
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 6"}, blocking=True
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 6"
+
+    # Only a single send probes: a message sent while the probe is still
+    # unconfirmed keeps routing via cloud
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 6b"}, blocking=True
+    )
+    assert len(aioclient_mock.mock_calls) == 6
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_confirm",
+            "webhook_id": "mock-webhook_id",
+            "confirm_id": msg_result["event"]["hass_confirm_id"],
+        }
+    )
+    result = await client.receive_json()
+    assert result["success"]
+
+    # The confirmed probe restored local delivery
+    await hass.services.async_call(
+        "notify", "mobile_app_test", {"message": "Hello world 7"}, blocking=True
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 7"
+    assert len(aioclient_mock.mock_calls) == 6
+
+    # Dropping the channel still flushes the unconfirmed message via cloud
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
+
+    assert len(aioclient_mock.mock_calls) == 7
+
+
+@pytest.mark.usefixtures("setup_push_receiver")
+async def test_send_message_degraded_channel(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test notify.send_message bypasses a degraded channel via cloud."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_channel",
+            "webhook_id": "mock-webhook_id",
+            "support_confirm": True,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    sub_id = sub_result["id"]
+
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            NOTIFY_DOMAIN,
+            SERVICE_SEND_MESSAGE,
+            {ATTR_ENTITY_ID: "notify.test", ATTR_MESSAGE: "Hello world 1"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        await hass.services.async_call(
+            NOTIFY_DOMAIN,
+            SERVICE_SEND_MESSAGE,
+            {ATTR_ENTITY_ID: "notify.test", ATTR_MESSAGE: "Hello world 2"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+    assert len(aioclient_mock.mock_calls) == 2
+
+    # The degraded channel is bypassed straight via cloud
+    await hass.services.async_call(
+        NOTIFY_DOMAIN,
+        SERVICE_SEND_MESSAGE,
+        {ATTR_ENTITY_ID: "notify.test", ATTR_MESSAGE: "Hello world 3"},
+        blocking=True,
+    )
+    assert len(aioclient_mock.mock_calls) == 3
+
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
 
     assert len(aioclient_mock.mock_calls) == 3
 
 
+@pytest.mark.usefixtures("setup_websocket_channel_only_push")
+async def test_local_push_only_stays_local_when_degraded(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test a degraded channel keeps local delivery for local-push-only targets."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_channel",
+            "webhook_id": "websocket-push-webhook-id",
+            "support_confirm": True,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    sub_id = sub_result["id"]
+
+    with (
+        patch(CONFIRM_TIMEOUT_PATCH, 0),
+        patch(
+            "homeassistant.components.mobile_app.notify._send_message"
+        ) as mock_cloud_send,
+    ):
+        await hass.services.async_call(
+            "notify",
+            "mobile_app_websocket_push_name",
+            {"message": "Hello world 1"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        await hass.services.async_call(
+            "notify",
+            "mobile_app_websocket_push_name",
+            {"message": "Hello world 2"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+    # A local-push-only registration has no cloud path to fall back to
+    assert len(mock_cloud_send.mock_calls) == 0
+
+    # The channel reached the degraded threshold, but a local-push-only
+    # target has no cloud path: the legacy service still delivers locally
+    await hass.services.async_call(
+        "notify",
+        "mobile_app_websocket_push_name",
+        {"message": "Hello world 3"},
+        blocking=True,
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 3"
+    confirm_id_3 = msg_result["event"]["hass_confirm_id"]
+
+    # The notify entity still delivers locally on the degraded channel as well
+    await hass.services.async_call(
+        NOTIFY_DOMAIN,
+        SERVICE_SEND_MESSAGE,
+        {ATTR_ENTITY_ID: "notify.websocket_push_name", ATTR_MESSAGE: "Hello world 4"},
+        blocking=True,
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 4"
+    confirm_id_4 = msg_result["event"]["hass_confirm_id"]
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_confirm",
+            "webhook_id": "websocket-push-webhook-id",
+            "confirm_id": confirm_id_3,
+        }
+    )
+    result = await client.receive_json()
+    assert result["success"]
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_confirm",
+            "webhook_id": "websocket-push-webhook-id",
+            "confirm_id": confirm_id_4,
+        }
+    )
+    result = await client.receive_json()
+    assert result["success"]
+
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
+
+
+@pytest.mark.usefixtures("setup_websocket_channel_only_push")
+async def test_local_push_only_missed_confirm(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test a missed confirm without a cloud fallback keeps the channel working."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "mobile_app/push_notification_channel",
+            "webhook_id": "websocket-push-webhook-id",
+            "support_confirm": True,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    sub_id = sub_result["id"]
+
+    # The confirm times out and there is no cloud delivery to fall back to
+    with patch(CONFIRM_TIMEOUT_PATCH, 0):
+        await hass.services.async_call(
+            "notify",
+            "mobile_app_websocket_push_name",
+            {"message": "Hello world 1"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 1"
+
+    # The channel is still registered and keeps delivering locally
+    await hass.services.async_call(
+        "notify",
+        "mobile_app_websocket_push_name",
+        {"message": "Hello world 2"},
+        blocking=True,
+    )
+    msg_result = await client.receive_json()
+    assert msg_result["event"]["message"] == "Hello world 2"
+
+    # The pending confirm of the second message is flushed by the teardown
+    await client.send_json_auto_id(
+        {
+            "type": "unsubscribe_events",
+            "subscription": sub_id,
+        }
+    )
+    sub_result = await client.receive_json()
+    assert sub_result["success"]
+    await hass.async_block_till_done()
+
+
 @pytest.mark.freeze_time("1970-01-01T00:00:00.000Z")
 async def test_local_push_only(
     hass: HomeAssistant,