feat(auth): add first-run administrator initialization

app/api/endpoints/login.py

@@ -1,3 +1,4 @@
+import asyncio
 from datetime import timedelta
 from typing import Annotated, Any, List
 
@@ -7,19 +8,104 @@ from fastapi.security import OAuth2PasswordRequestForm
 
 from app.adapters.web.security.access import set_or_refresh_resource_token_cookie
 from app.api.context import get_api_runtime_config, resolve_api_runtime_config
+from app.api.dependencies.auth import get_user_service
 from app.api.response import RAW_RESPONSE_OPENAPI_KEY, ResponseAPIRouter
-from app.application.configuration import ApiRuntimeConfig, get_configured_system_config
+from app.application.configuration import ApiRuntimeConfig, get_runtime_settings
 from app.application.image import WallpaperHelper
-from app.application.security.token import create_access_token
+from app.application.security.token import PasswordTooLongError, create_access_token, get_password_hash
+from app.application.security.user import UserNameConflictError, UserService
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
 from app.chain.user import MfaRequired, UserChain
 from app.schemas.response import Response as _SchemaResponse
+from app.schemas.initialization import InitializationRequest as _SchemaInitializationRequest
+from app.schemas.initialization import InitializationStatus as _SchemaInitializationStatus
 from app.schemas.token import MfaChallenge as _SchemaMfaChallenge
 from app.schemas.token import Token as _SchemaToken
 from app.schemas.token import TokenPayload as _SchemaTokenPayload
-from app.schemas.types import SystemConfigKey
 
 router = ResponseAPIRouter()
+_INITIALIZATION_LOCK = asyncio.Lock()
+
+
+@router.get(
+    "/initialization",
+    summary="查询首次初始化状态",
+    response_model=_SchemaResponse[_SchemaInitializationStatus],
+)
+async def get_initialization_status(
+    service: UserService = Depends(get_user_service),
+) -> _SchemaResponse[_SchemaInitializationStatus]:
+    """返回当前实例是否已经存在用户，供启动页决定是否接管导航。"""
+    return _SchemaResponse(
+        success=True,
+        data=_SchemaInitializationStatus(initialized=await service.is_initialized()),
+    )
+
+
+@router.post(
+    "/initialization",
+    summary="完成首次初始化",
+    response_model=_SchemaResponse[None],
+)
+async def initialize_instance(
+    payload: _SchemaInitializationRequest,
+    service: UserService = Depends(get_user_service),
+) -> _SchemaResponse[None]:
+    """原子创建首个超级管理员，并保存 API Key 供后续服务认证。"""
+    async with _INITIALIZATION_LOCK:
+        if await service.is_initialized():
+            raise HTTPException(status_code=409, detail="系统已经完成初始化")
+
+        runtime_settings = get_runtime_settings()
+        previous_superuser = runtime_settings.get("SUPERUSER", "")
+        previous_api_token = runtime_settings.get("API_TOKEN")
+        updated_keys: list[str] = []
+        try:
+            for key, value in (("SUPERUSER", payload.username), ("API_TOKEN", payload.api_key)):
+                success, message = runtime_settings.update(key, value)
+                if success is False:
+                    raise RuntimeError(message or f"配置项 {key} 更新失败")
+                updated_keys.append(key)
+
+            try:
+                hashed_password = get_password_hash(payload.password)
+            except PasswordTooLongError as error:
+                raise HTTPException(status_code=422, detail=str(error)) from error
+
+            created_user = await service.create(
+                {
+                    "name": payload.username,
+                    "email": "admin@movie-pilot.org",
+                    "hashed_password": hashed_password,
+                    "is_active": True,
+                    "is_superuser": True,
+                    "avatar": "",
+                    "is_otp": False,
+                    "otp_secret": None,
+                    "permissions": {},
+                    "settings": {},
+                }
+            )
+            if created_user is None:
+                raise RuntimeError("管理员用户创建失败")
+        except UserNameConflictError as error:
+            raise HTTPException(status_code=409, detail="用户名已被使用") from error
+        except HTTPException:
+            for key in reversed(updated_keys):
+                runtime_settings.update(
+                    key,
+                    previous_superuser if key == "SUPERUSER" else previous_api_token,
+                )
+            raise
+        except Exception as error:
+            for key in reversed(updated_keys):
+                runtime_settings.update(
+                    key,
+                    previous_superuser if key == "SUPERUSER" else previous_api_token,
+                )
+            raise HTTPException(status_code=400, detail=str(error)) from error
+
+    return _SchemaResponse(success=True)
 
 
 @router.post(
@@ -68,11 +154,6 @@ def login_access_token(
 
     # 用户等级
     level = SitesHelper().auth_level
-    # 是否显示配置向导
-    show_wizard = (
-        not get_configured_system_config().get(SystemConfigKey.SetupWizardState)
-        and not runtime_config.advanced_mode
-    )
     access_token = create_access_token(
         userid=user_or_message.id,
         username=user_or_message.name,
@@ -101,7 +182,6 @@ def login_access_token(
         avatar=user_or_message.avatar,
         level=level,
         permissions=user_or_message.permissions or {},
-        wizard=show_wizard,
     )
 
 

app/api/endpoints/system.py

@@ -478,7 +478,6 @@ def get_global_setting(token: str):
         include={
             "TMDB_IMAGE_DOMAIN",
             "GLOBAL_IMAGE_CACHE",
-            "ADVANCED_MODE",
         }
     )
     # 追加版本信息（用于版本检查）

app/application/configuration.py

@@ -99,7 +99,6 @@ class TokenRuntimeConfig:
 class ApiRuntimeConfig:
     """单次 API 请求使用的宿主配置快照。"""
 
-    advanced_mode: bool
     access_token_expire_minutes: int
     btrfs_fsid_dedup: bool
     ai_agent_enable: bool
@@ -172,7 +171,7 @@ class ChainRuntimeConfig:
     root_path: Path = Path(".")
     config_path: Path = Path(".")
     frontend_path: Path = Path(".")
-    superuser: str = "admin"
+    superuser: str = ""
     media_recognize_share: bool = False
     auxiliary_auth_enable: bool = False
     global_image_cache: bool = False

app/application/security/auth.py

@@ -13,7 +13,6 @@ from app.application.site.sites import SitesHelper  # pylint: disable=import-err
 from app.foundation.singleton import Singleton
 from app.schemas.token import Token as _SchemaToken
 from app.schemas.token import TokenPayload as _SchemaTokenPayload
-from app.schemas.types import SystemConfigKey
 from app.schemas.user import UserPermissions
 
 
@@ -209,10 +208,6 @@ class AuthService:
         """使用统一逻辑构造登录 Token 响应。"""
         level = SitesHelper().auth_level
         config = get_api_runtime_config_snapshot()
-        show_wizard = (
-            not self._config.get(SystemConfigKey.SetupWizardState)
-            and not config.advanced_mode
-        )
         return _SchemaToken(
             access_token=create_access_token(
                 userid=user.id,
@@ -228,7 +223,6 @@ class AuthService:
             avatar=user.avatar,
             level=level,
             permissions=cast(UserPermissions, dict(user.permissions)),
-            wizard=show_wizard,
         )
 
 

app/application/security/user.py

@@ -176,6 +176,9 @@ class ChainUserRepository(Protocol):
 class UserRepository(Protocol):
     """用户用例所需的最小异步数据端口。"""
 
+    async def async_has_users(self) -> bool:
+        """判断数据库中是否已经存在任意用户。"""
+
     async def async_list(self) -> list[UserSnapshot]:
         """返回全部用户。"""
 
@@ -251,6 +254,10 @@ class UserService:
         """返回用户列表。"""
         return await self._repository.async_list()
 
+    async def is_initialized(self) -> bool:
+        """判断系统是否已经完成首次用户初始化。"""
+        return await self._repository.async_has_users()
+
     async def get_by_name(self, name: str) -> Optional[UserSnapshot]:
         """按用户名查询用户。"""
         return await self._repository.async_get_by_name(name)

app/db/adapters/user.py

@@ -69,6 +69,12 @@ class SqlAlchemyUserRepository(UserRepository):
         model = self._oper.get_by_id(user_id)
         return _to_snapshot(model) if model else None
 
+    async def async_has_users(self) -> bool:
+        """使用最小列查询判断数据库中是否已有用户。"""
+        session = self._require_async_session()
+        result = await session.execute(select(User.id).limit(1))
+        return result.scalar_one_or_none() is not None
+
     async def async_list(self) -> list[UserSnapshot]:
         """在异步请求会话中读取全部冻结用户快照。"""
         return [_to_snapshot(model) for model in await self._oper.async_list()]

app/runtime/config.py

@@ -106,9 +106,6 @@ class ConfigModel(BaseModel):
     DEBUG: bool = False
     # 是否开发模式
     DEV: bool = False
-    # 高级设置模式
-    ADVANCED_MODE: bool = True
-
     # ==================== 安全认证配置 ====================
     # 密钥
     SECRET_KEY: str = secrets.token_urlsafe(32)
@@ -120,10 +117,10 @@ class ConfigModel(BaseModel):
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
     # RESOURCE_TOKEN过期时间
     RESOURCE_ACCESS_TOKEN_EXPIRE_SECONDS: int = 60 * 30
-    # 超级管理员初始用户名
-    SUPERUSER: str = "admin"
-    # 超级管理员初始密码
-    SUPERUSER_PASSWORD: Optional[str] = None
+    # 超级管理员用户名；V3 首次启动时为空，由初始化页面设置
+    SUPERUSER: str = ""
+    # 超级管理员密码不再通过部署配置初始化，始终由数据库存储哈希
+    SUPERUSER_PASSWORD: str = ""
     # 辅助认证，允许通过外部服务进行认证、单点登录以及自动创建用户
     AUXILIARY_AUTH_ENABLE: bool = False
     # API密钥，需要更换
@@ -800,16 +797,13 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
         if isinstance(value, (list, dict, set)):
             value = copy.deepcopy(value)
         value = value.strip() if isinstance(value, str) else None
-        if not value or len(value) < 16:
+        if not value:
+            return None, str(original_value) not in {"", "None"}
+        if len(value) < 16:
             new_token = secrets.token_urlsafe(16)
-            if not value:
-                logger.info(
-                    f"'API_TOKEN' 未设置，已随机生成新的【API_TOKEN】{new_token}"
-                )
-            else:
-                logger.warning(
-                    f"'API_TOKEN' 长度不足 16 个字符，存在安全隐患，已随机生成新的【API_TOKEN】{new_token}"
-                )
+            logger.warning(
+                f"'API_TOKEN' 长度不足 16 个字符，存在安全隐患，已随机生成新的【API_TOKEN】{new_token}"
+            )
             return new_token, True
         return value, str(value) != str(original_value)
 
@@ -1339,7 +1333,7 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
     def VAPID(self):
         """返回 Web Push 使用的 VAPID 配置。"""
         return {
-            "subject": f"mailto:{self.SUPERUSER}@movie-pilot.org",
+            "subject": f"mailto:{self.SUPERUSER or 'moviepilot'}@movie-pilot.org",
             "publicKey": "BH3w49sZA6jXUnE-yt4jO6VKh73lsdsvwoJ6Hx7fmPIDKoqGiUl2GEoZzy-iJfn4SfQQcx7yQdHf9RknwrL_lSM",
             "privateKey": "JTixnYY0vEw97t9uukfO3UWKfHKJdT5kCQDiv3gu894",
         }

app/schemas/initialization.py

@@ -0,0 +1,41 @@
+"""首次初始化接口使用的请求与状态模型。"""
+
+import re
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+
+class InitializationStatus(BaseModel):
+    """描述 MoviePilot 是否已经存在可用的本地用户记录。"""
+
+    initialized: bool
+
+
+class InitializationRequest(BaseModel):
+    """首次初始化超级管理员、密码和 API Key 所需的提交数据。"""
+
+    username: str = Field(min_length=1, max_length=50)
+    password: str = Field(min_length=6, max_length=50)
+    confirm_password: str = Field(min_length=6, max_length=50)
+    api_key: str = Field(min_length=16, max_length=256)
+
+    @field_validator("username", "api_key")
+    @classmethod
+    def strip_text(cls, value: str) -> str:
+        """去除用户名和 API Key 的首尾空白，避免产生不可见凭据差异。"""
+        return value.strip()
+
+    @field_validator("password", "confirm_password")
+    @classmethod
+    def validate_password_shape(cls, value: str) -> str:
+        """复用用户管理中的密码复杂度约束。"""
+        if not re.match(r"^(?![a-zA-Z]+$)(?!\d+$)(?![^\da-zA-Z\s]+$).{6,50}$", value):
+            raise ValueError("密码需要同时包含字母、数字、特殊字符中的至少两项，且长度为 6-50 位")
+        return value
+
+    @model_validator(mode="after")
+    def passwords_match(self) -> "InitializationRequest":
+        """确保两次输入的密码完全一致。"""
+        if self.password != self.confirm_password:
+            raise ValueError("两次输入的密码不一致")
+        return self

app/schemas/token.py

@@ -31,10 +31,6 @@ class Token(BaseModel):
     level: int = 1
     # 详细权限
     permissions: Optional[UserPermissions] = Field(default_factory=dict)
-    # 是否显示配置向导
-    wizard: Optional[bool] = None
-
-
 class TokenPayload(BaseModel):
     """访问令牌中携带的用户身份与授权信息。"""
 

app/schemas/types.py

@@ -420,8 +420,6 @@ class SystemConfigKey(Enum):
     ScrapingSwitchs = "ScrapingSwitchs"
     # 插件安装统计
     PluginInstallReport = "PluginInstallReport"
-    # 配置向导状态
-    SetupWizardState = "SetupWizardState"
     # 绿联影视登录会话缓存
     UgreenSessionCache = "UgreenSessionCache"
     # 共享媒体识别成功次数

app/startup/composition/configuration.py

@@ -112,7 +112,6 @@ def normalize_subscribe_rss_interval(value: object) -> int:
 def build_api_runtime_config(settings: Settings) -> ApiRuntimeConfig:
     """从可热更新的部署设置构建一次 API 请求配置快照。"""
     return ApiRuntimeConfig(
-        advanced_mode=settings.ADVANCED_MODE,
         access_token_expire_minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
         btrfs_fsid_dedup=settings.BTRFS_FSID_DEDUP,
         ai_agent_enable=settings.AI_AGENT_ENABLE,

database/versions/294b007932ef_2_0_0.py

@@ -6,13 +6,8 @@ Create Date: 2024-07-20 08:43:40.741251
 
 """
 
-import secrets
-
 from alembic import op
 import sqlalchemy as sa
-from app.runtime.config import settings
-from app.application.security.token import get_password_hash
-from app.runtime.log import logger
 
 # revision identifiers, used by Alembic.
 revision = '294b007932ef'
@@ -26,46 +21,6 @@ def upgrade() -> None:
     v2.0.0 数据库初始化
     """
     connection = op.get_bind()
-    user = sa.table(
-        "user",
-        sa.column("name", sa.String()),
-        sa.column("email", sa.String()),
-        sa.column("hashed_password", sa.String()),
-        sa.column("is_active", sa.Boolean()),
-        sa.column("is_superuser", sa.Boolean()),
-        sa.column("avatar", sa.String()),
-        sa.column("is_otp", sa.Boolean()),
-        sa.column("otp_secret", sa.String()),
-        sa.column("permissions", sa.JSON()),
-        sa.column("settings", sa.JSON()),
-    )
-    # 初始化超级管理员
-    existing_user = connection.execute(
-        sa.select(user.c.name).where(user.c.name == settings.SUPERUSER)
-    ).first()
-    if not existing_user:
-        if settings.SUPERUSER_PASSWORD:
-            init_password = settings.SUPERUSER_PASSWORD
-        else:
-            # 生成随机密码
-            init_password = secrets.token_urlsafe(16)
-            logger.info(
-                f"【超级管理员初始密码】{init_password} 请登录系统后在设定中修改。 注：该密码只会显示一次，请注意保存。")
-        connection.execute(
-            user.insert().values(
-                name=settings.SUPERUSER,
-                hashed_password=get_password_hash(init_password),
-                email="admin@movie-pilot.org",
-                is_active=True,
-                is_superuser=True,
-                avatar="",
-                is_otp=False,
-                otp_secret=None,
-                permissions={},
-                settings={},
-            )
-        )
-
     # 初始化本地存储
     systemconfig = sa.table(
         "systemconfig",

scripts/local_setup.py

@@ -2573,7 +2573,6 @@ def _apply_local_system_config_inner(config_payload: dict[str, Any]) -> None:
         except Exception as exc:
             print_step(f"已保存站点认证配置，当前未完成校验：{exc}")
 
-    system_config.set(SystemConfigKey.SetupWizardState, True)
     print_step("已写入本地系统配置")
 
 

tests/test_agent_api_lazy_imports.py

@@ -139,7 +139,7 @@ from app.application.configuration import ApiRuntimeConfig
 from app.runtime.config import settings
 
 runtime_config = ApiRuntimeConfig(
-    False, 60, False, settings.AI_AGENT_ENABLE,
+    60, False, settings.AI_AGENT_ENABLE,
     api_token=settings.API_TOKEN,
 )
 anthropic.get_api_runtime_config_snapshot = lambda: runtime_config
@@ -382,7 +382,7 @@ from app.runtime.config import settings
 
 settings.AI_AGENT_ENABLE = True
 runtime_config = ApiRuntimeConfig(
-    False, 60, False, True,
+    60, False, True,
     api_token=settings.API_TOKEN,
 )
 anthropic.get_api_runtime_config_snapshot = lambda: runtime_config

tests/test_api_authorization.py

@@ -187,24 +187,11 @@ def test_login_sets_resource_token_cookie(monkeypatch):
                 permissions={"discovery": True, "features": {}},
             )
 
-    class FakeSystemConfigOper:
-        """返回已完成向导状态的系统配置桩。"""
-
-        def get(self, key):
-            """返回测试配置值。"""
-            return "1"
-
     form_data = SimpleNamespace(username="user", password="password")
     request = _build_request()
     response = Response()
 
     monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain)
-    monkeypatch.setattr(
-        login_endpoint,
-        "get_configured_system_config",
-        lambda: FakeSystemConfigOper(),
-    )
-
     token = login_endpoint.login_access_token(
         request=request,
         response=response,

tests/test_configuration_ports.py

@@ -146,7 +146,6 @@ def test_api_runtime_provider_returns_frozen_snapshot_per_request() -> None:
     configure_runtime_configuration(
         RuntimeConfiguration(
             api=lambda: ApiRuntimeConfig(
-                advanced_mode=False,
                 access_token_expire_minutes=60,
                 btrfs_fsid_dedup=False,
                 ai_agent_enable=state["enabled"],

tests/test_database_migration_startup.py

@@ -563,10 +563,8 @@ def test_migration_config_write_rolls_back_with_alembic_transaction(
         assert stored_templates == original_templates
 
 
-def test_initial_migration_rolls_back_user_and_storages_together(
-        monkeypatch,
-) -> None:
-    """2.0.0 管理员与存储初始化必须共享 Alembic 事务。"""
+def test_initial_migration_does_not_seed_user_and_initializes_storages(monkeypatch) -> None:
+    """2.0.0 初始化只准备存储，管理员由首次访问页面创建。"""
     migration = importlib.import_module(
         "database.versions.294b007932ef_2_0_0"
     )
@@ -597,50 +595,18 @@ def test_initial_migration_rolls_back_user_and_storages_together(
     )
     metadata.create_all(engine)
 
-    monkeypatch.setattr(migration.settings, "SUPERUSER", "migration-admin")
-    monkeypatch.setattr(
-        migration.settings,
-        "SUPERUSER_PASSWORD",
-        "migration-password",
-    )
-    monkeypatch.setattr(
-        migration,
-        "get_password_hash",
-        lambda password: f"hashed:{password}",
-    )
-
     with engine.connect() as connection:
-        transaction = connection.begin()
         monkeypatch.setattr(
             migration,
             "op",
             Operations(MigrationContext.configure(connection)),
         )
-
-        def fail_storages_write(
-                _connection,
-                _cursor,
-                statement,
-                _parameters,
-                _context,
-                _executemany,
-        ) -> None:
-            if statement.lstrip().upper().startswith("INSERT INTO SYSTEMCONFIG"):
-                raise RuntimeError("injected storages failure")
-
-        event.listen(engine, "after_cursor_execute", fail_storages_write)
-        try:
-            with pytest.raises(RuntimeError, match="injected storages failure"):
-                migration.upgrade()
-        finally:
-            event.remove(engine, "after_cursor_execute", fail_storages_write)
-            transaction.rollback()
+        migration.upgrade()
+        connection.commit()
 
     with engine.connect() as connection:
         assert connection.execute(text("SELECT COUNT(*) FROM user")).scalar_one() == 0
-        assert connection.execute(
-            text("SELECT COUNT(*) FROM systemconfig")
-        ).scalar_one() == 0
+        assert connection.execute(text("SELECT value FROM systemconfig WHERE key = 'Storages'")).scalar_one()
 
 
 def test_userconfig_cleanup_migration_uses_alembic_transaction(

tests/test_history_ai_retry_gate.py

@@ -61,7 +61,6 @@ async def _record_async(target: list[dict], payload: dict) -> None:
 def _runtime(*, ai_enabled: bool = True) -> ApiRuntimeConfig:
     """构造历史端点需要的最小稳定配置快照。"""
     return ApiRuntimeConfig(
-        advanced_mode=False,
         access_token_expire_minutes=30,
         btrfs_fsid_dedup=False,
         ai_agent_enable=ai_enabled,

tests/test_host_runtime_context.py

@@ -205,7 +205,7 @@ def _runtime() -> HostRuntime:
             system_config=lambda: _Repository(object()),
         ),
         configuration=RuntimeConfiguration(
-            api=lambda: ApiRuntimeConfig(False, 60, False, True),
+            api=lambda: ApiRuntimeConfig(60, False, True),
             scheduler=lambda: SchedulerRuntimeConfig(
                 False,
                 "Asia/Shanghai",

tests/test_initialization.py

@@ -0,0 +1,96 @@
+"""首次初始化 API 的业务契约测试。"""
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+
+from app.api.endpoints import login as login_endpoint
+from app.schemas.initialization import InitializationRequest
+
+
+class _FakeRuntimeSettings:
+    """记录首次初始化对部署设置的更新与回滚。"""
+
+    def __init__(self) -> None:
+        """初始化空设置快照。"""
+        self.values = {"SUPERUSER": "", "API_TOKEN": None}
+        self.updates: list[tuple[str, object]] = []
+
+    def get(self, key: str, default=None):
+        """读取测试设置。"""
+        return self.values.get(key, default)
+
+    def update(self, key: str, value: object):
+        """记录并应用测试设置。"""
+        self.updates.append((key, value))
+        self.values[key] = value
+        return True, ""
+
+
+class _FakeUserService:
+    """提供初始化接口所需的最小异步用户端口。"""
+
+    def __init__(self, initialized: bool = False) -> None:
+        """保存初始用户状态。"""
+        self.initialized = initialized
+        self.created: list[dict] = []
+
+    async def is_initialized(self) -> bool:
+        """返回测试用户状态。"""
+        return self.initialized
+
+    async def create(self, payload: dict):
+        """记录创建的管理员。"""
+        self.created.append(payload)
+        self.initialized = True
+        return SimpleNamespace(id=1)
+
+
+def _payload() -> InitializationRequest:
+    """构造有效初始化请求。"""
+    return InitializationRequest(
+        username="admin",
+        password="Admin123!",
+        confirm_password="Admin123!",
+        api_key="a" * 32,
+    )
+
+
+def test_initialization_status_reports_existing_users():
+    """状态接口只暴露是否已有用户，不暴露用户名或 API Key。"""
+    service = _FakeUserService(initialized=True)
+
+    response = asyncio.run(login_endpoint.get_initialization_status(service))
+
+    assert response.success is True
+    assert response.data.initialized is True
+    assert not hasattr(response.data, "api_key")
+
+
+def test_initialize_instance_updates_settings_and_creates_superuser(monkeypatch):
+    """首次初始化应先保存部署凭据，再创建唯一超级管理员。"""
+    service = _FakeUserService()
+    settings = _FakeRuntimeSettings()
+    monkeypatch.setattr(login_endpoint, "get_runtime_settings", lambda: settings)
+    monkeypatch.setattr(login_endpoint, "get_password_hash", lambda password: "hashed:" + password)
+
+    response = asyncio.run(login_endpoint.initialize_instance(_payload(), service))
+
+    assert response.success is True
+    assert settings.updates == [("SUPERUSER", "admin"), ("API_TOKEN", "a" * 32)]
+    assert service.created[0]["hashed_password"] == "hashed:Admin123!"
+    assert service.created[0]["is_superuser"] is True
+
+
+def test_initialize_instance_rejects_second_claim(monkeypatch):
+    """已经存在用户时，第二次初始化必须返回冲突而不是覆盖账号。"""
+    service = _FakeUserService(initialized=True)
+    monkeypatch.setattr(login_endpoint, "get_runtime_settings", lambda: _FakeRuntimeSettings())
+
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(login_endpoint.initialize_instance(_payload(), service))
+
+    assert error.value.status_code == 409
+    assert service.created == []