fix(auth): preserve administrator and credential consistency (#7141)

app/api/endpoints/user.py

@@ -96,12 +96,12 @@ async def create_user(
     新增用户
     """
     user_info = user_in.model_dump()
-    if user_info.get("password"):
+    password = user_info.pop("password", None)
+    if password:
         try:
-            user_info["hashed_password"] = get_password_hash(user_info["password"])
+            user_info["hashed_password"] = get_password_hash(password)
         except PasswordTooLongError as error:
             return _SchemaResponse(success=False, message=str(error))
-        user_info.pop("password")
     try:
         user = await service.create(user_info)
     except UserNameConflictError:
@@ -192,7 +192,7 @@ async def upload_avatar(
         raise HTTPException(status_code=403, detail="用户权限不足")
 
     # 将文件转换为Base64
-    file_base64 = str(base64.b64encode(file.file.read()))
+    file_base64 = base64.b64encode(await file.read()).decode("ascii")
     # 更新到用户表
     user = await service.get_by_id(user_id)
     if not user:

app/application/github_auth.py

@@ -141,6 +141,8 @@ class GithubAuthService:
         self._client_id = client_id
         self._clock = clock
         self._lock = asyncio.Lock()
+        self._credential_lock = asyncio.Lock()
+        self._refresh_lock = asyncio.Lock()
         self._pending_sessions: dict[str, _PendingDeviceSession] = {}
 
     async def start_device_auth(self) -> GithubDeviceAuthStartResult:
@@ -281,6 +283,8 @@ class GithubAuthService:
                 token,
                 metadata,
             )
+            if not token:
+                return await self.status()
             granted_scopes = set(str(metadata.get("scope") or "").replace(",", " ").split())
             needs_reauthorization = needs_reauthorization or not set(
                 GITHUB_DEVICE_AUTH_SCOPES
@@ -328,26 +332,39 @@ class GithubAuthService:
         normalized = token.strip()
         if not normalized:
             raise GithubAuthError("GitHub Token 不能为空")
-        success, message = self._settings.update("GITHUB_TOKEN", normalized)
-        if success is False:
-            raise GithubAuthError(message or "GitHub Token 保存失败")
-        await self.forget_oauth_metadata()
+        await self._persist_credentials(normalized, None)
         return await self.status()
 
     async def clear_token(self) -> None:
         """
         清除 GitHub Token 和 OAuth 刷新元数据。
         """
-        success, message = self._settings.update("GITHUB_TOKEN", None)
-        if success is False:
-            raise GithubAuthError(message or "GitHub Token 清除失败")
-        await self.forget_oauth_metadata()
+        await self._persist_credentials(None, None)
 
     async def forget_oauth_metadata(self) -> None:
         """
         清除 OAuth 元数据但保留当前部署设置中的 Token 值。
         """
-        await self._system_config.async_delete(self._AUTH_CONFIG_KEY)
+        async with self._credential_lock:
+            await self._system_config.async_delete(self._AUTH_CONFIG_KEY)
+
+    async def _persist_credentials(
+        self, token: str | None, metadata: dict[str, Any] | None,
+    ) -> None:
+        """
+        串行发布凭据和对应元数据，防止迟到刷新覆盖手动清除或更换。
+
+        :param token: 新访问令牌；None 表示清除
+        :param metadata: OAuth 元数据；None 表示手动凭据或清除
+        """
+        async with self._credential_lock:
+            success, message = self._settings.update("GITHUB_TOKEN", token)
+            if success is False:
+                raise GithubAuthError(message or "GitHub Token 保存失败")
+            if metadata is None:
+                await self._system_config.async_delete(self._AUTH_CONFIG_KEY)
+            else:
+                await self._system_config.async_set(self._AUTH_CONFIG_KEY, metadata)
 
     async def _save_oauth_token(
         self,
@@ -359,12 +376,6 @@ class GithubAuthService:
         """
         if not exchange.access_token:
             raise GithubAuthError("GitHub 未返回访问 Token")
-        success, message = self._settings.update(
-            "GITHUB_TOKEN",
-            exchange.access_token,
-        )
-        if success is False:
-            raise GithubAuthError(message or "GitHub Token 保存失败")
         metadata: dict[str, Any] = {
             "login": github_user.login,
             "scope": exchange.scope or "",
@@ -375,7 +386,7 @@ class GithubAuthService:
             metadata["expires_at"] = int(self._clock() + max(exchange.expires_in, 0))
         if exchange.refresh_token_expires_in is not None:
             metadata["refresh_token_expires_at"] = int(self._clock() + max(exchange.refresh_token_expires_in, 0))
-        await self._system_config.async_set(self._AUTH_CONFIG_KEY, metadata)
+        await self._persist_credentials(exchange.access_token, metadata)
 
     async def _refresh_if_needed(
         self,
@@ -383,7 +394,27 @@ class GithubAuthService:
         metadata: dict[str, Any],
     ) -> tuple[str, dict[str, Any], bool]:
         """
-        在过期窗口内刷新 OAuth Token，失败时保留原值供诊断和重试。
+        串行刷新 OAuth 快照，已被清除或替换时返回当前凭据。
+
+        :param token: 调用方读取的访问令牌快照
+        :param metadata: 与令牌对应的 OAuth 元数据快照
+        :return: 当前访问令牌、对应元数据及是否需要重新授权
+        """
+        async with self._refresh_lock:
+            current_token, current_metadata = self._read_access_token(), self._read_auth_metadata()
+            if (current_token, current_metadata) != (token, metadata):
+                return current_token, current_metadata, False
+            return await self._refresh_current_token(token, metadata)
+
+    async def _refresh_current_token(
+        self, token: str, metadata: dict[str, Any],
+    ) -> tuple[str, dict[str, Any], bool]:
+        """
+        刷新单份仍有效的快照，并在写回时核对它未被其他凭据操作替换。
+
+        :param token: 刷新锁内已确认仍为当前值的访问令牌
+        :param metadata: 已确认对应当前令牌的 OAuth 元数据
+        :return: 当前访问令牌、对应元数据及是否需要重新授权
         """
         expires_at = self._as_int(metadata.get("expires_at"))
         if expires_at is None or expires_at > int(self._clock()) + 60:
@@ -397,9 +428,9 @@ class GithubAuthService:
                 refresh_token,
             )
         except (GithubAuthError, GithubAuthTransportError):
-            return token, metadata, True
+            return await self._failed_refresh_snapshot(token, metadata)
         if not exchange.access_token:
-            return token, metadata, True
+            return await self._failed_refresh_snapshot(token, metadata)
         refreshed_metadata = copy.deepcopy(metadata)
         if exchange.refresh_token:
             refreshed_metadata["refresh_token"] = exchange.refresh_token
@@ -409,15 +440,30 @@ class GithubAuthService:
             refreshed_metadata["refresh_token_expires_at"] = int(
                 self._clock() + max(exchange.refresh_token_expires_in, 0)
             )
-        success, _ = self._settings.update("GITHUB_TOKEN", exchange.access_token)
-        if success is False:
-            return token, metadata, True
-        await self._system_config.async_set(
-            self._AUTH_CONFIG_KEY,
-            refreshed_metadata,
-        )
+        async with self._credential_lock:
+            current_token, current_metadata = self._read_access_token(), self._read_auth_metadata()
+            if (current_token, current_metadata) != (token, metadata):
+                return current_token, current_metadata, False
+            success, _ = self._settings.update("GITHUB_TOKEN", exchange.access_token)
+            if success is False:
+                return token, metadata, True
+            await self._system_config.async_set(self._AUTH_CONFIG_KEY, refreshed_metadata)
         return exchange.access_token, refreshed_metadata, False
 
+    async def _failed_refresh_snapshot(
+        self, token: str, metadata: dict[str, Any],
+    ) -> tuple[str, dict[str, Any], bool]:
+        """
+        刷新失败后返回当前凭据，清除或更换已生效时不再向调用者交付旧令牌。
+
+        :param token: 本次远端刷新使用的访问令牌快照
+        :param metadata: 本次远端刷新使用的 OAuth 元数据快照
+        :return: 当前凭据；仅原快照仍有效时标记需要重新授权
+        """
+        async with self._credential_lock:
+            current_token, current_metadata = self._read_access_token(), self._read_auth_metadata()
+            return current_token, current_metadata, (current_token, current_metadata) == (token, metadata)
+
     async def _session_interval(self, session_id: str) -> int:
         """
         读取会话当前轮询间隔，并在会话刚过期时返回最小等待值。

app/application/security/throttle.py

@@ -90,6 +90,7 @@ class LoginThrottle:
         """
         now = self._clock()
         with self._lock:
+            self._cleanup(now)
             entry = self._entries.get(key)
             if entry is None:
                 return 0
@@ -102,6 +103,7 @@ class LoginThrottle:
         """
         now = self._clock()
         with self._lock:
+            self._cleanup(now)
             entry = self._entries.setdefault(key, _Entry())
             entry.failures += 1
             entry.updated_at = now

app/application/settings/contract.py

@@ -721,11 +721,13 @@ def _runtime_settings_schema() -> dict[str, Any]:
     return Settings.model_json_schema()
 
 
-def validate_setting_value(spec: SettingSpec, value: Any) -> None:
+def validate_setting_value(spec: SettingSpec, value: Any) -> Any:
     """
-    按合同类型验证数据库设置的完整候选值。
+    按合同类型验证数据库设置的完整候选值，并返回规范化后的值。
+
+    :return: 类型转换后的候选值，调用方应将其序列化后持久化
     """
-    TypeAdapter(spec.value_annotation).validate_python(value)
+    return TypeAdapter(spec.value_annotation).validate_python(value)
 
 
 CORE_SETTING_SPECS, SYSTEMCONFIG_SETTING_SPECS = build_setting_specs()

app/application/settings/service.py

@@ -9,6 +9,8 @@ import threading
 from collections.abc import Awaitable, Callable
 from typing import Any, ContextManager, Optional, cast
 
+from pydantic import TypeAdapter
+
 from app.application.configuration import (
     RuntimeSettingsService,
     SystemConfigService,
@@ -675,8 +677,8 @@ class SystemSettingsService:
             match_value,
         )
         normalized_value = self._normalize_systemconfig_value(next_value)
-        validate_setting_value(spec, normalized_value)
-        return normalized_value
+        validated_value = validate_setting_value(spec, normalized_value)
+        return TypeAdapter(spec.value_annotation).dump_python(validated_value, mode="json", exclude_unset=True)
 
     async def update(
         self,

app/db/adapters/user.py

@@ -8,7 +8,7 @@ from collections.abc import Callable, Mapping
 from contextlib import AbstractAsyncContextManager
 from typing import Any, Optional, cast
 
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
@@ -233,12 +233,21 @@ class SqlAlchemyUserRepository(UserRepository):
     ) -> Optional[User]:
         """
         先锁管理员集合再锁目标用户，保护并发下最后一个启用管理员。
+
+        SQLite 忽略 FOR UPDATE；先获取数据库写锁，并一直持有到调用方提交或回滚，
+        防止多个事务基于旧管理员计数分别移除最后的两个账号。
         """
+        if session.get_bind().dialect.name == "sqlite":
+            await session.execute(
+                update(User).where(User.is_superuser.is_(True)).values(id=User.id)
+                .execution_options(synchronize_session=False)
+            )
         result = await session.execute(
             select(User)
             .where(User.is_active.is_(True), User.is_superuser.is_(True))
             .order_by(User.id)
             .with_for_update()
+            .execution_options(populate_existing=True)
         )
         administrators = list(result.scalars().all())
         model = next(
@@ -246,7 +255,10 @@ class SqlAlchemyUserRepository(UserRepository):
             None,
         )
         if model is None:
-            locked = await session.execute(select(User).where(User.id == user_id).with_for_update())
+            locked = await session.execute(
+                select(User).where(User.id == user_id).with_for_update()
+                .execution_options(populate_existing=True)
+            )
             model = cast(Optional[User], locked.scalars().first())
             if model is None:
                 return None

app/db/oper/userconfig.py

@@ -8,6 +8,7 @@ from collections.abc import Iterator
 from contextlib import contextmanager
 from typing import Optional, Union
 
+from sqlalchemy import update
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
@@ -75,9 +76,15 @@ class UserConfigOper(DbOper, metaclass=Singleton):
     ) -> bool:
         """
         在调用方 Session 中暂存写入，返回提交后是否应移除缓存项。
+
+        SQLite 写锁在读取配置前获取，避免并发删除用户后对已消失配置执行过期更新。
+        事务提交和回滚仍由调用方负责，用户不存在时由既有外键约束拒绝新增。
         """
         if isinstance(key, UserConfigKey):
             key = key.value
+        if db.get_bind().dialect.name == "sqlite":
+            db.execute(update(UserConfig).where(UserConfig.username == username, UserConfig.key == key)
+                       .values(id=UserConfig.id).execution_options(synchronize_session=False))
         conf = UserConfig.get_by_key(db=db, username=username, key=key)
         if conf:
             if value:

app/schemas/initialization.py

@@ -28,13 +28,13 @@ class InitializationRequest(BaseModel):
     confirm_password: str = Field(min_length=6, max_length=50, description="再次输入的管理员密码，必须与 password 完全一致")
     api_key: str = Field(min_length=16, max_length=256, description="接口访问密钥，长度为 16 至 256 个字符")
 
-    @field_validator("username", "api_key")
+    @field_validator("username", "api_key", mode="before")
     @classmethod
     def strip_text(cls, value: str) -> str:
         """
         去除用户名和 API Key 的首尾空白，避免产生不可见凭据差异。
         """
-        return value.strip()
+        return value.strip() if isinstance(value, str) else value
 
     @field_validator("password", "confirm_password")
     @classmethod

tests/test_api_authorization.py

@@ -8,7 +8,7 @@ import io
 from types import SimpleNamespace
 
 import pytest
-from fastapi import HTTPException
+from fastapi import HTTPException, UploadFile
 from starlette.requests import Request
 from starlette.responses import Response
 
@@ -506,7 +506,7 @@ def test_upload_avatar_rejects_other_user_for_non_superuser():
     普通用户不能通过 user_id 参数修改其他用户头像。
     """
     current_user = SimpleNamespace(id=1, is_superuser=False)
-    upload_file = SimpleNamespace(file=io.BytesIO(b"avatar"), filename="avatar.png")
+    upload_file = UploadFile(file=io.BytesIO(b"avatar"), filename="avatar.png")
 
     with pytest.raises(HTTPException) as exc_info:
         asyncio.run(
@@ -529,7 +529,7 @@ def test_upload_avatar_returns_filename_in_data(monkeypatch):
 
     fake_user = SimpleNamespace()
     current_user = SimpleNamespace(id=1, is_superuser=False)
-    upload_file = SimpleNamespace(file=io.BytesIO(b"avatar"), filename="avatar.png")
+    upload_file = UploadFile(file=io.BytesIO(b"avatar"), filename="avatar.png")
 
     class FakeService:
         """
@@ -565,4 +565,4 @@ def test_upload_avatar_returns_filename_in_data(monkeypatch):
     assert response.data == {"filename": "avatar.png"}
     assert response.message == ""
     assert not hasattr(response, "message_i18n")
-    assert fake_user.values == {"avatar": "data:image/ico;base64,b'YXZhdGFy'"}
+    assert fake_user.values == {"avatar": "data:image/ico;base64,YXZhdGFy"}

tests/test_identity_settings_review_regressions.py

@@ -0,0 +1,264 @@
+"""
+用户凭据、并发管理员保护与设置类型的审查回归。
+"""
+
+import asyncio
+import threading
+from io import BytesIO
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from fastapi import UploadFile
+from pydantic import ValidationError
+from sqlalchemy import event, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from app.api.endpoints.user import create_user, upload_avatar
+from app.application.github_auth import GithubAuthService
+from app.application.security.throttle import LoginThrottle
+from app.application.security.user import LastActiveSuperuserError
+from app.application.settings.service import SystemSettingsService
+from app.db.adapters.configuration import TransactionalUserConfigurationRepository
+from app.db.adapters.user import SqlAlchemyUserRepository
+from app.db.models.user import User
+from app.db.models.userconfig import UserConfig
+from app.db.oper.userconfig import UserConfigOper
+from app.domain.github import GithubAuthTransportError, GithubTokenExchange, GithubUser
+from app.schemas.initialization import InitializationRequest
+from app.schemas.types import SystemConfigKey
+from app.schemas.user import UserCreate
+from tests.test_user_repository import _user_write_context
+
+
+@pytest.mark.asyncio
+async def test_avatar_contains_decodable_base64() -> None:
+    """
+    头像 data URL 只携带 Base64 文本，不包含 bytes 的 Python 表示。
+    """
+    import base64
+
+    service = SimpleNamespace(get_by_id=AsyncMock(return_value=object()), update=AsyncMock())
+    file = UploadFile(filename="avatar.ico", file=BytesIO(b"review-image"))
+    response = await upload_avatar(7, service, file, SimpleNamespace(id=7, is_superuser=False))
+    assert response.success
+    encoded = service.update.call_args.args[1]["avatar"].split(",", 1)[1]
+    assert base64.b64decode(encoded, validate=True) == b"review-image"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("password", [None, ""])
+async def test_create_user_removes_empty_optional_password(password) -> None:
+    """
+    可选密码为空时移除明文字段，创建无本地密码用户仍符合合同。
+    """
+    service = SimpleNamespace(create=AsyncMock(return_value=object()))
+    response = await create_user(service=service, user_in=UserCreate(name="review", password=password))
+    assert response.success
+    assert "password" not in service.create.call_args.args[0]
+    assert "hashed_password" not in service.create.call_args.args[0]
+
+
+@pytest.mark.parametrize("field,value", [("username", "   "), ("api_key", " " * 16)])
+def test_initialization_rejects_blank_after_normalization(field, value) -> None:
+    """
+    首尾空白在长度约束之前去除，空账号和空令牌不能完成初始化。
+    """
+    payload = {"username": "review", "password": "Test123!", "confirm_password": "Test123!",
+               "api_key": "review-api-token-12345678", field: value}
+    with pytest.raises(ValidationError):
+        InitializationRequest.model_validate(payload)
+
+
+def test_login_failure_ttl_resets_same_key() -> None:
+    """
+    同一用户隔超时再次失败从首次计数开始，未过期锁仍按预算保留。
+    """
+    now = [0.0]
+    throttle = LoginThrottle(max_failures=2, ttl_seconds=10, clock=lambda: now[0])
+    key = ("127.0.0.1", "review")
+    assert throttle.record_failure(key) == 0
+    now[0] = 11
+    assert throttle.retry_after(key) == 0
+    assert throttle.record_failure(key) == 0
+    assert throttle.record_failure(key) == 30
+    now[0] = 22
+    assert throttle.retry_after(key) == 19
+
+
+@pytest.mark.asyncio
+async def test_settings_persists_type_adapter_result() -> None:
+    """
+    合同接受的站点数字文本持久化为整数，消费者按站点主键匹配。
+    """
+    saved = {}
+    async def write(key, value, **_kwargs):
+        """
+        在内存端口记录已规范化候选，保留真实服务的类型处理。
+        """
+        saved[key] = value
+        return SimpleNamespace(normalized_value=value, changed=True)
+    config = SimpleNamespace(get=lambda key: saved.get(key), async_set_with_normalized_value=write)
+    service = SystemSettingsService(None, config, AsyncMock())
+    await service.update(setting_key="IndexerSites", value=["1", "2"])
+    assert saved[SystemConfigKey.IndexerSites] == [1, 2]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mutation", ["demote", "disable", "delete"])
+async def test_concurrent_sqlite_admin_changes_keep_one(tmp_path, mutation) -> None:
+    """
+    两个独占 SQLite 请求同时移除不同管理员时，后一个事务必须看到最新人数。
+    """
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'users.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(User.__table__.create)
+        async with AsyncSession(engine) as session:
+            session.add_all([User(name="a", is_active=True, is_superuser=True),
+                             User(name="b", is_active=True, is_superuser=True)])
+            await session.commit()
+        first_flushed, release_first = asyncio.Event(), asyncio.Event()
+        async def change(user_id, first=False):
+            """
+            首事务在提交前暂停，确保次事务遇到仍未落账的管理员修改。
+            """
+            async with AsyncSession(engine, expire_on_commit=False) as session:
+                repository = SqlAlchemyUserRepository(session)
+                try:
+                    if mutation == "delete":
+                        await repository.async_delete(user_id)
+                    else:
+                        await repository.async_update(user_id, {
+                            "is_superuser" if mutation == "demote" else "is_active": False})
+                    if first:
+                        first_flushed.set()
+                        await release_first.wait()
+                    await session.commit()
+                    return "changed"
+                except LastActiveSuperuserError:
+                    await session.rollback()
+                    return "protected"
+        first = asyncio.create_task(change(1, first=True))
+        await asyncio.wait_for(first_flushed.wait(), 5)
+        second = asyncio.create_task(change(2))
+        try:
+            # 次事务必须等首事务提交，或在首事务完成后以新快照拒绝。
+            await asyncio.sleep(0.05)
+        finally:
+            release_first.set()
+        assert sorted(await asyncio.wait_for(asyncio.gather(first, second), 5)) == ["changed", "protected"]
+        async with AsyncSession(engine) as session:
+            remaining = (await session.execute(select(User).where(
+                User.is_active.is_(True), User.is_superuser.is_(True)))).scalars().all()
+            assert len(remaining) == 1
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_sqlite_userconfig_waits_for_user_delete_before_reading(monkeypatch, tmp_path) -> None:
+    """
+    删除已暂存但未提交时，配置写入先等数据库写锁，不能先读旧行再过期更新。
+    """
+    async with _user_write_context(tmp_path / "delete-config.db") as (session, sync_factory):
+        user = User(name="old", is_active=True, is_superuser=False)
+        session.add_all([user, UserConfig(username="old", key="theme", value="initial")])
+        await session.commit()
+        snapshot = object.__new__(UserConfigOper)
+        snapshot.__init__()
+        configuration = TransactionalUserConfigurationRepository(sync_factory, snapshot)
+        configuration.load_snapshot()
+        await SqlAlchemyUserRepository(session).async_delete(user.id)
+        attempted, read_entered = threading.Event(), threading.Event()
+        original_read = UserConfig.get_by_key
+
+        def record_read(**kwargs):
+            """
+            观察真实查询入口，保留查询和持久化算法。
+            """
+            read_entered.set()
+            return original_read(**kwargs)
+
+        def before_sql(_connection, _cursor, _statement, _parameters, _context, _executemany):
+            """
+            在独立同步连接第一次尝试执行 SQL 时唤醒测试。
+            """
+            attempted.set()
+
+        monkeypatch.setattr(UserConfig, "get_by_key", record_read)
+        bind = sync_factory.kw["bind"]
+        event.listen(bind, "before_cursor_execute", before_sql)
+        writing = asyncio.create_task(asyncio.to_thread(configuration.set, "old", "theme", "concurrent"))
+        try:
+            assert await asyncio.to_thread(attempted.wait, 5)
+            assert not read_entered.is_set()
+        finally:
+            await session.commit()
+            outcomes = await asyncio.wait_for(asyncio.gather(writing, return_exceptions=True), 5)
+            event.remove(bind, "before_cursor_execute", before_sql)
+        assert len(outcomes) == 1 and isinstance(outcomes[0], IntegrityError), outcomes
+        with sync_factory() as verify:
+            assert verify.execute(select(UserConfig)).scalars().all() == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("action", ["clear", "manual"])
+@pytest.mark.parametrize("refresh_fails", [False, True])
+async def test_github_refresh_does_not_restore_replaced_credentials(action, refresh_fails) -> None:
+    """
+    远端刷新等待期间清除或更换凭据，迟到结果不能恢复旧账号授权。
+    """
+    settings = Mock()
+    token = ["old-access"]
+    metadata = [{"login": "old-user", "expires_at": 1000, "refresh_token": "old-refresh"}]
+    entered, release = asyncio.Event(), asyncio.Event()
+    settings.get.side_effect = lambda _key: token[0]
+    def update(_key, value):
+        """
+        更新内存部署设置，避免写入真实配置。
+        """
+        token[0] = value
+        return True, ""
+    settings.update.side_effect = update
+    config = Mock()
+    config.get.side_effect = lambda _key: dict(metadata[0]) if metadata[0] else None
+    async def set_config(_key, value):
+        """
+        保存独立的 OAuth 元数据快照。
+        """
+        metadata[0] = dict(value)
+    async def delete_config(_key):
+        """
+        删除内存 OAuth 快照。
+        """
+        metadata[0] = None
+    config.async_set = set_config
+    config.async_delete = delete_config
+    async def refresh(*_args):
+        """
+        在远端响应边界制造可控并发窗口。
+        """
+        entered.set()
+        await release.wait()
+        if refresh_fails:
+            raise GithubAuthTransportError("隔离刷新失败")
+        return GithubTokenExchange(access_token="stale-refreshed", refresh_token="rotated", expires_in=3600)
+    transport = SimpleNamespace(refresh_access_token=refresh,
+                                get_user=AsyncMock(return_value=GithubUser(login="manual-user")))
+    service = GithubAuthService(settings=settings, system_config=config, transport=transport,
+                                client_id="review", clock=lambda: 1000)
+    refreshing = asyncio.create_task(service.access_token())
+    await asyncio.wait_for(entered.wait(), 5)
+    try:
+        if action == "clear":
+            await service.clear_token()
+        else:
+            await service.set_manual_token("new-manual")
+    finally:
+        release.set()
+    result = await asyncio.wait_for(refreshing, 5)
+    assert token[0] == (None if action == "clear" else "new-manual")
+    assert result == ("" if action == "clear" else "new-manual")
+    assert metadata[0] is None