diff --git a/homeassistant/components/sonos/__init__.py b/homeassistant/components/sonos/__init__.py
index 949b7acc440c..7594061a997f 100644
--- a/homeassistant/components/sonos/__init__.py
+++ b/homeassistant/components/sonos/__init__.py
@@ -8,6 +8,7 @@ from http import HTTPStatus
 from ipaddress import AddressValueError, IPv4Address
 import logging
 import socket
+import threading
 from typing import Any, cast
 from urllib.parse import urlparse
 
@@ -70,6 +71,7 @@ CONF_ADVERTISE_ADDR = "advertise_addr"
 CONF_INTERFACE_ADDR = "interface_addr"
 DISCOVERY_IGNORED_MODELS = ["Sonos Boost"]
 ZGS_SUBSCRIPTION_TIMEOUT = 2
+SHUTDOWN_TIMEOUT = 10
 
 CONFIG_SCHEMA = vol.Schema(
     {
@@ -180,10 +182,28 @@ class SonosDiscoveryManager:
         self.creation_lock = asyncio.Lock()
         self._known_invisible: set[SoCo] = set()
         self._manual_config_required = bool(hosts)
+        self._stop_event = threading.Event()
 
     async def async_shutdown(self) -> None:
         """Stop all running tasks."""
+        self._stop_event.set()
+        # Stop the event listener first so new topology events cannot schedule
+        # additional async_add_speakers runs while shutdown is waiting for
+        # creation_lock to drain existing work.
         await self._async_stop_event_listener()
+        # Wait for any in-flight _add_speakers executor job to finish before
+        # tearing down speakers and the event listener. Every async_add_speakers
+        # call holds creation_lock for its entire duration (including blocking
+        # network IO), so acquiring it here serializes cleanup after creation.
+        # Bound the wait so shutdown stays responsive under poor network conditions.
+        try:
+            async with asyncio.timeout(SHUTDOWN_TIMEOUT):
+                async with self.creation_lock:
+                    pass
+        except TimeoutError:
+            _LOGGER.warning(
+                "Timed out waiting for in-flight speaker discovery to complete"
+            )
         self._stop_manual_heartbeat()
 
     def is_device_invisible(self, ip_address: str) -> bool:
@@ -396,6 +416,13 @@ class SonosDiscoveryManager:
                 sub = None
                 if soco.uid == zgs_subscription_uid and zgs_subscription:
                     sub = zgs_subscription
+                if self._stop_event.is_set():
+                    # Entry was unloaded during IO; skip adding this speaker.
+                    _LOGGER.debug(
+                        "Config entry unloaded while adding speakers speaker %s, skipping",
+                        soco.uid,
+                    )
+                    return
                 self._add_speaker(soco, sub)
 
         async with self.creation_lock:
@@ -407,6 +434,12 @@ class SonosDiscoveryManager:
         """Create and set up a new SonosSpeaker instance."""
         try:
             speaker_info = soco.get_speaker_info(True, timeout=7)
+            if self._stop_event.is_set():
+                # Entry was unloaded during IO; skip adding this speaker.
+                _LOGGER.debug(
+                    "Config entry unloaded while adding speaker %s, skipping", soco.uid
+                )
+                return
             if soco.uid not in self.data.boot_counts:
                 self.data.boot_counts[soco.uid] = soco.boot_seqnum
             _LOGGER.debug("Adding new speaker: %s", speaker_info)
diff --git a/tests/components/sonos/test_init.py b/tests/components/sonos/test_init.py
index 307580000a14..dde22673d85a 100644
--- a/tests/components/sonos/test_init.py
+++ b/tests/components/sonos/test_init.py
@@ -33,7 +33,7 @@ from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 
-from .conftest import MockSoCo, SoCoMockFactory
+from .conftest import MockSoCo, SoCoMockFactory, SonosMockEvent
 
 from tests.common import MockConfigEntry, async_fire_time_changed
 
@@ -722,3 +722,44 @@ async def test_ipv6_not_supported(
         await hass.async_block_till_done()
     assert "invalid ip_address received" in caplog.text
     assert "2001:db8:3333:4444:5555:6666:7777:8888" in caplog.text
+
+
+async def test_no_exception_when_entry_unloaded_during_speaker_discovery(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    soco: MockSoCo,
+    soco_factory: SoCoMockFactory,
+    async_autosetup_sonos,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test no exception when a discovery event arrives after entry unload.
+
+    Exercises the scenario where async_add_speakers is called after the entry
+    has been unloaded. The _stop_event guard in _add_speaker must detect that
+    the entry is inactive and skip speaker creation without raising an exception.
+    """
+    new_soco = soco_factory.cache_mock(MockSoCo(), "10.10.10.99", "New Room")
+
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
+
+    # Fire a ZGS discovery event for the new speaker after the entry is
+    # unloaded. This simulates a late-arriving network event.
+    subscription = soco.zoneGroupTopology.subscribe.return_value
+    sub_callback = await subscription.wait_for_callback_to_be_set()
+    with (
+        patch.object(
+            MockSoCo,
+            "visible_zones",
+            new_callable=PropertyMock,
+            return_value={soco, new_soco},
+        ),
+        caplog.at_level(logging.DEBUG),
+    ):
+        caplog.clear()
+        sub_callback(SonosMockEvent(soco, soco.zoneGroupTopology, {}))
+        await hass.async_block_till_done(wait_background_tasks=True)
+
+    assert any(
+        "Config entry unloaded while adding speaker" in message
+        for message in caplog.messages
+    )