Allow client settings to configure application-owned token managers

README.md

@@ -285,3 +285,34 @@ normal field validation, including validation during request-mode instance reval
 is a typed, copyable, picklable singleton; it is omitted from JSON Schema and serialization by the
 field configuration above. A literal string `"UNSET"` remains an ordinary string value. Keep the
 `default_factory` and `exclude_if` configuration when defining patch models manually.
+
+## Application-owned token persistence
+
+```python
+settings = ClientSettings.resolve(
+    base_url, alternatives, credentials,
+    token_manager_factory=lambda providers: TokenManager(
+        providers, store=application_store, expiry_skew=30, now=application_clock,
+    ),
+)
+```
+
+The same optional `token_manager_factory: TokenManagerFactory | None` keyword is available on the
+settings initializer. Clock values and expiry skew are seconds. Use the manager on its application's
+asyncio loop. The native manager has no close/reset method: cancel/await application requests and
+discard the manager at session end; it does not own HTTPX clients or close application storage.
+
+The hook is invoked once with the resolved provider map, after security validation, and is skipped
+when no providers are selected. It must only construct a manager: do not acquire tokens or read
+storage in the hook. Omitting it keeps the existing in-memory default. Settings retain the returned
+manager, not the factory. All operations on those settings share it; generated aggregate children
+therefore retain the same cache and single-flight renewal. Independently created managers do not
+coordinate concurrent refreshes, even if their stores are the same. Reuse a client/aggregate within
+an active session; use successive managers to reopen saved sessions.
+
+The application owns persistence, encryption, store access and session boundaries. Provider/client,
+grant, profile and endpoint identities must distinguish environments and users; the API base URL
+alone is not an implicit store namespace. Use a new grant identity for a fresh authorization session.
+For logout, stop requests and wait for pending refresh/persistence to finish before removing the
+session's store entries, then construct fresh settings. `invalidate` expires an access token for
+renewal; it is not logout and deliberately retains refresh state. No disk storage is enabled automatically.

src/sunday/__init__.py

@@ -72,6 +72,7 @@
 from .sse import EventParser, EventStreamOptions, ServerSentEvent
 from .streaming import StreamingBody, StreamingBodyChunk, StreamingBodyClose, StreamingBodyContent
 from .token_manager import TokenLease, TokenManager
+from .token_manager_factory import TokenManagerFactory
 from .token_provider import (
     AuthorizationRequiredError,
     RefreshingTokenProvider,
@@ -168,6 +169,7 @@
     "TokenConfiguration",
     "TokenLease",
     "TokenManager",
+    "TokenManagerFactory",
     "TokenProvider",
     "TokenProviderError",
     "TokenRequest",

src/sunday/client_settings.py

@@ -26,6 +26,7 @@
 )
 from .security import SecurityBinding, SecurityEndpoints
 from .token_manager import TokenManager
+from .token_manager_factory import TokenManagerFactory
 from .token_provider import TokenConfiguration, TokenProvider, TokenRequest, TokenSet
 
 
@@ -43,6 +44,8 @@ def __init__(
         base_url: str,
         bindings: Mapping[str, Sequence[SecurityBinding]] | None = None,
         credentials: Mapping[str, Credentials] | None = None,
+        *,
+        token_manager_factory: TokenManagerFactory | None = None,
     ) -> None:
         endpoint = urlsplit(base_url)
         if (
@@ -77,7 +80,7 @@ def __init__(
         object.__setattr__(self, "base_url", base_url)
         object.__setattr__(self, "bindings", MappingProxyType(values))
         object.__setattr__(self, "_credentials", MappingProxyType(supplied))
-        object.__setattr__(self, "token_manager", self._prepare_token_manager())
+        object.__setattr__(self, "token_manager", self._prepare_token_manager(token_manager_factory))
 
     @staticmethod
     def server_url(template: str, variables: Mapping[str, str], document_base_url: str | None = None) -> str:
@@ -104,6 +107,8 @@ def resolve(
         credentials: Mapping[str, Credentials],
         selection: Mapping[str, Sequence[str]] | None = None,
         alternative_selection: Mapping[str, int] | None = None,
+        *,
+        token_manager_factory: TokenManagerFactory | None = None,
     ) -> ClientSettings:
         """Choose complete alternatives; alternative_selection selects a zero-based candidate including scopes."""
         if (set(selection or {}) | set(alternative_selection or {})) - alternatives.keys():
@@ -127,9 +132,9 @@ def resolve(
             if len(usable) != 1:
                 raise ValueError(f"Operation '{operation}' requires one complete security alternative")
             bindings[operation] = usable[0]
-        return cls(base_url, bindings, credentials)
+        return cls(base_url, bindings, credentials, token_manager_factory=token_manager_factory)
 
-    def _prepare_token_manager(self) -> TokenManager | None:
+    def _prepare_token_manager(self, factory: TokenManagerFactory | None) -> TokenManager | None:
         """Prepare providers through the selected transport module without acquiring tokens."""
         owners: dict[str, str] = {}
         providers: dict[str, TokenProvider] = {}
@@ -161,7 +166,9 @@ def _prepare_token_manager(self) -> TokenManager | None:
                     providers[binding.provider] = credential.provider_factory(resolved)
                 else:
                     providers[binding.provider] = _StaticProvider(credential)
-        return TokenManager(providers) if providers else None
+        if not providers:
+            return None
+        return factory(MappingProxyType(providers)) if factory is not None else TokenManager(providers)
 
 
 class _StaticProvider:

src/sunday/token_manager_factory.py

@@ -0,0 +1,19 @@
+# Copyright 2026 Outfox, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+"""Application hook for the manager of resolved providers.
+
+Factories must not acquire tokens or perform storage I/O. They may supply the native
+manager's store, expiry_skew and now options. Called once for secured settings and
+never for settings without selected providers. The application owns the manager,
+store and session lifecycle on its asyncio event loop.
+"""
+
+from collections.abc import Callable, Mapping
+
+from .token_manager import TokenManager
+from .token_provider import TokenProvider
+
+type TokenManagerFactory = Callable[[Mapping[str, TokenProvider]], TokenManager]

tests/test_client_settings_persistence.py

@@ -0,0 +1,142 @@
+"""Application-owned manager storage survives settings instances without eager I/O."""
+
+import asyncio
+from collections.abc import Mapping
+from dataclasses import replace
+
+import pytest
+
+from sunday import (
+    BearerCredentials,
+    ClientSettings,
+    ProviderCredentials,
+    SecurityBinding,
+    SecurityTransport,
+    TokenConfiguration,
+    TokenManager,
+    TokenProvider,
+    TokenRequest,
+    TokenSet,
+)
+
+BINDING = SecurityBinding(
+    scheme="identity",
+    provider="resolved-provider",
+    flow="authorizationCode",
+    profile="development",
+    transport=SecurityTransport(location="header", name="Authorization", prefix="Bearer"),
+)
+
+
+def test_persisted_sessions_rotation_isolation_and_logout() -> None:
+    async def run() -> None:
+        values: dict[str, TokenSet] = {}
+        reads = saves = acquisitions = factories = 0
+        refreshes: list[str] = []
+
+        class Store:
+            async def load(self, key: str) -> TokenSet | None:
+                nonlocal reads
+                reads += 1
+                return values.get(key)
+
+            async def save(self, key: str, tokens: TokenSet) -> None:
+                nonlocal saves
+                saves += 1
+                values[key] = tokens
+
+            async def remove(self, key: str) -> None:
+                values.pop(key, None)
+
+        store = Store()
+
+        def settings(
+            time: float, session: str = "session", profile: str = "development", *, direct: bool = False
+        ) -> ClientSettings:
+            class Provider:
+                identity = "application"
+
+                def configure(self, binding: SecurityBinding) -> TokenConfiguration:
+                    return TokenConfiguration("client", session)
+
+                async def acquire(self, request: TokenRequest) -> TokenSet:
+                    nonlocal acquisitions
+                    acquisitions += 1
+                    return TokenSet("initial", expires_at=100, refresh_token="refresh-1")
+
+                async def refresh(self, request: TokenRequest, refresh_token: str) -> TokenSet:
+                    refreshes.append(refresh_token)
+                    return TokenSet(
+                        f"rotated-{len(refreshes)}",
+                        expires_at=time + 100,
+                        refresh_token=f"refresh-{len(refreshes) + 1}",
+                    )
+
+            provider = Provider()
+
+            def factory(providers: Mapping[str, TokenProvider]) -> TokenManager:
+                nonlocal factories
+                factories += 1
+                assert dict(providers) == {"resolved-provider": provider}
+                return TokenManager(providers, store=store, expiry_skew=5, now=lambda: time)
+
+            selected = replace(BINDING, profile=profile)
+            credentials = {"identity": ProviderCredentials(provider)}
+            if direct:
+                return ClientSettings(
+                    "https://api.example", {"read": [selected]}, credentials, token_manager_factory=factory
+                )
+            return ClientSettings.resolve(
+                "https://api.example",
+                {"read": [[selected]], "other": [[selected]], "public": [[]]},
+                credentials,
+                token_manager_factory=factory,
+            )
+
+        async def token(settings: ClientSettings) -> TokenSet:
+            assert settings.token_manager is not None
+            return (await settings.token_manager.credentials(settings.bindings["read"][0])).tokens
+
+        first = settings(0)
+        assert (reads, saves, acquisitions, factories) == (0, 0, 0, 1)
+        assert first.token_manager is not None
+        lease = await first.token_manager.credentials(first.bindings["read"][0])
+        assert lease.tokens.access_token == "initial"
+        assert (await token(settings(0, direct=True))).access_token == "initial"
+        assert acquisitions == 1
+        third = settings(96)
+        tokens = await asyncio.gather(*(token(third) for _ in range(20)))
+        assert all(value.access_token == "rotated-1" for value in tokens)
+        assert refreshes == ["refresh-1"]
+        assert (await token(settings(96))).refresh_token == "refresh-2"
+        await token(settings(192))
+        assert refreshes == ["refresh-1", "refresh-2"]
+        assert saves == 3
+        await token(settings(0, "other-session"))
+        await token(settings(0, profile="production"))
+        assert acquisitions == 3
+        await store.remove(lease.key)
+        await token(settings(0))
+        assert acquisitions == 4
+
+    asyncio.run(run())
+
+
+def test_public_invalid_and_failing_factory() -> None:
+    def factory(providers: Mapping[str, TokenProvider]) -> TokenManager:
+        raise RuntimeError("application factory failed")
+
+    assert ClientSettings("https://api.example", token_manager_factory=factory).token_manager is None
+    assert (
+        ClientSettings.resolve("https://api.example", {"public": [[]]}, {}, token_manager_factory=factory).token_manager
+        is None
+    )
+    with pytest.raises(ValueError, match="Missing credentials"):
+        ClientSettings("https://api.example", {"read": [BINDING]}, token_manager_factory=factory)
+    with pytest.raises(RuntimeError, match="application factory failed"):
+        ClientSettings(
+            "https://api.example",
+            {"read": [replace(BINDING, flow="static")]},
+            {"identity": BearerCredentials("secret")},
+            token_manager_factory=factory,
+        )