refactor(provider): unify OAuth persistence and resource cleanup

Replace four credential-write paths with one fail-closed persistence method.
Clear the in-memory token before awaiting storage so failed or cancelled
writes cannot reuse an old account or an already-rotated refresh credential.

Reuse terminate() for partial initialization cleanup. Use AsyncExitStack to
attempt all owned provider shutdowns, adapter unregistration and HTTP closure
even when an earlier cleanup raises. Close auth before inspecting the current
provider instances, retaining the original reload ordering and class checks.

Consolidate duplicate endpoint/path validation tests, remove unused test
store history, and add parameterized storage/lifecycle regressions. Isolate
the example subclass registration during test imports.

Validation against original PR files with verified Git blob hashes:
- Original OAuth tests: 59 passed; SDK test blocked by missing openai.
- Persistence regressions reproduced 3 failures on the original code.
- Initial cleanup regressions reproduced 3 failures on the original code.
- Updated OAuth tests: 67 passed, 1 SDK test deselected locally.
- Example lifecycle tests: 8 passed, including replacement during auth close,
  with unrelated AstrBot runtime imports stubbed by a local-only runner.
  This is not a full AstrBot runtime integration test.
- Python compilation, Python 3.10 grammar and git diff --check passed.
- Ruff, the SDK test, full suite and real supplier login remain unverified
  locally due to unavailable dependencies/network. No CI rules weakened.

Relative to 444ddff: production +37/-39 (net -2); tests +136/-27.
Public API, existing security validation and dependencies are unchanged.

astrbot/core/provider/oauth.py

@@ -255,12 +255,8 @@ class OAuth2Session:
                     "code_verifier": pending[2],
                 }
             )
-            try:
-                await self._save_token(token)
-            except Exception:
-                raise OAuth2Error("Could not persist OAuth credentials.") from None
             # A new account must never inherit the previous account's refresh token.
-            self._token, self._loaded = token, True
+            await self._persist_token(token)
 
     async def cancel_authorization(self, *, owner: str) -> None:
         """Cancel this user's pending login without disconnecting the account.
@@ -308,15 +304,7 @@ class OAuth2Session:
                 {"grant_type": "refresh_token", "refresh_token": token.refresh_token},
                 previous=token,
             )
-            try:
-                await self._save_token(refreshed)
-            except Exception:
-                # A rotated refresh token cannot safely be retried with the old one.
-                self._token = None
-                raise OAuth2Error(
-                    "Could not persist OAuth credentials; sign in again."
-                ) from None
-            self._token = refreshed
+            await self._persist_token(refreshed)
             return refreshed.access_token
 
     async def authorize_request(self, request: httpx.Request) -> None:
@@ -355,11 +343,7 @@ class OAuth2Session:
         """
         async with self._lock:
             self._pending = None
-            self._token, self._loaded = None, True
-            try:
-                await self._save_token(None)
-            except Exception:
-                raise OAuth2Error("Could not delete OAuth credentials.") from None
+            await self._persist_token(None)
 
     async def close(self) -> None:
         """Disable authentication on plugin unload; retain persisted credentials."""
@@ -368,6 +352,19 @@ class OAuth2Session:
             self._pending = None
             self._token = None
 
+    async def _persist_token(self, token: OAuth2Token | None) -> None:
+        """Persist under the session lock, failing closed even on cancellation."""
+        # Never reuse an old, possibly rotated token after a failed/cancelled save.
+        self._token, self._loaded = None, True
+        try:
+            await self._save_token(token)
+        except Exception:
+            action = "delete" if token is None else "persist"
+            raise OAuth2Error(
+                f"Could not {action} OAuth credentials; sign in again."
+            ) from None
+        self._token = token
+
     async def _request_token(
         self, data: dict[str, str], previous: OAuth2Token | None = None
     ) -> OAuth2Token:
@@ -402,8 +399,7 @@ class OAuth2Session:
                     raise OAuth2Error("Invalid OAuth token response.")
                 if not response.is_success or "error" in payload:
                     if previous is not None and payload.get("error") == "invalid_grant":
-                        self._token, self._loaded = None, True
-                        await self._save_token(None)
+                        await self._persist_token(None)
                         raise OAuth2Error(
                             "OAuth authorization has expired; sign in again."
                         )

examples/astrbot_plugin_oauth_provider/main.py

@@ -3,6 +3,7 @@
 import hashlib
 import json
 import os
+from contextlib import AsyncExitStack
 from dataclasses import asdict
 
 import httpx
@@ -95,12 +96,7 @@ class OAuthProviderPlugin(Star):
                 "Manage the plugin provider's OAuth account",
             )
         except BaseException:
-            if self.oauth is not None:
-                await self.oauth.close()
-            if self.provider_class is not None:
-                unregister_provider_adapter(PROVIDER_TYPE, self.provider_class)
-            if self.http is not None:
-                await self.http.aclose()
+            await self.terminate()
             raise
 
     async def load_token(self) -> OAuth2Token | None:
@@ -174,16 +170,22 @@ class OAuthProviderPlugin(Star):
             return error_response("Provider operation failed; check the configuration.")
 
     async def terminate(self) -> None:
-        """Close auth before removing instances; preserve encrypted credentials."""
-        if self.oauth is not None:
-            await self.oauth.close()
-        try:
-            manager = self.context.provider_manager
-            for provider_id, provider in list(manager.inst_map.items()):
-                if type(provider) is self.provider_class:
-                    await manager.terminate_provider(provider_id)
-        finally:
-            if self.provider_class is not None:
-                unregister_provider_adapter(PROVIDER_TYPE, self.provider_class)
+        """Close all owned resources, even if an earlier cleanup fails."""
+        async with AsyncExitStack() as cleanup:
+            # Always attempt instance, registration and HTTP cleanup after auth.
             if self.http is not None:
-                await self.http.aclose()
+                cleanup.push_async_callback(self.http.aclose)
+            if self.provider_class is not None:
+                cleanup.callback(
+                    unregister_provider_adapter, PROVIDER_TYPE, self.provider_class
+                )
+            try:
+                if self.oauth is not None:
+                    await self.oauth.close()
+            finally:
+                manager = self.context.provider_manager
+                for provider_id, provider in reversed(list(manager.inst_map.items())):
+                    if type(provider) is self.provider_class:
+                        cleanup.push_async_callback(
+                            manager.terminate_provider, provider_id
+                        )

tests/test_provider_oauth.py

@@ -18,7 +18,6 @@ pytestmark = pytest.mark.asyncio
 class MemoryStore:
     def __init__(self, token=None):
         self.token = token
-        self.saved = []
         self.loads = 0
 
     async def load(self):
@@ -27,7 +26,6 @@ class MemoryStore:
 
     async def save(self, token):
         self.token = token
-        self.saved.append(token)
 
 
 def make_session(handler=None, token=None, **kwargs):
@@ -160,6 +158,9 @@ async def test_denial_consumes_attempt_without_echoing_provider_text():
 
 @pytest.mark.parametrize("field,value", [
     ("authorization_endpoint", "http://evil.example/authorize"),
+    ("authorization_endpoint", " https://login.example.com/auth"),
+    ("authorization_endpoint", "https://login.example.com/au\nth"),
+    ("authorization_endpoint", "https://login.example.com/a b"),
     ("token_endpoint", "https://user:secret@example.com/token"),
     ("token_endpoint", "https://@example.com/token"),
     ("token_endpoint", "https://example.com:bad/token"),
@@ -324,16 +325,45 @@ async def test_close_retains_store_but_blocks_old_transport():
             await session.authorize_request(httpx.Request("GET", "https://api.example.com/v1/models"))
 
 
-async def test_storage_errors_are_sanitized_and_refresh_fails_closed():
-    async def fail(*args):
-        raise RuntimeError("PRIVATE_REFRESH_TOKEN")
-    session, _, _, client = make_session(token=OAuth2Token("old", "refresh", 0), save_token=fail)
+@pytest.mark.parametrize("failure", [RuntimeError, asyncio.CancelledError])
+@pytest.mark.parametrize("operation", ["refresh", "login", "disconnect", "invalid_grant"])
+async def test_failed_storage_never_reuses_old_credentials(failure, operation):
+    async def fail(token):
+        raise failure("PRIVATE_REFRESH_TOKEN")
+
+    def invalid_grant(_):
+        return httpx.Response(400, json={"error": "invalid_grant"})
+
+    expired = operation in {"refresh", "invalid_grant"}
+    old = OAuth2Token("old", "refresh", 0 if expired else None)
+    session, store, calls, client = make_session(
+        invalid_grant if operation == "invalid_grant" else None,
+        token=old, save_token=fail,
+    )
     async with client:
-        with pytest.raises(OAuth2Error, match="persist") as error:
-            await session.get_access_token()
-        assert "PRIVATE_REFRESH_TOKEN" not in str(error.value)
+        if not expired:
+            assert await session.get_access_token() == "old"
+        expected = asyncio.CancelledError if failure is asyncio.CancelledError else OAuth2Error
+        with pytest.raises(expected) as error:
+            if operation == "login":
+                await session.complete_authorization(await callback(session), owner="admin")
+            elif operation == "disconnect":
+                await session.disconnect()
+            else:
+                await session.get_access_token()
+        if expected is OAuth2Error:
+            assert "PRIVATE_REFRESH_TOKEN" not in str(error.value)
+        # A failed/cancelled write must not reload or reuse an old rotated token.
         with pytest.raises(OAuth2Error, match="sign-in"):
             await session.get_access_token()
+        assert len(calls) == (0 if operation == "disconnect" else 1)
+        assert store.loads == 1
+
+
+async def test_load_error_is_sanitized():
+    async def fail():
+        raise RuntimeError("PRIVATE_REFRESH_TOKEN")
+
     session, _, _, client = make_session(load_token=fail)
     async with client:
         with pytest.raises(OAuth2Error, match="load") as error:
@@ -345,6 +375,8 @@ async def test_storage_errors_are_sanitized_and_refresh_fails_closed():
     "https://evil.example/v1/models", "http://api.example.com/v1/models",
     "https://api.example.com:444/v1/models", "https://api.example.com/v10/models",
     "https://api.example.com/other",
+    "https://api.example.com/v1/%2e%2e/admin",
+    "https://api.example.com/v1/%5c..%5cadmin",
 ])
 async def test_auth_hook_refuses_other_resources_before_loading_credentials(url):
     session, store, _, client = make_session(token=OAuth2Token("private"))
@@ -486,21 +518,3 @@ async def test_real_openai_sdk_discovery_chat_and_stream_use_current_bearer():
         assert sdk.api_key == "oauth-managed"
         assert len(token_requests) == 1
         assert len(requests) == 3
-
-
-@pytest.mark.parametrize("url", [
-    "https://api.example.com/v1/%2e%2e/admin",
-    "https://api.example.com/v1/%5c..%5cadmin",
-])
-async def test_encoded_api_path_escape_is_rejected(url):
-    session, store, _, client = make_session(token=OAuth2Token("secret"))
-    async with client:
-        with pytest.raises(OAuth2Error, match="outside"):
-            await session.authorize_request(httpx.Request("GET", url))
-        assert store.loads == 0
-
-
-@pytest.mark.parametrize("url", [" https://login.example.com/auth", "https://login.example.com/au\nth", "https://login.example.com/a b"])
-async def test_ambiguous_endpoint_controls_are_rejected(url):
-    with pytest.raises(ValueError):
-        make_session(authorization_endpoint=url)

tests/test_provider_oauth_example.py

@@ -0,0 +1,95 @@
+"""Lifecycle checks for the example, without starting AstrBot or calling a supplier."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from cryptography.fernet import Fernet
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+def example(monkeypatch):
+    from astrbot.core.star import base
+
+    # Importing a Star subclass registers it; keep the example out of global state.
+    monkeypatch.setattr(base, "star_map", {})
+    monkeypatch.setattr(base, "star_registry", [])
+    from examples.astrbot_plugin_oauth_provider import main
+
+    return main
+
+
+@pytest.mark.parametrize("failure", [None, "oauth", "first", "second", "unregister", "http", "replacement"])
+async def test_cleanup_attempts_all_owned_resources(monkeypatch, failure, example):
+    events = []
+
+    def record(name):
+        events.append(name)
+        if name == failure:
+            raise RuntimeError(name)
+
+    async def close_auth():
+        record("oauth")
+        if failure == "replacement":
+            # close() can wait for a refresh while hot reload replaces an instance.
+            plugin.context.provider_manager.inst_map["first"] = object()
+
+    async def close_http():
+        record("http")
+
+    async def terminate_provider(provider_id):
+        record(provider_id)
+
+    def unregister(provider_type, provider_class):
+        assert provider_type == example.PROVIDER_TYPE
+        assert provider_class is plugin.provider_class
+        record("unregister")
+
+    plugin = example.OAuthProviderPlugin.__new__(example.OAuthProviderPlugin)
+    plugin.provider_class = type("OwnedProvider", (), {})
+    plugin.oauth = SimpleNamespace(close=close_auth)
+    plugin.http = SimpleNamespace(aclose=close_http)
+    plugin.context = SimpleNamespace(provider_manager=SimpleNamespace(
+        inst_map={"first": plugin.provider_class(), "second": plugin.provider_class(), "foreign": object()},
+        terminate_provider=terminate_provider,
+    ))
+    monkeypatch.setattr(example, "unregister_provider_adapter", unregister)
+
+    if failure in {None, "replacement"}:
+        await plugin.terminate()
+    else:
+        with pytest.raises(RuntimeError, match=failure):
+            await plugin.terminate()
+    expected = ["oauth", "first", "second", "unregister", "http"]
+    if failure == "replacement":
+        expected.remove("first")
+    assert events == expected
+
+
+async def test_initialization_failure_releases_acquired_resources(monkeypatch, example):
+    monkeypatch.setenv("ASTRBOT_OAUTH_STORAGE_KEY", Fernet.generate_key().decode())
+    http = SimpleNamespace(aclose=AsyncMock())
+    session = SimpleNamespace(close=AsyncMock())
+    unregister = Mock()
+    monkeypatch.setattr(example.httpx, "AsyncClient", Mock(return_value=http))
+    monkeypatch.setattr(example, "OAuth2Session", Mock(return_value=session))
+    monkeypatch.setattr(example, "register_provider_adapter", lambda *a, **kw: lambda cls: cls)
+    monkeypatch.setattr(example, "unregister_provider_adapter", unregister)
+    plugin = example.OAuthProviderPlugin.__new__(example.OAuthProviderPlugin)
+    plugin.oauth = plugin.http = plugin.provider_class = None
+    plugin.config = dict.fromkeys(
+        ("client_id", "authorization_endpoint", "token_endpoint", "redirect_uri", "api_base"),
+        "unused-by-session-stub",
+    )
+    plugin.context = SimpleNamespace(
+        provider_manager=SimpleNamespace(inst_map={}),
+        register_web_api=Mock(side_effect=ValueError("route registration failed")),
+    )
+
+    with pytest.raises(ValueError, match="route registration failed"):
+        await plugin.initialize()
+    session.close.assert_awaited_once()
+    http.aclose.assert_awaited_once()
+    unregister.assert_called_once_with(example.PROVIDER_TYPE, plugin.provider_class)