diff --git a/src/globus_cli/commands/cli_profile_list.py b/src/globus_cli/commands/cli_profile_list.py
index e2bb2ee26..39cd44e54 100644
--- a/src/globus_cli/commands/cli_profile_list.py
+++ b/src/globus_cli/commands/cli_profile_list.py
@@ -5,7 +5,7 @@
 
 import click
 
-from globus_cli.login_manager import is_client_login, token_storage_adapter
+from globus_cli.login_manager import LoginManager, is_client_login
 from globus_cli.parsing import command
 from globus_cli.termio import Field, display, formatters
 
@@ -29,13 +29,16 @@ def _profilestr_to_datadict(s: str) -> dict[str, t.Any] | None:
 
 
 def _parse_and_filter_profiles(
+    login_manager: LoginManager,
     all: bool,
 ) -> tuple[list[dict[str, t.Any]], list[dict[str, t.Any]]]:
     globus_env = os.getenv("GLOBUS_SDK_ENVIRONMENT", "production")
 
     client_profiles = []
     user_profiles = []
-    for n in token_storage_adapter().iter_namespaces(include_config_namespaces=True):
+    for n in login_manager.storage.adapter.iter_namespaces(
+        include_config_namespaces=True
+    ):
         data = _profilestr_to_datadict(n)
         if not data:  # skip any parse failures
             continue
@@ -86,8 +89,8 @@ def cli_profile_list(*, all: bool) -> None:
     These are the values for GLOBUS_PROFILE which have been recorded, as well as
     GLOBUS_CLI_CLIENT_ID values which have been used.
     """
-
-    client_profiles, user_profiles = _parse_and_filter_profiles(all)
+    login_manager = LoginManager()
+    client_profiles, user_profiles = _parse_and_filter_profiles(login_manager, all)
 
     if user_profiles:
         fields = [
diff --git a/src/globus_cli/commands/collection/_common.py b/src/globus_cli/commands/collection/_common.py
index f0e203d39..e4244a0b8 100644
--- a/src/globus_cli/commands/collection/_common.py
+++ b/src/globus_cli/commands/collection/_common.py
@@ -3,7 +3,7 @@
 import click
 import globus_sdk
 
-from globus_cli.login_manager.utils import get_current_identity_id
+from globus_cli.login_manager import LoginManager
 from globus_cli.termio import Field, formatters
 from globus_cli.types import DATA_CONTAINER_T
 
@@ -12,10 +12,9 @@ class LazyCurrentIdentity:
     def __init__(self, value: str | None) -> None:
         self._value = value
 
-    @property
-    def value(self) -> str:
+    def resolve(self, login_manager: LoginManager) -> str:
         if self._value is None:
-            self._value = get_current_identity_id()
+            self._value = login_manager.get_current_identity_id()
         return str(self._value)
 
 
diff --git a/src/globus_cli/commands/collection/create/guest.py b/src/globus_cli/commands/collection/create/guest.py
index 1bf7b8b01..5fc2086cf 100644
--- a/src/globus_cli/commands/collection/create/guest.py
+++ b/src/globus_cli/commands/collection/create/guest.py
@@ -91,7 +91,10 @@ def collection_create_guest(
 
     if not user_credential_id:
         user_credential_id = _select_user_credential_id(
-            gcs_client, mapped_collection_id, local_username, identity_id.value
+            gcs_client,
+            mapped_collection_id,
+            local_username,
+            identity_id.resolve(login_manager),
         )
 
     converted_kwargs: dict[str, t.Any] = ExplicitNullType.nullify_dict(
@@ -105,7 +108,7 @@ def collection_create_guest(
             "display_name": display_name,
             "enable_https": enable_https,
             "force_encryption": force_encryption,
-            "identity_id": identity_id.value,
+            "identity_id": identity_id.resolve(login_manager),
             "info_link": info_link,
             "keywords": keywords,
             "mapped_collection_id": mapped_collection_id,
diff --git a/src/globus_cli/commands/collection/create/mapped.py b/src/globus_cli/commands/collection/create/mapped.py
index 238663963..e8a33ced5 100644
--- a/src/globus_cli/commands/collection/create/mapped.py
+++ b/src/globus_cli/commands/collection/create/mapped.py
@@ -284,7 +284,7 @@ def collection_create_mapped(
             "domain_name": domain_name,
             "enable_https": enable_https,
             "force_encryption": force_encryption,
-            "identity_id": identity_id.value,
+            "identity_id": identity_id.resolve(login_manager),
             "info_link": info_link,
             "keywords": keywords,
             "organization": organization,
diff --git a/src/globus_cli/commands/flows/run/resume.py b/src/globus_cli/commands/flows/run/resume.py
index 9f099d3f1..8283b2aab 100644
--- a/src/globus_cli/commands/flows/run/resume.py
+++ b/src/globus_cli/commands/flows/run/resume.py
@@ -8,7 +8,6 @@
 import globus_sdk
 
 from globus_cli.login_manager import LoginManager
-from globus_cli.login_manager.utils import get_current_identity_id
 from globus_cli.parsing import command, run_id_arg
 from globus_cli.termio import Field, display, formatters
 from globus_cli.utils import CLIAuthRequirementsError
@@ -124,6 +123,6 @@ def _has_required_consent(
     login_manager: LoginManager, required_scopes: list[str]
 ) -> bool:
     auth_client = login_manager.get_auth_client()
-    user_identity_id = get_current_identity_id()
+    user_identity_id = login_manager.get_current_identity_id()
     consents = auth_client.get_consents(user_identity_id).to_forest()
     return consents.meets_scope_requirements(required_scopes)
diff --git a/src/globus_cli/commands/logout.py b/src/globus_cli/commands/logout.py
index f4aaa2a90..8a8ee49aa 100644
--- a/src/globus_cli/commands/logout.py
+++ b/src/globus_cli/commands/logout.py
@@ -1,14 +1,7 @@
 import click
 import globus_sdk
 
-from globus_cli.login_manager import (
-    LoginManager,
-    delete_templated_client,
-    internal_native_client,
-    is_client_login,
-    remove_well_known_config,
-    token_storage_adapter,
-)
+from globus_cli.login_manager import LoginManager, is_client_login
 from globus_cli.parsing import command
 
 
@@ -105,7 +98,7 @@ def logout_command(
     # Always skip for client logins, which don't use a templated client
     if delete_client and not is_client_login():
         try:
-            delete_templated_client()
+            login_manager.storage.delete_templated_client()
         except globus_sdk.AuthAPIError:
             if not ignore_errors:
                 warnecho(
@@ -121,10 +114,9 @@ def logout_command(
 
     # Attempt to revoke all tokens in storage; use the internal native client to ensure
     # we have a valid Auth client
-    native_client = internal_native_client()
-    adapter = token_storage_adapter()
+    native_client = login_manager.storage.cli_native_client
 
-    for rs, tokendata in adapter.get_by_resource_server().items():
+    for rs, tokendata in login_manager.storage.adapter.get_by_resource_server().items():
         for tok_key in ("access_token", "refresh_token"):
             token = tokendata[tok_key]
 
@@ -145,9 +137,9 @@ def logout_command(
                         "Continuing... (--ignore-errors)",
                     )
 
-        adapter.remove_tokens_for_resource_server(rs)
+        login_manager.storage.adapter.remove_tokens_for_resource_server(rs)
 
-    remove_well_known_config("auth_user_data")
+    login_manager.storage.remove_well_known_config("auth_user_data")
 
     if is_client_login():
         click.echo(_CLIENT_LOGOUT_EPILOG)
diff --git a/src/globus_cli/commands/session/show.py b/src/globus_cli/commands/session/show.py
index c8d4eecec..190be42c0 100644
--- a/src/globus_cli/commands/session/show.py
+++ b/src/globus_cli/commands/session/show.py
@@ -5,13 +5,7 @@
 
 import globus_sdk
 
-from globus_cli.login_manager import (
-    LoginManager,
-    get_client_login,
-    internal_auth_client,
-    is_client_login,
-    token_storage_adapter,
-)
+from globus_cli.login_manager import LoginManager, get_client_login, is_client_login
 from globus_cli.parsing import command
 from globus_cli.termio import Field, display, print_command_hint
 
@@ -54,7 +48,6 @@ def session_show(login_manager: LoginManager) -> None:
     the time the user authenticated with that identity.
     """
     auth_client = login_manager.get_auth_client()
-    adapter = token_storage_adapter()
 
     # get a token to introspect, refreshing if necessary
     try:
@@ -64,7 +57,7 @@ def session_show(login_manager: LoginManager) -> None:
     except AttributeError:  # if we have no RefreshTokenAuthorizor
         pass
 
-    tokendata = adapter.get_token_data("auth.globus.org")
+    tokendata = login_manager.storage.adapter.get_token_data("auth.globus.org")
     # if there's no token (e.g. not logged in), stub with empty data
     if not tokendata:
         session_info: dict[str, t.Any] = {}
@@ -73,7 +66,7 @@ def session_show(login_manager: LoginManager) -> None:
         if is_client_login():
             introspect_client = get_client_login()
         else:
-            introspect_client = internal_auth_client()
+            introspect_client = login_manager.storage.cli_confidential_client
 
         access_token = tokendata["access_token"]
         res = introspect_client.oauth2_token_introspect(
diff --git a/src/globus_cli/commands/timer/create/transfer.py b/src/globus_cli/commands/timer/create/transfer.py
index c767f950c..c30184982 100644
--- a/src/globus_cli/commands/timer/create/transfer.py
+++ b/src/globus_cli/commands/timer/create/transfer.py
@@ -10,7 +10,6 @@
 
 from globus_cli.endpointish import Endpointish
 from globus_cli.login_manager import LoginManager, is_client_login
-from globus_cli.login_manager.utils import get_current_identity_id
 from globus_cli.parsing import (
     ENDPOINT_PLUS_OPTPATH,
     TimedeltaType,
@@ -265,7 +264,9 @@ def transfer_command(
         # If it's not a client login, we need to check
         # that the user has the required scopes
         if not is_client_login():
-            request_data_access = _derive_missing_scopes(auth_client, scopes_needed)
+            request_data_access = _derive_missing_scopes(
+                login_manager, auth_client, scopes_needed
+            )
 
             if request_data_access:
                 scope_request_opts = " ".join(
@@ -349,11 +350,12 @@ def _derive_needed_scopes(
 
 
 def _derive_missing_scopes(
+    login_manager: LoginManager,
     auth_client: CustomAuthClient,
     scopes_needed: dict[str, Scope],
 ) -> list[str]:
     # read the identity ID stored from the login flow
-    user_identity_id = get_current_identity_id()
+    user_identity_id = login_manager.get_current_identity_id()
 
     # get the user's Globus CLI consents
     consents = auth_client.get_consents(user_identity_id).to_forest()
diff --git a/src/globus_cli/commands/timer/resume.py b/src/globus_cli/commands/timer/resume.py
index 04b47dfbf..7c8c7e3e6 100644
--- a/src/globus_cli/commands/timer/resume.py
+++ b/src/globus_cli/commands/timer/resume.py
@@ -8,7 +8,6 @@
 import globus_sdk
 
 from globus_cli.login_manager import LoginManager
-from globus_cli.login_manager.utils import get_current_identity_id
 from globus_cli.parsing import command
 from globus_cli.termio import display
 from globus_cli.utils import CLIAuthRequirementsError
@@ -117,6 +116,6 @@ def _has_required_consent(
     login_manager: LoginManager, required_scopes: list[str]
 ) -> bool:
     auth_client = login_manager.get_auth_client()
-    user_identity_id = get_current_identity_id()
+    user_identity_id = login_manager.get_current_identity_id()
     consents = auth_client.get_consents(user_identity_id).to_forest()
     return consents.meets_scope_requirements(required_scopes)
diff --git a/src/globus_cli/login_manager/__init__.py b/src/globus_cli/login_manager/__init__.py
index 567569d64..67a0df8ea 100644
--- a/src/globus_cli/login_manager/__init__.py
+++ b/src/globus_cli/login_manager/__init__.py
@@ -2,29 +2,13 @@
 from .errors import MissingLoginError
 from .manager import LoginManager
 from .scopes import compute_timer_scope
-from .tokenstore import (
-    delete_templated_client,
-    internal_auth_client,
-    internal_native_client,
-    read_well_known_config,
-    remove_well_known_config,
-    store_well_known_config,
-    token_storage_adapter,
-)
 from .utils import is_remote_session
 
 __all__ = [
     "MissingLoginError",
     "is_remote_session",
     "LoginManager",
-    "delete_templated_client",
-    "internal_auth_client",
-    "internal_native_client",
-    "token_storage_adapter",
     "is_client_login",
     "get_client_login",
-    "store_well_known_config",
-    "read_well_known_config",
-    "remove_well_known_config",
     "compute_timer_scope",
 ]
diff --git a/src/globus_cli/login_manager/auth_flows.py b/src/globus_cli/login_manager/auth_flows.py
index 8c5e3c586..1872224a2 100644
--- a/src/globus_cli/login_manager/auth_flows.py
+++ b/src/globus_cli/login_manager/auth_flows.py
@@ -8,15 +8,11 @@
 import globus_sdk
 from globus_sdk.scopes import Scope
 
-from .tokenstore import (
-    internal_auth_client,
-    read_well_known_config,
-    store_well_known_config,
-    token_storage_adapter,
-)
+from .storage import CLIStorage
 
 
 def do_link_auth_flow(
+    storage: CLIStorage,
     scopes: str | t.Sequence[str | Scope],
     *,
     session_params: dict[str, str] | None = None,
@@ -28,7 +24,7 @@ def do_link_auth_flow(
     session_params = session_params or {}
 
     # get the ConfidentialApp client object
-    auth_client = internal_auth_client()
+    auth_client = storage.cli_confidential_client
 
     # start the Confidential App Grant flow
     auth_client.oauth2_start_flow(
@@ -53,11 +49,12 @@ def do_link_auth_flow(
     auth_code = click.prompt("Enter the resulting Authorization Code here").strip()
 
     # finish auth flow
-    exchange_code_and_store(auth_client, auth_code)
+    exchange_code_and_store(storage, auth_client, auth_code)
     return True
 
 
 def do_local_server_auth_flow(
+    storage: CLIStorage,
     scopes: str | t.Sequence[str | Scope],
     *,
     session_params: dict[str, str] | None = None,
@@ -78,7 +75,7 @@ def do_local_server_auth_flow(
         redirect_uri = f"http://localhost:{port}"
 
         # get the ConfidentialApp client object and start a flow
-        auth_client = internal_auth_client()
+        auth_client = storage.cli_confidential_client
         auth_client.oauth2_start_flow(
             refresh_tokens=True,
             redirect_uri=redirect_uri,
@@ -103,11 +100,12 @@ def do_local_server_auth_flow(
         click.get_current_context().exit(1)
 
     # finish auth flow and return true
-    exchange_code_and_store(auth_client, auth_code)
+    exchange_code_and_store(storage, auth_client, auth_code)
     return True
 
 
 def exchange_code_and_store(
+    storage: CLIStorage,
     auth_client: globus_sdk.ConfidentialAppAuthClient | globus_sdk.NativeAppAuthClient,
     auth_code: str,
 ) -> None:
@@ -121,7 +119,6 @@ def exchange_code_and_store(
     """
     import jwt.exceptions
 
-    adapter = token_storage_adapter()
     tkn = auth_client.oauth2_exchange_code_for_tokens(auth_code)
 
     # use a leeway of 300s
@@ -156,7 +153,7 @@ def exchange_code_and_store(
                 err=True,
             )
         raise
-    auth_user_data = read_well_known_config("auth_user_data")
+    auth_user_data = storage.read_well_known_config("auth_user_data")
     if auth_user_data and sub_new != auth_user_data.get("sub"):
         try:
             for tokens in tkn.by_resource_server.values():
@@ -171,8 +168,8 @@ def exchange_code_and_store(
             )
         click.get_current_context().exit(1)
     if not auth_user_data:
-        store_well_known_config("auth_user_data", {"sub": sub_new})
-    adapter.store(tkn)
+        storage.store_well_known_config("auth_user_data", {"sub": sub_new})
+    storage.store(tkn)
 
 
 def _response_clock_delta(response: globus_sdk.GlobusHTTPResponse) -> float | None:
diff --git a/src/globus_cli/login_manager/manager.py b/src/globus_cli/login_manager/manager.py
index 5e9124f09..e1707463a 100644
--- a/src/globus_cli/login_manager/manager.py
+++ b/src/globus_cli/login_manager/manager.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import functools
+import os
 import sys
 import typing as t
 import uuid
@@ -29,12 +30,8 @@
 from .context import LoginContext
 from .errors import MissingLoginError
 from .scopes import CLI_SCOPE_REQUIREMENTS
-from .tokenstore import (
-    internal_auth_client,
-    read_well_known_config,
-    token_storage_adapter,
-)
-from .utils import get_current_identity_id, is_remote_session
+from .storage import CLIStorage
+from .utils import is_remote_session
 
 if t.TYPE_CHECKING:
     from ..services.auth import CustomAuthClient
@@ -52,9 +49,12 @@
 
 class LoginManager:
     def __init__(self) -> None:
-        self._token_storage = token_storage_adapter()
+        self.storage = CLIStorage()
         self._nonstatic_requirements: dict[str, list[str | Scope]] = {}
 
+    def close(self) -> None:
+        self.storage.close()
+
     def add_requirement(self, rs_name: str, scopes: t.Sequence[str | Scope]) -> None:
         self._nonstatic_requirements[rs_name] = list(scopes)
 
@@ -83,7 +83,7 @@ def is_logged_in(self) -> bool:
         )
 
     def _validate_token(self, token: str) -> bool:
-        auth_client = internal_auth_client()
+        auth_client = self.storage.cli_confidential_client
         try:
             res = auth_client.post(
                 "/v2/oauth2/token/validate", data={"token": token}, encoding="form"
@@ -105,7 +105,7 @@ def has_login(self, resource_server: str) -> bool:
         if is_client_login():
             return True
 
-        tokens = self._token_storage.get_token_data(resource_server)
+        tokens = self.storage.adapter.get_token_data(resource_server)
         if tokens is None or "refresh_token" not in tokens:
             return False
 
@@ -132,7 +132,9 @@ def _tokens_meet_static_requirements(
         # evaluate scope contract version requirements for this service
 
         # first, fetch the version data and if it is missing, treat it as empty
-        contract_versions = read_well_known_config("scope_contract_versions") or {}
+        contract_versions = (
+            self.storage.read_well_known_config("scope_contract_versions") or {}
+        )
         # determine which version we need, and compare against the version in
         # storage with a default of 0
         # if the comparison fails, reject the token as not a valid login for the
@@ -181,7 +183,7 @@ def _tokens_meet_nonstatic_requirements(
     @property
     @functools.lru_cache(maxsize=1)  # noqa: B019
     def _cached_consent_forest(self) -> ConsentForest:
-        identity_id = get_current_identity_id()
+        identity_id = self.get_current_identity_id()
 
         return self.get_auth_client().get_consents(identity_id).to_forest()
 
@@ -213,12 +215,14 @@ def run_login_flow(
                 scopes.append(s)
         # use a link login if remote session or user requested
         if no_local_server or is_remote_session():
-            do_link_auth_flow(scopes, session_params=session_params)
+            do_link_auth_flow(self.storage, scopes, session_params=session_params)
         # otherwise default to a local server login flow
         else:
             if local_server_message is not None:
                 click.echo(local_server_message)
-            do_local_server_auth_flow(scopes, session_params=session_params)
+            do_local_server_auth_flow(
+                self.storage, scopes, session_params=session_params
+            )
 
         if epilog is not None:
             click.echo(epilog)
@@ -286,6 +290,9 @@ def inner(
             @functools.wraps(func)
             def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                 manager = cls()
+                context = click.get_current_context()
+                context.call_on_close(manager.close)
+
                 manager.assert_logins(*resource_servers)
                 return func(manager, *args, **kwargs)
 
@@ -296,7 +303,7 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
     def _get_client_authorizer(
         self, resource_server: str, *, no_tokens_msg: str | None = None
     ) -> globus_sdk.ClientCredentialsAuthorizer | globus_sdk.RefreshTokenAuthorizer:
-        tokens = self._token_storage.get_token_data(resource_server)
+        tokens = self.storage.adapter.get_token_data(resource_server)
 
         if is_client_login():
             # construct scopes for the specified resource server.
@@ -321,7 +328,7 @@ def _get_client_authorizer(
                 scopes=scopes,
                 access_token=access_token,
                 expires_at=expires_at,
-                on_refresh=self._token_storage.on_refresh,
+                on_refresh=self.storage.store,
             )
 
         else:
@@ -337,10 +344,10 @@ def _get_client_authorizer(
 
             return globus_sdk.RefreshTokenAuthorizer(
                 tokens["refresh_token"],
-                internal_auth_client(),
+                self.storage.cli_confidential_client,
                 access_token=tokens["access_token"],
                 expires_at=tokens["expires_at_seconds"],
-                on_refresh=self._token_storage.on_refresh,
+                on_refresh=self.storage.store,
             )
 
     def get_transfer_client(self) -> CustomTransferClient:
@@ -485,3 +492,18 @@ def get_gcs_client(
             authorizer=authorizer,
             app_name=version.app_name,
         )
+
+    def get_current_identity_id(self) -> str:
+        """
+        Return the current user's identity ID.
+        For a client-authorized invocation, that's the client ID.
+        """
+
+        if is_client_login():
+            return os.environ["GLOBUS_CLI_CLIENT_ID"]
+        else:
+            user_data = self.storage.read_well_known_config(
+                "auth_user_data", allow_null=False
+            )
+            sub: str = user_data["sub"]
+            return sub
diff --git a/src/globus_cli/login_manager/storage.py b/src/globus_cli/login_manager/storage.py
new file mode 100644
index 000000000..b10be7a6a
--- /dev/null
+++ b/src/globus_cli/login_manager/storage.py
@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+import functools
+import os
+import sys
+import typing as t
+
+import globus_sdk
+from globus_sdk.tokenstorage import SQLiteAdapter
+
+from ._old_config import invalidate_old_config
+from .client_login import get_client_login, is_client_login
+from .scopes import CURRENT_SCOPE_CONTRACT_VERSION
+
+# env vars used throughout this module
+GLOBUS_ENV = os.environ.get("GLOBUS_SDK_ENVIRONMENT")
+
+
+class CLIStorage:
+    """
+    A wrapper over the globus-sdk's v1 tokenstorage which provides simplified
+    capabilities specific to the CLI.
+    """
+
+    def __init__(self) -> None:
+        self.adapter: SQLiteAdapter = self._construct_adapter()
+
+    def _construct_adapter(self) -> SQLiteAdapter:
+        # when initializing the token storage adapter, check if the storage file exists
+        # if it does not, then use this as a flag to clean the old config
+        fname = _get_storage_filename()
+        if not os.path.exists(fname):
+            invalidate_old_config(self.cli_native_client)
+
+        return SQLiteAdapter(fname, namespace=_resolve_namespace())
+
+    def close(self) -> None:
+        self.adapter.close()
+
+    @functools.cached_property
+    def cli_native_client(self) -> globus_sdk.NativeAppAuthClient:
+        """
+        This is the client that represents the CLI itself (prior to templating).
+        """
+        template_id = _template_client_id()
+        return globus_sdk.NativeAppAuthClient(
+            template_id, app_name="Globus CLI (native client)"
+        )
+
+    @functools.cached_property
+    def cli_confidential_client(self) -> globus_sdk.ConfidentialAppAuthClient:
+        """
+        Get the client which represents the CLI as a templated app, as distinct from a
+        confidential client built from a user's credentials.
+
+        In the event that credentials are not found, template a new client via the
+        Auth API, save the credentials for that client, and then build and return the
+        ConfidentialAppAuthClient.
+        """
+        if is_client_login():
+            raise ValueError("client logins shouldn't create internal auth clients")
+
+        client_data = self.read_well_known_config("auth_client_data")
+        if client_data is not None:
+            client_id = client_data["client_id"]
+            client_secret = client_data["client_secret"]
+        else:
+            # register a new instance client with auth
+            nc = self.cli_native_client
+            res = nc.post(
+                "/v2/api/clients",
+                data={"client": {"template_id": nc.client_id, "name": "Globus CLI"}},
+            )
+            # get values and write to config
+            credential_data = res["included"]["client_credential"]
+            client_id = credential_data["client"]
+            client_secret = credential_data["secret"]
+
+            self.store_well_known_config(
+                "auth_client_data",
+                {"client_id": client_id, "client_secret": client_secret},
+            )
+
+        return globus_sdk.ConfidentialAppAuthClient(
+            client_id, client_secret, app_name="Globus CLI"
+        )
+
+    def delete_templated_client(self) -> None:
+        # first, get the templated credentialed client
+        ac = self.cli_confidential_client
+
+        # now, remove its relevant data from storage
+        self.remove_well_known_config("auth_client_data")
+        self.remove_well_known_config("scope_contract_versions")
+
+        # finally, try to delete via the API
+        # note that this could raise an exception if the creds are already invalid --
+        # the caller may or may not want to ignore, so allow it to raise from here
+        ac.delete(f"/v2/api/clients/{ac.client_id}")
+
+        # clear the cached_property
+        del self.cli_confidential_client
+
+    def store_well_known_config(
+        self,
+        name: t.Literal[
+            "auth_client_data", "auth_user_data", "scope_contract_versions"
+        ],
+        data: dict[str, t.Any],
+    ) -> None:
+        self.adapter.store_config(name, data)
+
+    @t.overload
+    def read_well_known_config(
+        self,
+        name: t.Literal[
+            "auth_client_data", "auth_user_data", "scope_contract_versions"
+        ],
+        *,
+        allow_null: t.Literal[False],
+    ) -> dict[str, t.Any]: ...
+
+    @t.overload
+    def read_well_known_config(
+        self,
+        name: t.Literal[
+            "auth_client_data", "auth_user_data", "scope_contract_versions"
+        ],
+        *,
+        allow_null: bool = True,
+    ) -> dict[str, t.Any] | None: ...
+
+    def read_well_known_config(
+        self,
+        name: t.Literal[
+            "auth_client_data", "auth_user_data", "scope_contract_versions"
+        ],
+        *,
+        allow_null: bool = True,
+    ) -> dict[str, t.Any] | None:
+        data = self.adapter.read_config(name)
+        if not allow_null and data is None:
+            if name == "auth_user_data":
+                alias = "Identity Info"
+            else:
+                alias = name
+            raise RuntimeError(
+                f"{alias} was unexpectedly not visible in storage. "
+                "A new login should fix the issue. "
+                "Consider using `globus login --force`"
+            )
+        return data
+
+    def remove_well_known_config(
+        self,
+        name: t.Literal[
+            "auth_client_data", "auth_user_data", "scope_contract_versions"
+        ],
+    ) -> None:
+        self.adapter.remove_config(name)
+
+    def store(self, token_response: globus_sdk.OAuthTokenResponse) -> None:
+        self.adapter.store(token_response)
+        # store contract versions for all of the tokens which were acquired
+        # this could overwrite data from another CLI version *earlier or later* than
+        # the current one
+        #
+        # in the case that the old data was from a prior version, this makes sense
+        # because we have added new constraints or behaviors
+        #
+        # if the data was from a *newer* CLI version than what we are currently
+        # running we can't really know with certainty that "downgrading" the version
+        # numbers is correct, but because we can't know we need to just do our best
+        # to indicate that the tokens in storage may have lost capabilities
+        contract_versions: dict[str, t.Any] | None = self.read_well_known_config(
+            "scope_contract_versions"
+        )
+        if contract_versions is None:
+            contract_versions = {}
+        for rs_name in token_response.by_resource_server:
+            contract_versions[rs_name] = CURRENT_SCOPE_CONTRACT_VERSION
+        self.store_well_known_config("scope_contract_versions", contract_versions)
+
+
+def _template_client_id() -> str:
+    template_id = "95fdeba8-fac2-42bd-a357-e068d82ff78e"
+    if GLOBUS_ENV:
+        template_id = {
+            "sandbox": "33b6a241-bce4-4359-9c6d-09f88b3c9eef",
+            "integration": "e0c31fd1-663b-44e1-840f-f4304bb9ee7a",
+            "test": "0ebfd058-452f-40c3-babf-5a6b16a7b337",
+            "staging": "3029c3cb-c8d9-4f2b-979c-c53330aa7327",
+            "preview": "b2867dbb-0846-4579-8486-dc70763d700b",
+        }.get(GLOBUS_ENV, template_id)
+    return template_id
+
+
+def _get_data_dir() -> str:
+    # get the dir to store Globus CLI data
+    #
+    # on Windows, the datadir is typically
+    #   ~\AppData\Local\globus\cli
+    #
+    # on Linux and macOS, we use
+    #   ~/.globus/cli/
+    #
+    # This is not necessarily a match with XDG_DATA_HOME or macOS use of
+    # '~/Library/Application Support'. The simplified directories for non-Windows
+    # platforms will allow easier access to the dir if necessary in support of users
+    if sys.platform == "win32":
+        # try to get the app data dir, preferring the local appdata
+        datadir = os.getenv("LOCALAPPDATA", os.getenv("APPDATA"))
+        if not datadir:
+            home = os.path.expanduser("~")
+            datadir = os.path.join(home, "AppData", "Local")
+        return os.path.join(datadir, "globus", "cli")
+    else:
+        return os.path.expanduser("~/.globus/cli/")
+
+
+def _ensure_data_dir() -> str:
+    dirname = _get_data_dir()
+    try:
+        os.makedirs(dirname)
+    except FileExistsError:
+        pass
+    return dirname
+
+
+def _get_storage_filename() -> str:
+    datadir = _ensure_data_dir()
+    return os.path.join(datadir, "storage.db")
+
+
+def _resolve_namespace() -> str:
+    """
+    expected user namespaces are:
+
+    userprofile/production        (default)
+    userprofile/sandbox           (env is set to sandbox)
+    userprofile/test/myprofile    (env is set to test, profile is set to myprofile)
+
+    client namespaces ignore profile, and include client_id in the namespace:
+
+    clientprofile/production/926cc9c6-b481-4a5e-9ccd-b497f04c643b (default)
+    clientprofile/sandbox/926cc9c6-b481-4a5e-9ccd-b497f04c643b    (sandbox env)
+    """
+    env = GLOBUS_ENV if GLOBUS_ENV else "production"
+    profile = os.environ.get("GLOBUS_PROFILE")
+
+    if is_client_login():
+        client_id = get_client_login().client_id
+        return f"clientprofile/{env}/{client_id}"
+
+    else:
+        return "userprofile/" + env + (f"/{profile}" if profile else "")
diff --git a/src/globus_cli/login_manager/tokenstore.py b/src/globus_cli/login_manager/tokenstore.py
deleted file mode 100644
index fa4ab4482..000000000
--- a/src/globus_cli/login_manager/tokenstore.py
+++ /dev/null
@@ -1,265 +0,0 @@
-from __future__ import annotations
-
-import os
-import sys
-import typing as t
-
-import globus_sdk
-
-from .client_login import get_client_login, is_client_login
-from .scopes import CURRENT_SCOPE_CONTRACT_VERSION
-
-if t.TYPE_CHECKING:
-    from globus_sdk.tokenstorage import SQLiteAdapter
-
-# env vars used throughout this module
-GLOBUS_ENV = os.environ.get("GLOBUS_SDK_ENVIRONMENT")
-
-
-# stub to allow type casting of a function to an object with an attribute
-class _TokenStoreFuncProto:
-    _instance: SQLiteAdapter
-
-
-def _template_client_id() -> str:
-    template_id = "95fdeba8-fac2-42bd-a357-e068d82ff78e"
-    if GLOBUS_ENV:
-        template_id = {
-            "sandbox": "33b6a241-bce4-4359-9c6d-09f88b3c9eef",
-            "integration": "e0c31fd1-663b-44e1-840f-f4304bb9ee7a",
-            "test": "0ebfd058-452f-40c3-babf-5a6b16a7b337",
-            "staging": "3029c3cb-c8d9-4f2b-979c-c53330aa7327",
-            "preview": "b2867dbb-0846-4579-8486-dc70763d700b",
-        }.get(GLOBUS_ENV, template_id)
-    return template_id
-
-
-def internal_native_client() -> globus_sdk.NativeAppAuthClient:
-    """
-    This is the client that represents the CLI itself (prior to templating).
-    """
-    template_id = _template_client_id()
-    return globus_sdk.NativeAppAuthClient(
-        template_id, app_name="Globus CLI (native client)"
-    )
-
-
-def _get_data_dir() -> str:
-    # get the dir to store Globus CLI data
-    #
-    # on Windows, the datadir is typically
-    #   ~\AppData\Local\globus\cli
-    #
-    # on Linux and macOS, we use
-    #   ~/.globus/cli/
-    #
-    # This is not necessarily a match with XDG_DATA_HOME or macOS use of
-    # '~/Library/Application Support'. The simplified directories for non-Windows
-    # platforms will allow easier access to the dir if necessary in support of users
-    if sys.platform == "win32":
-        # try to get the app data dir, preferring the local appdata
-        datadir = os.getenv("LOCALAPPDATA", os.getenv("APPDATA"))
-        if not datadir:
-            home = os.path.expanduser("~")
-            datadir = os.path.join(home, "AppData", "Local")
-        return os.path.join(datadir, "globus", "cli")
-    else:
-        return os.path.expanduser("~/.globus/cli/")
-
-
-def _ensure_data_dir() -> str:
-    dirname = _get_data_dir()
-    try:
-        os.makedirs(dirname)
-    except FileExistsError:
-        pass
-    return dirname
-
-
-def _get_storage_filename() -> str:
-    datadir = _ensure_data_dir()
-    return os.path.join(datadir, "storage.db")
-
-
-def _resolve_namespace() -> str:
-    """
-    expected user namespaces are:
-
-    userprofile/production        (default)
-    userprofile/sandbox           (env is set to sandbox)
-    userprofile/test/myprofile    (env is set to test, profile is set to myprofile)
-
-    client namespaces ignore profile, and include client_id in the namespace:
-
-    clientprofile/production/926cc9c6-b481-4a5e-9ccd-b497f04c643b (default)
-    clientprofile/sandbox/926cc9c6-b481-4a5e-9ccd-b497f04c643b    (sandbox env)
-    """
-    env = GLOBUS_ENV if GLOBUS_ENV else "production"
-    profile = os.environ.get("GLOBUS_PROFILE")
-
-    if is_client_login():
-        client_id = get_client_login().client_id
-        return f"clientprofile/{env}/{client_id}"
-
-    else:
-        return "userprofile/" + env + (f"/{profile}" if profile else "")
-
-
-def build_storage_adapter(fname: str) -> SQLiteAdapter:
-    """
-    Customize the SQLiteAdapter with extra storage operation steps
-    In order to avoid eager imports, which have a perf impact on the CLI, we need to
-    define the class dynamically in this function.
-    """
-    from globus_sdk.tokenstorage import SQLiteAdapter
-
-    class GeneratedAdapterClass(SQLiteAdapter):
-        def store(self, token_response: globus_sdk.OAuthTokenResponse) -> None:
-            super().store(token_response)
-            # store contract versions for all of the tokens which were acquired
-            # this could overwrite data from another CLI version *earlier or later* than
-            # the current one
-            #
-            # in the case that the old data was from a prior version, this makes sense
-            # because we have added new constraints or behaviors
-            #
-            # if the data was from a *newer* CLI version than what we are currently
-            # running we can't really know with certainty that "downgrading" the version
-            # numbers is correct, but because we can't know we need to just do our best
-            # to indicate that the tokens in storage may have lost capabilities
-            contract_versions: dict[str, t.Any] | None = read_well_known_config(
-                "scope_contract_versions", adapter=self
-            )
-            if contract_versions is None:
-                contract_versions = {}
-            for rs_name in token_response.by_resource_server:
-                contract_versions[rs_name] = CURRENT_SCOPE_CONTRACT_VERSION
-            store_well_known_config(
-                "scope_contract_versions", contract_versions, adapter=self
-            )
-
-    return GeneratedAdapterClass(fname, namespace=_resolve_namespace())
-
-
-def token_storage_adapter() -> SQLiteAdapter:
-    as_proto = t.cast(_TokenStoreFuncProto, token_storage_adapter)
-    if not hasattr(as_proto, "_instance"):
-        # when initializing the token storage adapter, check if the storage file exists
-        # if it does not, then use this as a flag to clean the old config
-        fname = _get_storage_filename()
-        if not os.path.exists(fname):
-            from ._old_config import invalidate_old_config
-
-            invalidate_old_config(internal_native_client())
-        # namespace is equal to the current environment
-        as_proto._instance = build_storage_adapter(fname)
-    return as_proto._instance
-
-
-def internal_auth_client() -> globus_sdk.ConfidentialAppAuthClient:
-    """
-    Pull template client credentials from storage and use them to create a
-    ConfidentialAppAuthClient.
-    In the event that credentials are not found, template a new client via the Auth API,
-    save the credentials for that client, and then build and return the
-    ConfidentialAppAuthClient.
-    """
-    if is_client_login():
-        raise ValueError("client logins shouldn't create internal auth clients")
-
-    client_data = read_well_known_config("auth_client_data")
-    if client_data is not None:
-        client_id = client_data["client_id"]
-        client_secret = client_data["client_secret"]
-    else:
-        # register a new instance client with auth
-        nc = internal_native_client()
-        res = nc.post(
-            "/v2/api/clients",
-            data={"client": {"template_id": nc.client_id, "name": "Globus CLI"}},
-        )
-        # get values and write to config
-        credential_data = res["included"]["client_credential"]
-        client_id = credential_data["client"]
-        client_secret = credential_data["secret"]
-
-        store_well_known_config(
-            "auth_client_data",
-            {"client_id": client_id, "client_secret": client_secret},
-        )
-
-    return globus_sdk.ConfidentialAppAuthClient(
-        client_id, client_secret, app_name="Globus CLI"
-    )
-
-
-def delete_templated_client() -> None:
-    # first, get the templated credentialed client
-    ac = internal_auth_client()
-
-    # now, remove its relevant data from storage
-    remove_well_known_config("auth_client_data")
-    remove_well_known_config("scope_contract_versions")
-
-    # finally, try to delete via the API
-    # note that this could raise an exception if the creds are already invalid -- the
-    # caller may or may not want to ignore, so allow it to raise from here
-    ac.delete(f"/v2/api/clients/{ac.client_id}")
-
-
-def store_well_known_config(
-    name: t.Literal["auth_client_data", "auth_user_data", "scope_contract_versions"],
-    data: dict[str, t.Any],
-    *,
-    adapter: SQLiteAdapter | None = None,
-) -> None:
-    adapter = adapter or token_storage_adapter()
-    adapter.store_config(name, data)
-
-
-@t.overload
-def read_well_known_config(
-    name: t.Literal["auth_client_data", "auth_user_data", "scope_contract_versions"],
-    *,
-    adapter: SQLiteAdapter | None = None,
-    allow_null: t.Literal[False],
-) -> dict[str, t.Any]: ...
-
-
-@t.overload
-def read_well_known_config(
-    name: t.Literal["auth_client_data", "auth_user_data", "scope_contract_versions"],
-    *,
-    adapter: SQLiteAdapter | None = None,
-    allow_null: bool = True,
-) -> dict[str, t.Any] | None: ...
-
-
-def read_well_known_config(
-    name: t.Literal["auth_client_data", "auth_user_data", "scope_contract_versions"],
-    *,
-    adapter: SQLiteAdapter | None = None,
-    allow_null: bool = True,
-) -> dict[str, t.Any] | None:
-    adapter = adapter or token_storage_adapter()
-    data = adapter.read_config(name)
-    if not allow_null and data is None:
-        if name == "auth_user_data":
-            alias = "Identity Info"
-        else:
-            alias = name
-        raise RuntimeError(
-            f"{alias} was unexpectedly not visible in storage. "
-            "A new login should fix the issue. "
-            "Consider using `globus login --force`"
-        )
-    return data
-
-
-def remove_well_known_config(
-    name: t.Literal["auth_client_data", "auth_user_data", "scope_contract_versions"],
-    *,
-    adapter: SQLiteAdapter | None = None,
-) -> None:
-    adapter = adapter or token_storage_adapter()
-    adapter.remove_config(name)
diff --git a/src/globus_cli/login_manager/utils.py b/src/globus_cli/login_manager/utils.py
index 0b0bea82c..64ace8d87 100644
--- a/src/globus_cli/login_manager/utils.py
+++ b/src/globus_cli/login_manager/utils.py
@@ -1,22 +1,5 @@
 import os
-import typing as t
-
-from globus_cli.login_manager.client_login import is_client_login
-from globus_cli.login_manager.tokenstore import read_well_known_config
 
 
 def is_remote_session() -> bool:
     return bool(os.environ.get("SSH_TTY", os.environ.get("SSH_CONNECTION")))
-
-
-def get_current_identity_id() -> str:
-    """
-    Return the current user's identity ID.
-    For a client-authorized invocation, that's the client ID.
-    """
-
-    if is_client_login():
-        return os.environ["GLOBUS_CLI_CLIENT_ID"]
-    else:
-        user_data = read_well_known_config("auth_user_data", allow_null=False)
-        return t.cast(str, user_data["sub"])
diff --git a/tests/conftest.py b/tests/conftest.py
index 2adf04bda..625ed7251 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -9,17 +9,19 @@
 import uuid
 from unittest import mock
 
+import click
 import globus_sdk
 import pytest
 import responses
 from click.testing import CliRunner
 from globus_sdk._testing import register_response_set
 from globus_sdk.scopes import TimersScopes
+from globus_sdk.tokenstorage import SQLiteAdapter
 from globus_sdk.transport import RequestsTransport
 from ruamel.yaml import YAML
 
 import globus_cli
-from globus_cli.login_manager.tokenstore import build_storage_adapter
+from globus_cli.login_manager.scopes import CURRENT_SCOPE_CONTRACT_VERSION
 
 yaml = YAML()
 log = logging.getLogger(__name__)
@@ -37,6 +39,14 @@ def pytest_configure(config):
     globus_cli._warnings._TEST_WARNING_CONTROL = True
 
 
+@pytest.fixture
+def test_click_context():
+    from globus_cli.commands import main
+
+    with click.Context(main):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def mocksleep():
     with mock.patch("time.sleep") as m:
@@ -153,23 +163,32 @@ def mock_user_data():
 @pytest.fixture
 def test_token_storage(mock_login_token_response, mock_user_data):
     """Put memory-backed sqlite token storage in place for the testsuite to use."""
-    mockstore = build_storage_adapter(":memory:")
+    mockstore = SQLiteAdapter(":memory:")
+    real_close = mockstore.close
+    mockstore.close = mock.Mock()
     mockstore.store_config(
         "auth_client_data",
         {"client_id": "fakeClientIDString", "client_secret": "fakeClientSecret"},
     )
     mockstore.store_config("auth_user_data", mock_user_data)
     mockstore.store(mock_login_token_response)
-    return mockstore
+    mockstore.store_config(
+        "scope_contract_versions",
+        {
+            k: CURRENT_SCOPE_CONTRACT_VERSION
+            for k in mock_login_token_response.by_resource_server
+        },
+    )
+    yield mockstore
+    real_close()
 
 
 @pytest.fixture(autouse=True)
 def patch_tokenstorage(monkeypatch, test_token_storage):
     monkeypatch.setattr(
-        globus_cli.login_manager.token_storage_adapter,
-        "_instance",
-        test_token_storage,
-        raising=False,
+        globus_cli.login_manager.storage.CLIStorage,
+        "_construct_adapter",
+        lambda self: test_token_storage,
     )
 
 
diff --git a/tests/functional/test_login_command.py b/tests/functional/test_login_command.py
index 204187dde..d66ae43fd 100644
--- a/tests/functional/test_login_command.py
+++ b/tests/functional/test_login_command.py
@@ -5,11 +5,7 @@
 import pytest
 from globus_sdk._testing import load_response_set
 
-from globus_cli.login_manager import (
-    LoginManager,
-    read_well_known_config,
-    store_well_known_config,
-)
+from globus_cli.login_manager import LoginManager
 from globus_cli.login_manager.auth_flows import exchange_code_and_store
 from tests.conftest import _mock_token_response_data
 
@@ -20,10 +16,10 @@ def test_login_validates_token(
     # undo the validate_token disabling patch which is done for most tests
     disable_login_manager_validate_token.undo()
 
-    with mock.patch("globus_cli.login_manager.manager.internal_auth_client") as m:
-        ac = mock.MagicMock(spec=globus_sdk.ConfidentialAppAuthClient)
-        m.return_value = ac
-
+    ac = mock.MagicMock(spec=globus_sdk.ConfidentialAppAuthClient)
+    with mock.patch(
+        "globus_cli.login_manager.storage.CLIStorage.cli_confidential_client", ac
+    ):
         run_line("globus login")
 
         by_rs = mock_login_token_response.by_resource_server
@@ -63,7 +59,7 @@ def test_login_gcs_different_identity(
     mock_remote_session,
     mock_local_server_flow,
     mock_login_token_response,
-    test_token_storage,
+    test_click_context,
 ):
     """
     Test the `exchange_code_and_store` behavior where logging in with a different
@@ -71,13 +67,16 @@ def test_login_gcs_different_identity(
     remove the `sub` in config storage (which is what originally raises that error).
     """
     load_response_set("cli.logout")
-    store_well_known_config(
-        "auth_user_data", {"sub": str(uuid.UUID(int=0))}, adapter=test_token_storage
+    manager = LoginManager()
+    manager.storage.store_well_known_config(
+        "auth_user_data", {"sub": str(uuid.UUID(int=0))}
     )
     mock_auth_client = mock.MagicMock(spec=globus_sdk.NativeAppAuthClient)
     mock_auth_client.oauth2_exchange_code_for_tokens = lambda _: MockToken()
     mock_local_server_flow.side_effect = (
-        lambda *args, **kwargs: exchange_code_and_store(mock_auth_client, "bogus_code")
+        lambda *args, **kwargs: exchange_code_and_store(
+            manager.storage, mock_auth_client, "bogus_code"
+        )
     )
     mock_remote_session.return_value = False
     result = run_line(f"globus login --gcs {uuid.UUID(int=0)}", assert_exit_code=1)
@@ -92,10 +91,11 @@ def test_login_gcs_different_identity(
     )
 
     monkeypatch.setattr(
-        "globus_cli.commands.logout.internal_native_client", lambda: mock_auth_client
+        "globus_cli.login_manager.storage.CLIStorage.cli_confidential_client",
+        mock_auth_client,
     )
     run_line("globus logout --yes")
-    assert read_well_known_config("auth_user_data", adapter=test_token_storage) is None
+    assert manager.storage.read_well_known_config("auth_user_data") is None
 
 
 def test_login_with_flow(monkeypatch, run_line):
diff --git a/tests/functional/test_logout_command.py b/tests/functional/test_logout_command.py
index 516e29c39..1952940c0 100644
--- a/tests/functional/test_logout_command.py
+++ b/tests/functional/test_logout_command.py
@@ -4,22 +4,23 @@
 import responses
 from globus_sdk._testing import load_response_set
 
-from globus_cli.login_manager import read_well_known_config
+from globus_cli.login_manager import LoginManager
 
 
 @pytest.mark.parametrize("delete_client", [True, False])
-def test_logout(delete_client, run_line, test_token_storage, mock_login_token_response):
+def test_logout(delete_client, run_line, mock_login_token_response, test_click_context):
     load_response_set("cli.logout")
+    manager = LoginManager()
 
     # Collect all of the stored tokens
     stored_tokens = set()
-    for token_data in test_token_storage.get_by_resource_server().values():
+    for token_data in manager.storage.adapter.get_by_resource_server().values():
         stored_tokens.add(token_data["access_token"])
         stored_tokens.add(token_data["refresh_token"])
 
     assert len(stored_tokens) > 0
 
-    ac_data = read_well_known_config("auth_client_data", adapter=test_token_storage)
+    ac_data = manager.storage.read_well_known_config("auth_client_data")
     client_id = ac_data["client_id"]
 
     additional_args = ["--delete-client"] if delete_client else []
@@ -57,18 +58,19 @@ def test_logout(delete_client, run_line, test_token_storage, mock_login_token_re
 
     assert "You are now successfully logged out" in result.output
     # Make sure the storage was cleared out
-    assert read_well_known_config("auth_user_data", adapter=test_token_storage) is None
+    assert manager.storage.read_well_known_config("auth_user_data") is None
 
 
 @pytest.mark.parametrize("delete_client", [True, False])
 def test_logout_with_client_id(
-    delete_client, run_line, test_token_storage, mock_login_token_response, client_login
+    delete_client, run_line, mock_login_token_response, client_login, test_click_context
 ):
     load_response_set("cli.logout")
+    manager = LoginManager()
 
     # Collect all of the stored tokens
     stored_tokens = set()
-    for token_data in test_token_storage.get_by_resource_server().values():
+    for token_data in manager.storage.adapter.get_by_resource_server().values():
         stored_tokens.add(token_data["access_token"])
         stored_tokens.add(token_data["refresh_token"])
 
@@ -96,4 +98,4 @@ def test_logout_with_client_id(
 
     assert "Revoking all CLI tokens for" in result.output
     # Make sure the storage was cleared out
-    assert read_well_known_config("auth_user_data", adapter=test_token_storage) is None
+    assert manager.storage.read_well_known_config("auth_user_data") is None
diff --git a/tests/unit/test_login_manager.py b/tests/unit/test_login_manager.py
index e9b76332f..3362046f9 100644
--- a/tests/unit/test_login_manager.py
+++ b/tests/unit/test_login_manager.py
@@ -25,8 +25,8 @@
 
 
 @pytest.fixture
-def patched_tokenstorage():
-    def mock_get_tokens(resource_server):
+def patched_tokenstorage(test_token_storage):
+    def fake_get_tokens(resource_server):
         fake_tokens = {
             "a.globus.org": {
                 "access_token": "fake_a_access_token",
@@ -42,7 +42,7 @@ def mock_get_tokens(resource_server):
 
         return fake_tokens.get(resource_server)
 
-    def mock_read_config(config_name):
+    def fake_read_config(self, config_name):
         if config_name == "scope_contract_versions":
             return {
                 "a.globus.org": 1,
@@ -51,12 +51,14 @@ def mock_read_config(config_name):
         else:
             raise NotImplementedError
 
+    test_token_storage.get_token_data = mock.Mock()
+    test_token_storage.get_token_data.side_effect = fake_get_tokens
+
     with mock.patch(
-        "globus_cli.login_manager.tokenstore.token_storage_adapter._instance"
-    ) as mock_adapter:
-        mock_adapter.get_token_data = mock_get_tokens
-        mock_adapter.read_config = mock_read_config
-        yield mock_adapter
+        "globus_cli.login_manager.storage.CLIStorage.read_well_known_config",
+        fake_read_config,
+    ):
+        yield
 
 
 @pytest.fixture
@@ -103,7 +105,9 @@ def urlfmt_scope(rs: str, name: str) -> str:
 BASE_TIMER_SCOPE = urlfmt_scope("524230d7-ea86-4a52-8312-86065a9e0417", "timer")
 
 
-def test_requires_login_success(patch_scope_requirements, patched_tokenstorage):
+def test_requires_login_success(
+    patch_scope_requirements, patched_tokenstorage, test_click_context
+):
     # single server
     @LoginManager.requires_login("a")
     def dummy_command(login_manager):
@@ -113,7 +117,7 @@ def dummy_command(login_manager):
 
 
 def test_requires_login_multi_server_success(
-    patch_scope_requirements, patched_tokenstorage
+    patch_scope_requirements, patched_tokenstorage, test_click_context
 ):
     @LoginManager.requires_login("a", "b")
     def dummy_command(login_manager):
@@ -123,7 +127,7 @@ def dummy_command(login_manager):
 
 
 def test_requires_login_single_server_fail(
-    patch_scope_requirements, patched_tokenstorage
+    patch_scope_requirements, patched_tokenstorage, test_click_context
 ):
     @LoginManager.requires_login("c.globus.org")
     def dummy_command(login_manager):
@@ -139,6 +143,7 @@ def dummy_command(login_manager):
 
 def test_requiring_login_for_multiple_known_servers_renders_nice_error(
     patch_scope_requirements,
+    test_click_context,
 ):
     @LoginManager.requires_login("a", "b")
     def dummy_command(login_manager):
@@ -153,7 +158,9 @@ def dummy_command(login_manager):
     )
 
 
-def test_requiring_new_scope_fails(patch_scope_requirements, patched_tokenstorage):
+def test_requiring_new_scope_fails(
+    patch_scope_requirements, patched_tokenstorage, test_click_context
+):
     CLI_SCOPE_REQUIREMENTS["a"]["scopes"].append("scopeA3")
 
     @LoginManager.requires_login("a")
@@ -168,7 +175,9 @@ def dummy_command(login_manager):
     )
 
 
-def test_scope_contract_version_bump_forces_login(patch_scope_requirements):
+def test_scope_contract_version_bump_forces_login(
+    patch_scope_requirements, test_click_context
+):
     CLI_SCOPE_REQUIREMENTS["a"]["min_contract_version"] = 2
 
     @LoginManager.requires_login("a")
@@ -184,7 +193,7 @@ def dummy_command(login_manager):
 
 
 def test_requires_login_fail_two_servers(
-    patch_scope_requirements, patched_tokenstorage
+    patch_scope_requirements, patched_tokenstorage, test_click_context
 ):
     @LoginManager.requires_login("c.globus.org", "d.globus.org")
     def dummy_command(login_manager):
@@ -203,7 +212,7 @@ def dummy_command(login_manager):
 
 
 def test_requires_login_fail_multi_server(
-    patch_scope_requirements, patched_tokenstorage
+    patch_scope_requirements, patched_tokenstorage, test_click_context
 ):
     @LoginManager.requires_login("c.globus.org", "d.globus.org", "e.globus.org")
     def dummy_command(login_manager):
@@ -220,7 +229,9 @@ def dummy_command(login_manager):
         assert server in str(ex.value)
 
 
-def test_requires_login_pass_manager(patch_scope_requirements, patched_tokenstorage):
+def test_requires_login_pass_manager(
+    patch_scope_requirements, patched_tokenstorage, test_click_context
+):
     @LoginManager.requires_login()
     def dummy_command(login_manager):
         assert login_manager.has_login("a.globus.org")
@@ -231,7 +242,9 @@ def dummy_command(login_manager):
     assert dummy_command()
 
 
-def test_login_manager_respects_context_error_message(patched_tokenstorage):
+def test_login_manager_respects_context_error_message(
+    patched_tokenstorage, test_click_context
+):
     dummy_id = str(uuid.uuid1())
 
     @LoginManager.requires_login()
@@ -249,7 +262,7 @@ def dummy_command(login_manager):
     assert expected == str(excinfo.value)
 
 
-def test_client_login_two_requirements(client_login):
+def test_client_login_two_requirements(client_login, test_click_context):
     @LoginManager.requires_login("auth", "transfer")
     def dummy_command(login_manager):
         transfer_client = login_manager.get_transfer_client()
@@ -267,7 +280,7 @@ def dummy_command(login_manager):
     assert dummy_command()
 
 
-def test_client_login_gcs(client_login, add_gcs_login):
+def test_client_login_gcs(client_login, add_gcs_login, test_click_context):
     with mock.patch.object(LoginManager, "_get_gcs_info") as mock_get_gcs_info:
 
         class fake_endpointish:
@@ -345,7 +358,7 @@ def test_cli_scope_requirements_min_contract_version_matches_current():
 def test_immature_signature_during_jwt_decode_emits_clock_skew_notice(
     capsys,
     monkeypatch,
-    test_token_storage,
+    test_click_context,
 ):
     """
     Test the `exchange_code_and_store` behavior when the id_token decoding fails
@@ -353,6 +366,8 @@ def test_immature_signature_during_jwt_decode_emits_clock_skew_notice(
 
     This should result in a clear error emitted to stderr.
     """
+    manager = LoginManager()
+
     mock_token_response = mock.Mock()
     mock_token_response.decode_id_token = mock.Mock(
         side_effect=jwt.exceptions.ImmatureSignatureError("test")
@@ -370,7 +385,7 @@ def test_immature_signature_during_jwt_decode_emits_clock_skew_notice(
     )
 
     with pytest.raises(jwt.exceptions.ImmatureSignatureError):
-        exchange_code_and_store(mock_auth_client, "bogus_code")
+        exchange_code_and_store(manager.storage, mock_auth_client, "bogus_code")
 
     stderr = capsys.readouterr().err
     assert "out of sync with the local clock" in stderr
@@ -390,6 +405,8 @@ def test_immature_signature_during_jwt_decode_skips_notice_if_date_cannot_parse(
 
     This should result in a clear error emitted to stderr.
     """
+    manager = LoginManager()
+
     mock_token_response = mock.Mock()
     mock_token_response.decode_id_token = mock.Mock(
         side_effect=jwt.exceptions.ImmatureSignatureError("test")
@@ -409,7 +426,7 @@ def test_immature_signature_during_jwt_decode_skips_notice_if_date_cannot_parse(
     )
 
     with pytest.raises(jwt.exceptions.ImmatureSignatureError):
-        exchange_code_and_store(mock_auth_client, "bogus_code")
+        exchange_code_and_store(manager.storage, mock_auth_client, "bogus_code")
 
     stderr = capsys.readouterr().err
     assert "This may indicate a clock skew problem." not in stderr
diff --git a/tests/unit/test_tokenstore.py b/tests/unit/test_tokenstore.py
index d148d6073..d369233b4 100644
--- a/tests/unit/test_tokenstore.py
+++ b/tests/unit/test_tokenstore.py
@@ -1,4 +1,9 @@
-from globus_cli.login_manager.tokenstore import _resolve_namespace
+import uuid
+
+import globus_sdk
+from globus_sdk._testing import RegisteredResponse, get_last_request
+
+from globus_cli.login_manager.storage import CLIStorage, _resolve_namespace
 
 
 def test_default_namespace():
@@ -11,3 +16,66 @@ def test_profile_namespace(user_profile):
 
 def test_client_namespace(client_login):
     assert _resolve_namespace() == "clientprofile/production/fake_client_id"
+
+
+def test_storage_reuses_clients():
+    # confirm that we do not recreate clients when we need to use them multiple times
+    #
+    # these are instrumented via `@functools.cached_property`
+    # if we switch to `@property` we must choose an implementation which
+    # does not regress this test
+    storage = CLIStorage()
+
+    confidential1 = storage.cli_confidential_client
+    confidential2 = storage.cli_confidential_client
+    assert confidential1 is confidential2
+
+    native1 = storage.cli_native_client
+    native2 = storage.cli_native_client
+    assert native1 is native2
+
+
+def test_storage_creates_fresh_client_after_delete():
+    RegisteredResponse(
+        service="auth",
+        path="/v2/api/clients/fakeClientIDString",
+        method="DELETE",
+        status=200,
+        json={},
+    ).add()
+    RegisteredResponse(
+        service="auth",
+        path="/v2/api/clients",
+        method="POST",
+        status=200,
+        json={
+            "included": {
+                "client_credential": {
+                    "client": str(uuid.UUID(int=1)),
+                    "secret": "bogusBOGUSbogus",
+                }
+            }
+        },
+    ).add()
+    # note: the `patch_tokenstorage` fixture in top-level conftest.py ensures
+    # that we get a memory backed DB with pre-existing client credentials
+    storage = CLIStorage()
+
+    # before delete: we have a client and data for that client
+    client1 = storage.cli_confidential_client
+    assert isinstance(client1, globus_sdk.ConfidentialAppAuthClient)
+    client_data1 = storage.read_well_known_config("auth_client_data")
+    assert client_data1 is not None
+
+    storage.delete_templated_client()
+
+    # after delete: no client data, fetching a client creates a new one
+    client_data2 = storage.read_well_known_config("auth_client_data")
+    assert client_data2 is None
+
+    client2 = storage.cli_confidential_client
+    assert client1 is not client2
+
+    last_req = get_last_request()
+    assert last_req.method == "POST"
+    assert last_req.url.endswith("/v2/api/clients")