diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/CHANGELOG.md b/sdk/appconfiguration/azure-appconfiguration-provider/CHANGELOG.md
index 54481723b4d3..8451398b3b8b 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/CHANGELOG.md
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/CHANGELOG.md
@@ -8,6 +8,12 @@
 
 ### Bugs Fixed
 
+- Fixed a resource leak where replica clients that were no longer part of the auto-failover set were not closed during client refresh.
+- Fixed auto-failover replica discovery so that a DNS SRV lookup timeout (for either the origin or replica records) is distinguished from an empty replica list. A timeout now correctly triggers the longer fallback refresh interval, while an empty result refreshes at the normal interval.
+- Fixed a thread-safety issue by publishing refreshed secret values through a new configuration mapping instead of mutating the existing mapping while readers may be iterating.
+- Fixed `refresh_on` handling so that a single-string watched setting is treated as a key with the default (no) label instead of being incorrectly unpacked character-by-character.
+- Fixed a `KeyError` when loading with an endpoint and credential (no connection string).
+
 ### Other Changes
 
 - Bumped minimum dependency on `azure-core` to `>=1.31.0`.
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py
index 330e9981631a..a481238c616f 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py
@@ -78,7 +78,7 @@ def __init__(self, **kwargs: Any) -> None:
         self._startup_timeout: int = kwargs.pop("startup_timeout", DEFAULT_STARTUP_TIMEOUT)
 
         self._replica_client_manager = ConfigurationClientManager(
-            connection_string=kwargs.pop("connection_string"),
+            connection_string=kwargs.pop("connection_string", None),
             endpoint=kwargs.pop("endpoint"),
             credential=kwargs.pop("credential", None),
             user_agent=user_agent,
@@ -207,7 +207,12 @@ def refresh(self, **kwargs) -> None:
                 self._secret_provider.secret_refresh_timer
                 and self._secret_provider.secret_refresh_timer.needs_refresh()
             ):
-                self._dict.update(self._secret_provider.refresh_secrets())
+                # Publish a new dict rather than mutating in place. Readers may hold a live view or
+                # iterator over self._dict, and an in-place update could raise "dictionary changed
+                # size during iteration". Every write to self._dict must rebind to a new object.
+                updated_settings = dict(self._dict)
+                updated_settings.update(self._secret_provider.refresh_secrets())
+                self._dict = updated_settings
             self._replica_client_manager.refresh_clients()
             self._replica_client_manager.find_active_clients()
             replica_count = self._replica_client_manager.get_client_count() - 1
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py
index 4d9ee90978a6..a2ced46b3d1b 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py
@@ -70,11 +70,13 @@ def is_json_content_type(content_type: str) -> bool:
 
 
 def _build_watched_setting(setting: Union[str, Tuple[str, str]]) -> Tuple[str, str]:
-    try:
-        key, label = setting  # type:ignore
-    except (IndexError, ValueError):
-        key = str(setting)  # Ensure key is a string
-        label = NULL_CHAR
+    if isinstance(setting, str):
+        key, label = setting, NULL_CHAR
+    else:
+        try:
+            key, label = setting
+        except (TypeError, ValueError):
+            key, label = str(setting), NULL_CHAR
     if "*" in key or "*" in label:
         raise ValueError("Wildcard key or label filters are not supported for refresh.")
     return key, label
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py
index a5d0033647a8..2c7de888427b 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py
@@ -132,7 +132,7 @@ def _check_configuration_setting(
                     self.LOGGER.debug("Refresh all triggered by key: %s label %s.", key, label)
                     return True, None
             else:
-                raise e
+                raise
         return False, None
 
     @distributed_trace
@@ -343,7 +343,7 @@ def _validate_snapshot(self, snapshot_name: str) -> bool:
             if e.status_code == 404:
                 self.LOGGER.warning("Snapshot '%s' not found when resolving snapshot.", snapshot_name)
                 return False
-            raise e
+            raise
         if snapshot.composition_type != SnapshotComposition.KEY:
             raise ValueError(f"Composition type for '{snapshot_name}' must be 'key'.")
         return True
@@ -484,18 +484,13 @@ def refresh_clients(self):
         if self._next_update_time and self._next_update_time > time.time():
             return
 
-        failover_endpoints = find_auto_failover_endpoints(self._original_endpoint, self._replica_discovery_enabled)
-
-        if failover_endpoints is None:
-            # SRV record not found, so we should refresh after a longer interval
+        try:
+            failover_endpoints = find_auto_failover_endpoints(self._original_endpoint, self._replica_discovery_enabled)
+        except TimeoutError:
+            # SRV record resolution timed out, so we should refresh after a longer interval
             self._next_update_time = time.time() + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
             return
 
-        if len(failover_endpoints) == 0:
-            # No failover endpoints in SRV record.
-            self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
-            return
-
         discovered_clients = []
         for failover_endpoint in failover_endpoints:
             found_client = False
@@ -531,6 +526,12 @@ def refresh_clients(self):
                         )
                     )
         self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
+        # Close any replica clients that are no longer part of the failover.
+        retained_endpoints = {self._original_client.endpoint}
+        retained_endpoints.update(client.endpoint for client in discovered_clients)
+        for client in self._replica_clients:
+            if client.endpoint not in retained_endpoints:
+                client.close()
         if not self._load_balancing_enabled:
             random.shuffle(discovered_clients)
             self._replica_clients = [self._original_client] + discovered_clients
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_discovery.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_discovery.py
index 6dbfe722a964..ff111645ec39 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_discovery.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_discovery.py
@@ -40,8 +40,8 @@ def find_auto_failover_endpoints(endpoint: str, replica_discovery_enabled: bool)
 
     replicas = _find_replicas(origin.target)
 
-    if not replicas:
-        return None  # Timeout
+    if replicas is None:
+        raise TimeoutError("Timed out while resolving auto-failover replica endpoints.")
 
     srv_records = [origin] + replicas
     endpoints = []
@@ -57,6 +57,8 @@ def _find_origin(endpoint):
     uri = urlparse(endpoint).hostname
     request = f"_origin._tcp.{uri}"
     srv_records = _request_record(request)
+    if srv_records is None:
+        raise TimeoutError("Timed out while resolving auto-failover origin endpoint.")
     if not srv_records:
         return None
     return SRVRecord(srv_records[0])
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_key_vault/_secret_provider_base.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_key_vault/_secret_provider_base.py
index 11832cd663ce..a817621cde56 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_key_vault/_secret_provider_base.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_key_vault/_secret_provider_base.py
@@ -26,7 +26,7 @@ def __init__(self, **kwargs: Any) -> None:
         )
 
         if kwargs.get("secret_refresh_interval", 60) < 1:
-            raise ValueError("Secret refresh interval must be greater than 1 second.")
+            raise ValueError("Secret refresh interval must be at least 1 second.")
 
         self.secret_refresh_timer: Optional[_RefreshTimer] = (
             _RefreshTimer(refresh_interval=kwargs.pop("secret_refresh_interval", 60))
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py
index 6afdb0fed5d1..65c681b632a6 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py
@@ -134,7 +134,7 @@ async def _check_configuration_setting(
                     self.LOGGER.debug("Refresh all triggered by key: %s label %s.", key, label)
                     return True, None
             else:
-                raise e
+                raise
         return False, None
 
     @distributed_trace
@@ -345,7 +345,7 @@ async def _validate_snapshot(self, snapshot_name: str) -> bool:
             if e.status_code == 404:
                 self.LOGGER.warning("Snapshot '%s' not found when resolving snapshot.", snapshot_name)
                 return False
-            raise e
+            raise
         if snapshot.composition_type != SnapshotComposition.KEY:
             raise ValueError(f"Composition type for '{snapshot_name}' must be 'key'.")
         return True
@@ -486,20 +486,15 @@ async def refresh_clients(self):
         if self._next_update_time and self._next_update_time > time.time():
             return
 
-        failover_endpoints = await find_auto_failover_endpoints(
-            self._original_endpoint, self._replica_discovery_enabled
-        )
-
-        if failover_endpoints is None:
-            # SRV record not found, so we should refresh after a longer interval
+        try:
+            failover_endpoints = await find_auto_failover_endpoints(
+                self._original_endpoint, self._replica_discovery_enabled
+            )
+        except TimeoutError:
+            # SRV record resolution timed out, so we should refresh after a longer interval
             self._next_update_time = time.time() + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
             return
 
-        if len(failover_endpoints) == 0:
-            # No failover endpoints in SRV record.
-            self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
-            return
-
         discovered_clients = []
         for failover_endpoint in failover_endpoints:
             found_client = False
@@ -535,6 +530,12 @@ async def refresh_clients(self):
                         )
                     )
         self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
+        # Close any replica clients that are no longer part of the failover.
+        retained_endpoints = {self._original_client.endpoint}
+        retained_endpoints.update(client.endpoint for client in discovered_clients)
+        for client in self._replica_clients:
+            if client.endpoint not in retained_endpoints:
+                await client.close()
         if not self._load_balancing_enabled:
             random.shuffle(discovered_clients)
             self._replica_clients = [self._original_client] + discovered_clients
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_discovery.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_discovery.py
index 1713c0d0d833..ebe76d904626 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_discovery.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_discovery.py
@@ -25,8 +25,8 @@ async def find_auto_failover_endpoints(endpoint: str, replica_discovery_enabled:
 
     replicas = await _find_replicas(origin.target)
 
-    if not replicas:
-        return None  # Timeout
+    if replicas is None:
+        raise TimeoutError("Timed out while resolving auto-failover replica endpoints.")
 
     srv_records = [origin] + replicas
     endpoints = []
@@ -42,6 +42,8 @@ async def _find_origin(endpoint):
     uri = urlparse(endpoint).hostname
     request = f"_origin._tcp.{uri}"
     srv_records = await _request_record(request)
+    if srv_records is None:
+        raise TimeoutError("Timed out while resolving auto-failover origin endpoint.")
     if not srv_records:
         return None
     return SRVRecord(srv_records[0])
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py
index 99d03a587468..0fca605ab02b 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py
@@ -217,7 +217,12 @@ async def refresh(self, **kwargs) -> None:
                 self._secret_provider.secret_refresh_timer
                 and self._secret_provider.secret_refresh_timer.needs_refresh()
             ):
-                self._dict.update(await self._secret_provider.refresh_secrets())
+                # Publish a new dict rather than mutating in place. Readers may hold a live view or
+                # iterator over self._dict, and an in-place update could raise "dictionary changed
+                # size during iteration". Every write to self._dict must rebind to a new object.
+                updated_settings = dict(self._dict)
+                updated_settings.update(await self._secret_provider.refresh_secrets())
+                self._dict = updated_settings
             await self._replica_client_manager.refresh_clients()
             self._replica_client_manager.find_active_clients()
             replica_count = self._replica_client_manager.get_client_count() - 1
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_discovery.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_discovery.py
index 89f4d9bafcfc..04de174b5919 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_discovery.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_discovery.py
@@ -114,7 +114,8 @@ async def test_find_replicas(self, mock_request_record):
     async def test_find_origin(self, mock_request_record):
         endpoint = "https://fake.endpoint"
         mock_request_record.return_value = None
-        assert not await _find_origin(endpoint)
+        with pytest.raises(TimeoutError):
+            await _find_origin(endpoint)
         mock_request_record.assert_called_once_with("_origin._tcp.fake.endpoint")
 
         mock_request_record.reset_mock()
@@ -202,3 +203,22 @@ async def test_find_auto_failover_endpoints(self, mock_find_replicas, mock_find_
         assert result[1] == "https://fake2.appconfig.io"
         mock_find_origin.assert_called_once_with("https://fake1.appconfig.io")
         mock_find_replicas.assert_called_once_with("fake.appconfig.io")
+
+        # A timeout while resolving replicas is surfaced as a TimeoutError
+        mock_find_origin.reset_mock()
+        mock_find_replicas.reset_mock()
+        mock_find_origin.return_value = FakeAnswer(1, 99, 5000, "fake.appconfig.io")
+        mock_find_replicas.return_value = None
+        with pytest.raises(TimeoutError):
+            await find_auto_failover_endpoints(endpoint, True)
+        mock_find_origin.assert_called_once_with(endpoint)
+        mock_find_replicas.assert_called_once_with("fake.appconfig.io")
+
+        # A timeout while resolving the origin is surfaced as a TimeoutError
+        mock_find_origin.reset_mock()
+        mock_find_replicas.reset_mock()
+        mock_find_origin.side_effect = TimeoutError("timeout")
+        with pytest.raises(TimeoutError):
+            await find_auto_failover_endpoints(endpoint, True)
+        mock_find_origin.assert_called_once_with(endpoint)
+        mock_find_replicas.assert_not_called()
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager.py
index e8c2c66dc2d9..904ad424f6f6 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager.py
@@ -3,12 +3,17 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from unittest.mock import patch, call, Mock, MagicMock
+import time
+from unittest.mock import patch, call, Mock, MagicMock, AsyncMock
 import pytest
 from azure.appconfiguration.provider.aio._async_client_manager import (
     AsyncConfigurationClientManager,
     _AsyncConfigurationClientWrapper,
 )
+from azure.appconfiguration.provider._client_manager_base import (
+    FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL,
+    MINIMAL_CLIENT_REFRESH_INTERVAL,
+)
 from azure.appconfiguration.provider._models import SettingSelector
 
 
@@ -26,6 +31,10 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
         self.credential = credential
         self.retry_total = retry_total
         self.retry_backoff = retry_backoff
+        self.closed = False
+
+    async def close(self):
+        self.closed = True
 
 
 @pytest.mark.usefixtures("caplog")
@@ -218,6 +227,35 @@ async def test_refresh_clients_credential(
         assert len(manager._replica_clients) == 2
         mock_client.assert_not_called()
 
+    @pytest.mark.asyncio
+    @patch("azure.appconfiguration.provider.aio._async_client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider.aio._async_client_manager._AsyncConfigurationClientWrapper.from_credential")
+    async def test_refresh_clients_closes_removed_replicas(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+        mock_update_failover_endpoints.return_value = []
+        mock_client.return_value = MockClient(endpoint, "", "fake-credential", 0, 0)
+        manager = AsyncConfigurationClientManager(
+            None, endpoint, _create_mock_credential(), "", 0, 0, True, 0, 0, False
+        )
+
+        original_client = MockClient(endpoint, "", "fake-credential", 0, 0)
+        retained_replica = MockClient("https://fake.endpoint2", "", "fake-credential", 0, 0)
+        removed_replica = MockClient("https://fake.endpoint3", "", "fake-credential", 0, 0)
+        manager._original_client = original_client
+        manager._replica_clients = [original_client, retained_replica, removed_replica]
+
+        # endpoint3 is no longer part of the failover set, so its client must be closed and dropped.
+        mock_update_failover_endpoints.return_value = ["https://fake.endpoint2"]
+        manager._next_update_time = 0
+        await manager.refresh_clients()
+
+        assert removed_replica.closed is True
+        assert retained_replica.closed is False
+        assert original_client.closed is False
+        assert removed_replica not in manager._replica_clients
+        assert retained_replica in manager._replica_clients
+        assert original_client in manager._replica_clients
+
     @pytest.mark.asyncio
     @patch("azure.appconfiguration.provider.aio._async_client_manager.find_auto_failover_endpoints")
     @patch(
@@ -325,6 +363,75 @@ async def test_refresh_clients_connection_string(
         assert len(manager._replica_clients) == 2
         mock_client.assert_not_called()
 
+    @pytest.mark.asyncio
+    @patch("azure.appconfiguration.provider.aio._async_client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider.aio._async_client_manager._AsyncConfigurationClientWrapper.from_credential")
+    async def test_refresh_clients_timeout_uses_fallback_interval(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+
+        mock_client.return_value = MockClient("https://fake.endpoint", "", "fake-credential", 0, 0)
+        mock_update_failover_endpoints.return_value = []
+        manager = AsyncConfigurationClientManager(None, endpoint, "fake-credential", "", 0, 0, True, 0, 0, False)
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_client.reset_mock()
+
+        # A timeout while resolving replicas falls back to the longer refresh interval
+        mock_update_failover_endpoints.side_effect = TimeoutError
+        manager._next_update_time = 0
+        before = time.time()
+        await manager.refresh_clients()
+        mock_update_failover_endpoints.assert_called_once_with(endpoint, True)
+        # No new clients should have been created on a timeout
+        mock_client.assert_not_called()
+        assert manager._next_update_time >= before + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_update_failover_endpoints.side_effect = None
+
+        # An empty result (no timeout) refreshes at the normal minimal interval
+        mock_update_failover_endpoints.return_value = []
+        manager._next_update_time = 0
+        before = time.time()
+        await manager.refresh_clients()
+        mock_update_failover_endpoints.assert_called_once_with(endpoint, True)
+        assert (
+            before + MINIMAL_CLIENT_REFRESH_INTERVAL
+            <= manager._next_update_time
+            < before + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
+        )
+
+    @pytest.mark.asyncio
+    @patch("azure.appconfiguration.provider.aio._async_client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider.aio._async_client_manager._AsyncConfigurationClientWrapper.from_credential")
+    async def test_refresh_clients_empty_removes_replicas(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+
+        mock_client.return_value = MockClient("https://fake.endpoint", "", "fake-credential", 0, 0)
+        mock_update_failover_endpoints.return_value = []
+        manager = AsyncConfigurationClientManager(None, endpoint, "fake-credential", "", 0, 0, True, 0, 0, False)
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_client.reset_mock()
+
+        # Discover a replica
+        replica = MockClient("https://fake.endpoint2", "", "fake-credential", 0, 0)
+        mock_client.return_value = replica
+        mock_update_failover_endpoints.return_value = ["https://fake.endpoint2"]
+        manager._next_update_time = 0
+        await manager.refresh_clients()
+        assert len(manager._replica_clients) == 2
+
+        # A subsequent successful discovery with no replicas (e.g. the last replica was deleted)
+        # closes and removes the discovered client, leaving only the original.
+        replica.close = AsyncMock()
+        mock_update_failover_endpoints.return_value = []
+        manager._next_update_time = 0
+        await manager.refresh_clients()
+        replica.close.assert_awaited_once()
+        assert len(manager._replica_clients) == 1
+        assert manager._replica_clients[0] is manager._original_client
+
     @patch("azure.appconfiguration.provider.aio._async_client_manager.find_auto_failover_endpoints")
     def test_calculate_backoff(self, mock_update_failover_endpoints):
         endpoint = "https://fake.endpoint"
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager_load_balance.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager_load_balance.py
index f252798367b1..7d5ac234453a 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager_load_balance.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_configuration_async_client_manager_load_balance.py
@@ -24,6 +24,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
         self.retry_total = retry_total
         self.retry_backoff = retry_backoff
 
+    async def close(self):
+        pass
+
 
 @pytest.mark.usefixtures("caplog")
 class TestConfigurationAsyncClientManagerLoadBalance:
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_azureappconfigurationproviderbase.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_azureappconfigurationproviderbase.py
index 5073f2afad31..8a13a686a322 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_azureappconfigurationproviderbase.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_azureappconfigurationproviderbase.py
@@ -67,6 +67,11 @@ def test_string_input(self):
         result = _build_watched_setting("test_key")
         self.assertEqual(result, ("test_key", NULL_CHAR))
 
+    def test_two_character_string_input(self):
+        """Test with a two-character string input is treated as a key, not unpacked character-by-character."""
+        result = _build_watched_setting("ab")
+        self.assertEqual(result, ("ab", NULL_CHAR))
+
     def test_tuple_input(self):
         """Test with tuple input."""
         result = _build_watched_setting(("test_key", "test_label"))
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager.py
index 5edaab158bf4..423322f41e3b 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager.py
@@ -3,10 +3,15 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
+import time
 import unittest
 from unittest.mock import patch, call, Mock, MagicMock
 import pytest
 from azure.appconfiguration.provider._client_manager import ConfigurationClientManager, _ConfigurationClientWrapper
+from azure.appconfiguration.provider._client_manager_base import (
+    FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL,
+    MINIMAL_CLIENT_REFRESH_INTERVAL,
+)
 from azure.appconfiguration.provider._models import SettingSelector
 
 
@@ -25,6 +30,10 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
         self.credential = credential
         self.retry_total = retry_total
         self.retry_backoff = retry_backoff
+        self.closed = False
+
+    def close(self):
+        self.closed = True
 
 
 @pytest.mark.usefixtures("caplog")
@@ -199,6 +208,32 @@ def test_refresh_clients_credential(
         assert len(manager._replica_clients) == 2
         mock_client.assert_not_called()
 
+    @patch("azure.appconfiguration.provider._client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider._client_manager._ConfigurationClientWrapper.from_credential")
+    def test_refresh_clients_closes_removed_replicas(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+        mock_update_failover_endpoints.return_value = []
+        mock_client.return_value = MockClient(endpoint, "", "fake-credential", 0, 0)
+        manager = ConfigurationClientManager(None, endpoint, _create_mock_credential(), "", 0, 0, True, 0, 0, False)
+
+        original_client = MockClient(endpoint, "", "fake-credential", 0, 0)
+        retained_replica = MockClient("https://fake.endpoint2", "", "fake-credential", 0, 0)
+        removed_replica = MockClient("https://fake.endpoint3", "", "fake-credential", 0, 0)
+        manager._original_client = original_client
+        manager._replica_clients = [original_client, retained_replica, removed_replica]
+
+        # endpoint3 is no longer part of the failover set, so its client must be closed and dropped.
+        mock_update_failover_endpoints.return_value = ["https://fake.endpoint2"]
+        manager._next_update_time = 0
+        manager.refresh_clients()
+
+        assert removed_replica.closed is True
+        assert retained_replica.closed is False
+        assert original_client.closed is False
+        assert removed_replica not in manager._replica_clients
+        assert retained_replica in manager._replica_clients
+        assert original_client in manager._replica_clients
+
     @patch("azure.appconfiguration.provider._client_manager.find_auto_failover_endpoints")
     @patch("azure.appconfiguration.provider._client_manager._ConfigurationClientWrapper.from_connection_string")
     def test_refresh_clients_connection_string(
@@ -302,6 +337,73 @@ def test_refresh_clients_connection_string(
         assert len(manager._replica_clients) == 2
         mock_client.assert_not_called()
 
+    @patch("azure.appconfiguration.provider._client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider._client_manager._ConfigurationClientWrapper.from_credential")
+    def test_refresh_clients_timeout_uses_fallback_interval(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+
+        mock_client.return_value = MockClient("https://fake.endpoint", "", "fake-credential", 0, 0)
+        mock_update_failover_endpoints.return_value = []
+        manager = ConfigurationClientManager(None, endpoint, "fake-credential", "", 0, 0, True, 0, 0, False)
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_client.reset_mock()
+
+        # A timeout while resolving replicas falls back to the longer refresh interval
+        mock_update_failover_endpoints.side_effect = TimeoutError
+        manager._next_update_time = 0
+        before = time.time()
+        manager.refresh_clients()
+        mock_update_failover_endpoints.assert_called_once_with(endpoint, True)
+        # No new clients should have been created on a timeout
+        mock_client.assert_not_called()
+        assert manager._next_update_time >= before + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_update_failover_endpoints.side_effect = None
+
+        # An empty result (no timeout) refreshes at the normal minimal interval
+        mock_update_failover_endpoints.return_value = []
+        manager._next_update_time = 0
+        before = time.time()
+        manager.refresh_clients()
+        mock_update_failover_endpoints.assert_called_once_with(endpoint, True)
+        assert (
+            before + MINIMAL_CLIENT_REFRESH_INTERVAL
+            <= manager._next_update_time
+            < before + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
+        )
+
+    @patch("azure.appconfiguration.provider._client_manager.find_auto_failover_endpoints")
+    @patch("azure.appconfiguration.provider._client_manager._ConfigurationClientWrapper.from_credential")
+    def test_refresh_clients_empty_removes_replicas(self, mock_client, mock_update_failover_endpoints):
+        endpoint = "https://fake.endpoint"
+
+        mock_client.return_value = MockClient("https://fake.endpoint", "", "fake-credential", 0, 0)
+        mock_update_failover_endpoints.return_value = []
+        manager = ConfigurationClientManager(None, endpoint, "fake-credential", "", 0, 0, True, 0, 0, False)
+
+        mock_update_failover_endpoints.reset_mock()
+        mock_client.reset_mock()
+
+        # Discover a replica
+        replica = MockClient("https://fake.endpoint2", "", "fake-credential", 0, 0)
+        mock_client.return_value = replica
+        mock_update_failover_endpoints.return_value = ["https://fake.endpoint2"]
+        manager._next_update_time = 0
+        manager.refresh_clients()
+        assert len(manager._replica_clients) == 2
+
+        # A subsequent successful discovery with no replicas (e.g. the last replica was deleted)
+        # closes and removes the discovered client, leaving only the original.
+        replica.close = Mock()
+        mock_update_failover_endpoints.return_value = []
+        manager._next_update_time = 0
+        manager.refresh_clients()
+        replica.close.assert_called_once()
+        assert len(manager._replica_clients) == 1
+        assert manager._replica_clients[0] is manager._original_client
+
     @patch("azure.appconfiguration.provider._client_manager.find_auto_failover_endpoints")
     def test_calculate_backoff(self, mock_update_failover_endpoints):
         endpoint = "https://fake.endpoint"
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager_load_balance.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager_load_balance.py
index f73942364e19..3d0fca436379 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager_load_balance.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_configuration_client_manager_load_balance.py
@@ -25,6 +25,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
         self.retry_total = retry_total
         self.retry_backoff = retry_backoff
 
+    def close(self):
+        pass
+
 
 @pytest.mark.usefixtures("caplog")
 class TestConfigurationClientManagerLoadBalance(unittest.TestCase):
diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_discovery.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_discovery.py
index 7e3e60b42bd2..51a22a45a38c 100644
--- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_discovery.py
+++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_discovery.py
@@ -111,7 +111,8 @@ def test_find_replicas(self, mock_request_record):
     def test_find_origin(self, mock_request_record):
         endpoint = "https://fake.endpoint"
         mock_request_record.return_value = None
-        assert not _find_origin(endpoint)
+        with pytest.raises(TimeoutError):
+            _find_origin(endpoint)
         mock_request_record.assert_called_once_with("_origin._tcp.fake.endpoint")
 
         mock_request_record.reset_mock()
@@ -198,3 +199,22 @@ def test_find_auto_failover_endpoints(self, mock_find_replicas, mock_find_origin
         assert result[1] == "https://fake2.appconfig.io"
         mock_find_origin.assert_called_once_with("https://fake1.appconfig.io")
         mock_find_replicas.assert_called_once_with("fake.appconfig.io")
+
+        # A timeout while resolving replicas is surfaced as a TimeoutError
+        mock_find_origin.reset_mock()
+        mock_find_replicas.reset_mock()
+        mock_find_origin.return_value = FakeAnswer(1, 99, 5000, "fake.appconfig.io")
+        mock_find_replicas.return_value = None
+        with pytest.raises(TimeoutError):
+            find_auto_failover_endpoints(endpoint, True)
+        mock_find_origin.assert_called_once_with(endpoint)
+        mock_find_replicas.assert_called_once_with("fake.appconfig.io")
+
+        # A timeout while resolving the origin is surfaced as a TimeoutError
+        mock_find_origin.reset_mock()
+        mock_find_replicas.reset_mock()
+        mock_find_origin.side_effect = TimeoutError("timeout")
+        with pytest.raises(TimeoutError):
+            find_auto_failover_endpoints(endpoint, True)
+        mock_find_origin.assert_called_once_with(endpoint)
+        mock_find_replicas.assert_not_called()