diff --git a/homeassistant/components/vistapool/coordinator.py b/homeassistant/components/vistapool/coordinator.py
index 5765fa555326..ad24ebf636bf 100644
--- a/homeassistant/components/vistapool/coordinator.py
+++ b/homeassistant/components/vistapool/coordinator.py
@@ -10,7 +10,7 @@ from aioaquarite import (
     ResilientPoolSubscription,
 )
 
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import DOMAIN
@@ -41,6 +41,7 @@ class VistapoolDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self.pool_id: str = pool_id
         self.pool_name: str = pool_name
         self.subscription: ResilientPoolSubscription | None = None
+        self._push_connected = True
 
         super().__init__(
             hass,
@@ -61,17 +62,49 @@ class VistapoolDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 translation_key="update_failed",
             ) from err
 
+    @property
+    def push_connected(self) -> bool:
+        """Whether pool data is still flowing in from the subscription."""
+        return self._push_connected
+
     async def subscribe(self) -> None:
         """Subscribe to Firestore real-time updates via the library."""
 
         def _on_data(data: dict[str, Any]) -> None:
             """Callback from the Firestore thread; push data to the HA loop."""
-            self.hass.loop.call_soon_threadsafe(self.async_set_updated_data, data)
+            self.hass.loop.call_soon_threadsafe(self._async_handle_push, data)
 
         self.subscription = await self.api.subscribe_pool_resilient(
-            self.pool_id, _on_data
+            self.pool_id, _on_data, on_health=self._async_on_subscription_health
         )
 
+    @callback
+    def _async_handle_push(self, data: dict[str, Any]) -> None:
+        """Apply a snapshot; its arrival is what proves the connection is up."""
+        if not self._push_connected:
+            self._push_connected = True
+            _LOGGER.info("Reconnected to %s, entities are available again", self.name)
+        self.async_set_updated_data(data)
+
+    @callback
+    def _async_on_subscription_health(self, healthy: bool) -> None:
+        """Mark entities unavailable while the push connection is down.
+
+        Tracked separately from last_update_success: an optimistic update
+        or a manual refresh sets that flag back to True while the
+        subscription is still down, and the health callback only fires on
+        transitions, so it would not correct it. Only an incoming snapshot
+        clears this.
+        """
+        if healthy or not self._push_connected:
+            return
+        self._push_connected = False
+        _LOGGER.warning(
+            "Lost the connection to %s, entities are unavailable until it recovers",
+            self.name,
+        )
+        self.async_update_listeners()
+
     @override
     async def async_shutdown(self) -> None:
         """Cleanly close the resilient subscription."""
diff --git a/homeassistant/components/vistapool/entity.py b/homeassistant/components/vistapool/entity.py
index 175019c901da..9b47f6e524d7 100644
--- a/homeassistant/components/vistapool/entity.py
+++ b/homeassistant/components/vistapool/entity.py
@@ -1,5 +1,7 @@
 """Shared base entity helpers for Vistapool."""
 
+from typing import override
+
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
@@ -24,6 +26,12 @@ class VistapoolEntity(CoordinatorEntity[VistapoolDataUpdateCoordinator]):
             sw_version=str(sw_version) if sw_version is not None else None,
         )
 
+    @property
+    @override
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return super().available and self.coordinator.push_connected
+
     @property
     def pool_id(self) -> str:
         """Return the pool ID for the entity."""
diff --git a/homeassistant/components/vistapool/quality_scale.yaml b/homeassistant/components/vistapool/quality_scale.yaml
index 575914c35349..96224b9b8f9d 100644
--- a/homeassistant/components/vistapool/quality_scale.yaml
+++ b/homeassistant/components/vistapool/quality_scale.yaml
@@ -39,7 +39,7 @@ rules:
   docs-troubleshooting: done
   entity-category: done
   entity-disabled-by-default: done
-  entity-unavailable: todo
+  entity-unavailable: done
   integration-owner: done
   log-when-unavailable: done
   parallel-updates: done
diff --git a/tests/components/vistapool/test_init.py b/tests/components/vistapool/test_init.py
index 32a6753337f2..31bd531a833d 100644
--- a/tests/components/vistapool/test_init.py
+++ b/tests/components/vistapool/test_init.py
@@ -6,10 +6,13 @@ from unittest.mock import AsyncMock, MagicMock
 
 from aioaquarite import AquariteError, AuthenticationError
 
+from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
 from homeassistant.components.vistapool.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_ON, STATE_UNAVAILABLE
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import device_registry as dr
+from homeassistant.setup import async_setup_component
 
 from .conftest import MOCK_POOL_ID, MOCK_POOL_NAME
 
@@ -18,6 +21,8 @@ from tests.common import MockConfigEntry
 _SECOND_POOL_ID = "ZYXWVU9876543210"
 _SECOND_POOL_NAME = "Spa"
 _THIRD_POOL_ID = "QQQQQQ1111111111"
+_TEMPERATURE_ENTITY = "sensor.my_pool_temperature"
+_LIGHT_ENTITY = "light.my_pool_light"
 
 
 async def test_setup_entry(
@@ -312,6 +317,82 @@ async def test_apply_optimistic_creates_missing_intermediate_dicts(
     assert coordinator.data["existing"] == {"nested": {"key": 1}}
 
 
+async def test_entities_unavailable_while_push_connection_is_down(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_vistapool_client: AsyncMock,
+) -> None:
+    """Test entities go unavailable when the Firestore subscription drops.
+
+    The integration has no polling interval, so without this the last
+    snapshot would stay on display as if it were still current.
+    """
+    mock_config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(_TEMPERATURE_ENTITY).state != STATE_UNAVAILABLE
+
+    call = mock_vistapool_client.subscribe_pool_resilient.call_args
+    on_data = call.args[1]
+    on_health = call.kwargs["on_health"]
+
+    on_health(False)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(_TEMPERATURE_ENTITY).state == STATE_UNAVAILABLE
+
+    # Only an incoming snapshot proves the connection is back.
+    on_data({"main": {"temperature": 25}})
+    await hass.async_block_till_done()
+
+    assert hass.states.get(_TEMPERATURE_ENTITY).state == "25.0"
+
+
+async def test_entities_stay_unavailable_on_local_updates_during_outage(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_vistapool_client: AsyncMock,
+) -> None:
+    """Test updates that are not push snapshots do not fake availability.
+
+    Both an optimistic write and a manual refresh set the coordinator's
+    success flag, so availability cannot ride on that flag alone.
+    """
+    mock_vistapool_client.fetch_pool_data.return_value = {"light": {"status": 0}}
+    mock_config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert await async_setup_component(hass, "homeassistant", {})
+
+    on_health = mock_vistapool_client.subscribe_pool_resilient.call_args.kwargs[
+        "on_health"
+    ]
+    on_health(False)
+    await hass.async_block_till_done()
+    assert hass.states.get(_LIGHT_ENTITY).state == STATE_UNAVAILABLE
+
+    # An optimistic write updates coordinator data while the push is down.
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        SERVICE_TURN_ON,
+        {ATTR_ENTITY_ID: _LIGHT_ENTITY},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(_LIGHT_ENTITY).state == STATE_UNAVAILABLE
+
+    # So does a successful manual refresh.
+    await hass.services.async_call(
+        "homeassistant",
+        "update_entity",
+        {ATTR_ENTITY_ID: _LIGHT_ENTITY},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(_LIGHT_ENTITY).state == STATE_UNAVAILABLE
+
+
 async def test_unload_entry(
     hass: HomeAssistant,
     mock_config_entry: MockConfigEntry,