refactor: hoist stale-loop comprehension into a named local

Address review feedback on the stale-lock cleanup in
AsyncAccessTokenManager._get_refresh_lock(). The list comprehension was
embedded directly in the `for` statement, which read awkwardly across the
wrapped line.

Hoist it into a named `stale_loops` local and iterate over that. This is purely
a readability change with no behavioural difference: the snapshot is still
materialized before mutating the mapping, which is what makes deleting during
iteration safe.

Note this was not a lint violation - flake8 passes on the file both before and
after the change.

Also align the async deferred-acquisition test with the naming and assertions
that PR #199 introduced for the synchronous manager, and assert explicitly that
constructing AsyncAccessTokenManager performs no credential call. The async
manager has always deferred its first acquisition, so the cold-start burst
flows through the same refresh guard as a later refresh; the test now states
that intent directly.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: bb08d0ed-484d-4015-b905-3e164f68678b

Author: Bernd Verst

durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py

@@ -86,8 +86,10 @@ def _get_refresh_lock(self) -> asyncio.Lock:
             if lock is None:
                 # Discard locks belonging to loops that are no longer usable so the
                 # mapping does not grow without bound.
-                for stale_loop in [
-                        existing for existing in self._refresh_locks if existing.is_closed()]:
+                stale_loops = [
+                    existing for existing in self._refresh_locks if existing.is_closed()
+                ]
+                for stale_loop in stale_loops:
                     del self._refresh_locks[stale_loop]
                 lock = asyncio.Lock()
                 self._refresh_locks[loop] = lock

tests/durabletask-azuremanaged/test_durabletask_grpc_interceptor.py

@@ -266,10 +266,15 @@ async def get_token(self, _scope):
 
 
 class TestAsyncAccessTokenManagerSingleFlight(unittest.IsolatedAsyncioTestCase):
-    async def test_concurrent_cold_start_performs_single_acquisition(self):
+    async def test_deferred_first_acquisition_performs_single_acquisition(self):
         credential = _TestAsyncTokenCredential()
         manager = AsyncAccessTokenManager(credential)
 
+        self.assertEqual(0, credential.calls, "Constructing the manager must not acquire a token")
+
+        # The async manager has always deferred the first acquisition (a constructor
+        # cannot await), so the cold-start burst goes through the same refresh guard
+        # as a later refresh and must likewise be single-flight.
         tokens = await asyncio.gather(*(manager.get_access_token() for _ in range(16)))
 
         self.assertEqual(1, credential.calls)