fix(auth): restore API token compatibility

app/adapters/web/security/access.py

@@ -113,9 +113,10 @@ def _decode_token(token: str | None, purpose: str) -> TokenPayload:
 
 def _get_api_token(
     token_query: Annotated[str | None, Security(api_token_query)] = None,
+    key_header: Annotated[str | None, Security(api_key_header)] = None,
 ) -> str | None:
-    """从 URL 查询参数读取兼容 API Token。"""
+    """优先从请求头、其次从查询参数读取兼容 API Token。"""
-    return token_query
+    return key_header or token_query
 
 
 def _get_api_key(
@@ -314,7 +315,7 @@ def _verify_key(key: str | None, expected_key: str, key_type: str) -> str:
 def verify_apitoken(
     token: Annotated[str | None, Security(_get_api_token)],
 ) -> str:
-    """校验 URL 查询参数中的兼容 API Token。"""
+    """校验请求头或 URL 查询参数中的兼容 API Token。"""
     value = _verify_key(token, get_runtime_setting("API_TOKEN"), "token")
     validate_api_credential_identity()
     return value

app/api/endpoints/message.py

@@ -123,7 +123,7 @@ async def user_message(
     _: _SchemaTokenPayload = Depends(verify_apitoken),
 ):
     """
-    用户消息响应，配置请求中需要添加参数：token=API_TOKEN&source=消息配置名
+    用户消息响应；推荐通过 X-API-KEY 请求头传递 API_TOKEN，查询参数 token 仅保留兼容。
     """
     body = await request.body()
     form = await request.form()

app/application/messaging/ingress.py

@@ -18,6 +18,7 @@ class _MessageIngressRequest:
 
     url: str
     payload: Mapping[str, Any]
+    headers: Mapping[str, str]
     source: str
     timeout: float
 
@@ -30,6 +31,7 @@ class MessageIngressPort(Protocol):
         url: str,
         payload: Mapping[str, Any],
         *,
+        headers: Mapping[str, str],
         timeout: float,
     ) -> Optional[int]:
         """同步投递 payload，并返回 HTTP 状态码或 None。"""
@@ -40,6 +42,7 @@ class MessageIngressPort(Protocol):
         url: str,
         payload: Mapping[str, Any],
         *,
+        headers: Mapping[str, str],
         timeout: float,
     ) -> Optional[int]:
         """异步投递 payload，并返回 HTTP 状态码或 None。"""
@@ -78,14 +81,17 @@ def _message_ingress_snapshot() -> MessageIngressPort:
 
 
 def build_message_ingress_url(source: str | None) -> str:
-    """按当前运行配置构造安全编码的本地消息入口 URL。"""
+    """构造仅含非敏感来源参数的本地消息入口 URL。"""
-    query = {"token": get_runtime_setting('API_TOKEN')}
+    query: dict[str, str] = {}
     if source:
         query["source"] = source
-    return (
+    base_url = f"http://127.0.0.1:{get_runtime_setting('PORT')}/api/v1/message"
-        f"http://127.0.0.1:{get_runtime_setting('PORT')}/api/v1/message?"
+    return f"{base_url}?{urlencode(query)}" if query else base_url
-        f"{urlencode(query)}"
+
-    )
+
+def build_message_ingress_headers() -> dict[str, str]:
+    """把本地回环凭据放入请求头，避免访问日志记录明文 Token。"""
+    return {"X-API-KEY": str(get_runtime_setting("API_TOKEN") or "")}
 
 
 def _message_ingress_request(
@@ -97,6 +103,7 @@ def _message_ingress_request(
     return _MessageIngressRequest(
         url=build_message_ingress_url(source),
         payload=dict(payload),
+        headers=build_message_ingress_headers(),
         source=source or "-",
         timeout=timeout,
     )
@@ -142,6 +149,7 @@ def forward_message_to_host(
         status_code = _message_ingress_snapshot().post(
             request.url,
             request.payload,
+            headers=request.headers,
             timeout=request.timeout,
         )
         return _message_ingress_confirmed(request, status_code)
@@ -161,6 +169,7 @@ async def async_forward_message_to_host(
         status_code = await _message_ingress_snapshot().async_post(
             request.url,
             request.payload,
+            headers=request.headers,
             timeout=request.timeout,
         )
         return _message_ingress_confirmed(request, status_code)

app/application/security/auth.py

@@ -154,6 +154,9 @@ class AuthUserRepository(Protocol):
     def get_by_id(self, user_id: int) -> Optional[AuthUser]:
         """按 ID 查询用户。"""
 
+    def get_active_superuser(self) -> Optional[AuthUser]:
+        """返回按稳定顺序选出的启用超级管理员。"""
+
 
 class AuthPasskeyRepository(Protocol):
     """认证提供方查询端口。"""
@@ -195,7 +198,9 @@ class AuthService:
 
     def build_superuser_token_payload(self) -> _SchemaTokenPayload:
         """从持久化用户和站点认证状态构造超级用户令牌载荷。"""
-        configured_name = get_chain_runtime_config_snapshot().superuser
+        configured_name = str(
+            get_chain_runtime_config_snapshot().superuser or ""
+        ).strip()
         if (
             self._superuser_binding_id is not None
             and configured_name == self._superuser_binding_name
@@ -203,12 +208,22 @@ class AuthService:
             # 配置保存用户名；持久化 ID 保证管理员改名不会让管理员级集成失效。
             user = self._users.get_by_id(self._superuser_binding_id)
         else:
-            user = self._users.get_by_name(configured_name)
+            user = (
+                self._users.get_by_name(configured_name)
+                if configured_name
+                else self._users.get_active_superuser()
+            )
             if user:
                 self._superuser_binding_name = configured_name
                 self._superuser_binding_id = user.id
         if not user or not user.is_active or not user.is_superuser:
-            raise PermissionError("用户权限不足")
+            if not configured_name:
+                raise PermissionError(
+                    "未配置 SUPERUSER，且数据库中没有可用超级管理员"
+                )
+            raise PermissionError(
+                "SUPERUSER 对应用户不存在、未启用或非超级管理员"
+            )
         return _SchemaTokenPayload(
             sub=user.id,
             username=user.name,

app/cli.py

@@ -17,9 +17,12 @@ import click
 import psutil
 
 from app.application.backup import BackupArtifact
-from app.application.configuration import get_runtime_settings
 from app.runtime.config import Settings
-from app.runtime.settings import get_runtime_setting
+from app.runtime.settings import (
+    get_runtime_setting,
+    has_runtime_setting,
+    update_runtime_setting,
+)
 from app.runtime.state import SystemHelper
 from app.runtime.version import get_app_version, get_frontend_version
 from app.startup.composition.database import build_database_governance
@@ -801,7 +804,9 @@ def _ensure_local_api_token() -> bool:
     if get_runtime_setting("API_TOKEN") and len(str(get_runtime_setting("API_TOKEN")).strip()) >= 16:
         return False
 
-    result, message = get_runtime_settings().update("API_TOKEN", get_runtime_setting("API_TOKEN") or "")
+    result, message = update_runtime_setting(
+        "API_TOKEN", get_runtime_setting("API_TOKEN") or ""
+    )
     if result is False:
         raise click.ClickException(message or "初始化 API_TOKEN 失败")
     return result is True
@@ -1324,7 +1329,7 @@ def config_path() -> None:
 @click.option("--show-secrets", is_flag=True, help="显示敏感配置原文")
 def config_list(show_secrets: bool) -> None:
     """列出当前配置"""
-    values = get_runtime_settings().snapshot()
+    values = {key: get_runtime_setting(key) for key in Settings.model_fields}
     for key in sorted(values):
         click.echo(f"{key}={_format_value(_mask_value(key, values[key], show_secrets))}")
 
@@ -1334,9 +1339,9 @@ def config_list(show_secrets: bool) -> None:
 def config_get(key: str) -> None:
     """读取单个配置项"""
     setting_fields = Settings.model_fields.keys()
-    if key not in setting_fields and not get_runtime_settings().contains(key):
+    if key not in setting_fields and not has_runtime_setting(key):
         raise click.ClickException(f"配置项不存在：{key}")
-    click.echo(_format_value(get_runtime_settings().get(key)))
+    click.echo(_format_value(get_runtime_setting(key)))
 
 
 @config.command("set", context_settings=CONTEXT_SETTINGS)
@@ -1344,7 +1349,7 @@ def config_get(key: str) -> None:
 @click.argument("value")
 def config_set(key: str, value: str) -> None:
     """写入单个配置项"""
-    result, message = get_runtime_settings().update(key, value)
+    result, message = update_runtime_setting(key, value)
     if result is False:
         raise click.ClickException(message or f"配置项更新失败：{key}")
     if result is None:
@@ -1376,7 +1381,7 @@ def config_keys(pattern: Optional[str], show_current: bool, show_secrets: bool)
         if pattern and pattern.lower() not in key.lower():
             continue
         default_value = _field_default(field)
-        current_value = get_runtime_settings().get(key, default_value)
+        current_value = get_runtime_setting(key, default_value)
         rows.append(
             (
                 key,
@@ -1408,7 +1413,7 @@ def config_describe(key: str, show_secrets: bool) -> None:
         raise click.ClickException(f"配置项不存在：{key}")
 
     default_value = _field_default(field)
-    current_value = get_runtime_settings().get(key, default_value)
+    current_value = get_runtime_setting(key, default_value)
     click.echo(f"Key: {key}")
     click.echo(f"Type: {_annotation_name(field.annotation)}")
     click.echo(f"Default: {_format_value(_mask_value(key, default_value, show_secrets))}")

app/db/adapters/user.py

@@ -69,6 +69,17 @@ class SqlAlchemyUserRepository(UserRepository):
         model = self._oper.get_by_id(user_id)
         return _to_snapshot(model) if model else None
 
+    def get_active_superuser(self) -> Optional[UserSnapshot]:
+        """按主键顺序返回首个启用的超级管理员快照。"""
+        session = cast(Session, self._session)
+        model = session.execute(
+            select(User)
+            .where(User.is_active.is_(True), User.is_superuser.is_(True))
+            .order_by(User.id)
+            .limit(1)
+        ).scalars().first()
+        return _to_snapshot(model) if model else None
+
     async def async_has_users(self) -> bool:
         """使用最小列查询判断数据库中是否已有用户。"""
         session = self._require_async_session()
@@ -225,6 +236,11 @@ class TransactionalUserRepository(ChainUserRepository):
         with self._sync_session() as session:
             return SqlAlchemyUserRepository(session).get_by_id(user_id)
 
+    def get_active_superuser(self) -> Optional[UserSnapshot]:
+        """在独立会话中返回首个启用的超级管理员快照。"""
+        with self._sync_session() as session:
+            return SqlAlchemyUserRepository(session).get_active_superuser()
+
     def get_auth_by_name(self, name: str) -> Optional[UserAuthSnapshot]:
         """按用户名读取认证凭据快照。"""
         with self._sync_session() as session:

app/startup/composition/network.py

@@ -128,10 +128,14 @@ class _MessageIngressAdapter:
         url: str,
         payload: Mapping[str, Any],
         *,
+        headers: Mapping[str, str],
         timeout: float,
     ) -> Optional[int]:
         """同步投递消息，关闭响应后返回状态码。"""
-        response = RequestUtils(timeout=timeout).post_res(  # type: ignore[arg-type]
+        response = RequestUtils(
+            timeout=timeout,  # type: ignore[arg-type]
+            headers=dict(headers),
+        ).post_res(
             url,
             json=dict(payload),
         )
@@ -150,11 +154,13 @@ class _MessageIngressAdapter:
         url: str,
         payload: Mapping[str, Any],
         *,
+        headers: Mapping[str, str],
         timeout: float,
     ) -> Optional[int]:
         """异步投递消息，关闭响应后返回状态码。"""
         response = await AsyncRequestUtils(
-            timeout=timeout  # type: ignore[arg-type]
+            timeout=timeout,  # type: ignore[arg-type]
+            headers=dict(headers),
         ).post_res(
             url,
             json=dict(payload),

app/startup/composition/security.py

@@ -12,6 +12,7 @@ from app.adapters.web.security.access import (
 from app.application.configuration import get_configured_system_config
 from app.application.security.auth import (
     AuthService,
+    AuthUserRepository,
     build_superuser_token_payload,
     configure_auth_service,
     reset_auth_service,
@@ -30,6 +31,8 @@ from app.db.adapters.user import SqlAlchemyUserRepository
 from app.db.oper.passkey import PassKeyOper
 from app.db.oper.systemconfig import SystemConfigOper
 from app.runtime.cache import TTLCache
+from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting, update_runtime_setting
 from app.startup.composition.context import (
     RepositoryFactory,
     StandaloneRepositoryFactory,
@@ -48,8 +51,27 @@ class SecurityComposition:
     passkey: StandaloneRepositoryFactory
 
 
+def _backfill_superuser_setting(users: AuthUserRepository) -> None:
+    """用现有数据库管理员补全 V2 升级后缺失的 SUPERUSER。"""
+    if str(get_runtime_setting("SUPERUSER") or "").strip():
+        return
+    user = users.get_active_superuser()
+    if user is None:
+        return
+    success, message = update_runtime_setting("SUPERUSER", user.name)
+    if success is False:
+        logger.warning(
+            f"检测到数据库超级管理员 {user.name}，但自动补全 SUPERUSER 失败："
+            f"{message or '未知错误'}"
+        )
+        return
+    logger.info(f"已根据数据库超级管理员自动补全 SUPERUSER：{user.name}")
+
+
 def configure_security_services() -> SecurityComposition:
     """构造并登记认证、用户查询和 PassKey 服务。"""
+    users = build_transactional_user_repository()
+    _backfill_superuser_setting(users)
     configure_user_lookups(
         by_id=lambda user_id: build_transactional_user_repository().get_by_id(user_id),
         by_name=lambda username: build_transactional_user_repository().get_by_name(username),
@@ -57,7 +79,7 @@ def configure_security_services() -> SecurityComposition:
     )
     configure_auth_service(
         AuthService(
-            users=build_transactional_user_repository(),
+            users=users,
             config=get_configured_system_config(),
             passkeys=PassKeyOper(),
         )

tests/test_auth_contracts.py

@@ -360,6 +360,12 @@ def test_standalone_api_credential_revalidates_current_identity(monkeypatch, dep
     assert exc_info.value.headers == {"WWW-Authenticate": "Bearer"}
 
 
+def test_api_token_reader_prefers_header_and_preserves_query_compatibility():
+    """兼容 Token 依赖应优先使用请求头，同时继续接受旧查询参数。"""
+    assert access._get_api_token("query-token", "header-token") == "header-token"
+    assert access._get_api_token("query-token", None) == "query-token"
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("endpoint", "credential"),
@@ -475,6 +481,52 @@ def test_superuser_payload_provider_rejects_inactive_user(monkeypatch):
         service.build_superuser_token_payload()
 
 
+def test_superuser_payload_provider_falls_back_to_database_admin(monkeypatch):
+    """V2 升级后 SUPERUSER 为空时，API 凭据应绑定现有启用管理员。"""
+    user = _user(active=True, superuser=True)
+    users = SimpleNamespace(
+        get_by_name=Mock(),
+        get_active_superuser=Mock(return_value=user),
+    )
+    service = AuthService(
+        users=users,
+        config=SimpleNamespace(),
+        passkeys=SimpleNamespace(),
+    )
+    monkeypatch.setattr(
+        auth_service_module,
+        "get_chain_runtime_config_snapshot",
+        lambda: SimpleNamespace(superuser=""),
+    )
+
+    payload = service.build_superuser_token_payload()
+
+    assert payload.sub == user.id
+    assert payload.username == user.name
+    users.get_by_name.assert_not_called()
+    users.get_active_superuser.assert_called_once_with()
+
+
+def test_superuser_payload_provider_explains_missing_binding(monkeypatch):
+    """配置和数据库都没有管理员时，应返回可操作的认证失败原因。"""
+    service = AuthService(
+        users=SimpleNamespace(get_active_superuser=Mock(return_value=None)),
+        config=SimpleNamespace(),
+        passkeys=SimpleNamespace(),
+    )
+    monkeypatch.setattr(
+        auth_service_module,
+        "get_chain_runtime_config_snapshot",
+        lambda: SimpleNamespace(superuser=""),
+    )
+
+    with pytest.raises(
+        PermissionError,
+        match="未配置 SUPERUSER，且数据库中没有可用超级管理员",
+    ):
+        service.build_superuser_token_payload()
+
+
 def test_auth_service_reads_user_by_id_and_accepts_current_token_identity():
     """认证服务按稳定用户 ID 查询，并接受与当前账号一致的令牌声明。"""
     user = _user()

tests/test_cli_configuration.py

@@ -0,0 +1,63 @@
+"""离线配置 CLI 的运行时端口契约测试。"""
+
+import pytest
+from click.testing import CliRunner
+
+from app import cli as cli_module
+
+
+def test_config_get_works_without_web_lifespan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """配置读取必须使用启动前可用的 runtime 端口，不能依赖 Web 组合根。"""
+    monkeypatch.setattr(cli_module, "has_runtime_setting", lambda _key: True)
+    monkeypatch.setattr(
+        cli_module,
+        "get_runtime_setting",
+        lambda key, *_args: "admin" if key == "SUPERUSER" else None,
+    )
+
+    result = CliRunner().invoke(
+        cli_module.cli,
+        ["config", "get", "SUPERUSER"],
+    )
+
+    assert result.exit_code == 0
+    assert result.output == "admin\n"
+
+
+def test_config_set_works_without_web_lifespan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """配置写入必须走离线可用的 runtime 更新端口。"""
+    updates: list[tuple[str, str]] = []
+
+    def update_setting(key: str, value: str) -> tuple[bool, str]:
+        """记录离线配置写入。"""
+        updates.append((key, value))
+        return True, ""
+
+    monkeypatch.setattr(
+        cli_module,
+        "update_runtime_setting",
+        update_setting,
+    )
+    monkeypatch.setattr(
+        cli_module,
+        "_managed_backend_status",
+        lambda: ("stopped", None, None, None),
+    )
+    monkeypatch.setattr(
+        cli_module,
+        "_managed_frontend_status",
+        lambda: ("stopped", None, None, None),
+    )
+
+    result = CliRunner().invoke(
+        cli_module.cli,
+        ["config", "set", "SUPERUSER", "admin"],
+    )
+
+    assert result.exit_code == 0
+    assert updates == [("SUPERUSER", "admin")]
+    assert result.output == "SUPERUSER 已更新\n"

tests/test_message_ingress.py

@@ -28,14 +28,14 @@ class _FakeMessageIngressPort:
         self.sync_calls = []
         self.async_calls = []
 
-    def post(self, url, payload, *, timeout):
+    def post(self, url, payload, *, headers, timeout):
         """记录同步投递并返回固定状态码。"""
-        self.sync_calls.append((url, payload, timeout))
+        self.sync_calls.append((url, payload, headers, timeout))
         return self.status_code
 
-    async def async_post(self, url, payload, *, timeout):
+    async def async_post(self, url, payload, *, headers, timeout):
         """记录异步投递并返回固定状态码。"""
-        self.async_calls.append((url, payload, timeout))
+        self.async_calls.append((url, payload, headers, timeout))
         return self.status_code
 
 
@@ -57,8 +57,8 @@ def _patch_ingress_settings(monkeypatch, **values):
     )
 
 
-def test_forward_message_to_host_encodes_source_and_copies_payload(monkeypatch):
+def test_forward_message_to_host_keeps_token_out_of_url(monkeypatch):
-    """统一入口必须安全编码查询参数，并向端口传递 payload 副本。"""
+    """统一入口仅编码来源参数，并通过请求头传递凭据和 payload 副本。"""
     port = _FakeMessageIngressPort()
     ingress.configure_message_ingress_port(port)
     _patch_ingress_settings(monkeypatch, PORT=3000, API_TOKEN="token value")
@@ -70,12 +70,12 @@ def test_forward_message_to_host_encodes_source_and_copies_payload(monkeypatch):
         timeout=9,
     ) is True
 
-    url, forwarded, timeout = port.sync_calls[0]
+    url, forwarded, headers, timeout = port.sync_calls[0]
     assert urlparse(url).path == "/api/v1/message"
     assert parse_qs(urlparse(url).query) == {
-        "token": ["token value"],
         "source": ["channel & one"],
     }
+    assert headers == {"X-API-KEY": "token value"}
     assert forwarded == {"text": "hello"}
     assert forwarded is not payload
     assert timeout == 9
@@ -114,11 +114,11 @@ async def test_async_forward_message_to_host_uses_same_contract(monkeypatch):
         timeout=10,
     ) is True
 
-    url, payload, timeout = port.async_calls[0]
+    url, payload, headers, timeout = port.async_calls[0]
     assert parse_qs(urlparse(url).query) == {
-        "token": ["token value"],
         "source": ["discord & one"],
     }
+    assert headers == {"X-API-KEY": "token value"}
     assert payload == {"text": "hello"}
     assert timeout == 10
 
@@ -139,10 +139,15 @@ def test_startup_message_ingress_adapter_closes_sync_response(monkeypatch):
     status_code = network_composition._MessageIngressAdapter().post(
         "http://127.0.0.1/message",
         {"text": "hello"},
+        headers={"X-API-KEY": "secret"},
         timeout=9,
     )
 
     assert status_code == 202
+    network_composition.RequestUtils.assert_called_once_with(
+        timeout=9,
+        headers={"X-API-KEY": "secret"},
+    )
     response.close.assert_called_once_with()
 
 
@@ -163,10 +168,15 @@ async def test_startup_message_ingress_adapter_closes_async_response(monkeypatch
     status_code = await network_composition._MessageIngressAdapter().async_post(
         "http://127.0.0.1/message",
         {"text": "hello"},
+        headers={"X-API-KEY": "secret"},
         timeout=10,
     )
 
     assert status_code == 202
+    network_composition.AsyncRequestUtils.assert_called_once_with(
+        timeout=10,
+        headers={"X-API-KEY": "secret"},
+    )
     response.aclose.assert_awaited_once_with()
 
 

tests/test_security_composition.py

@@ -0,0 +1,60 @@
+"""认证组合根的部署兼容修复测试。"""
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from app.startup.composition import security as security_composition
+
+
+def test_backfill_superuser_setting_from_existing_database_admin(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """V2 数据库已有管理员而配置缺失时，应持久化稳定管理员用户名。"""
+    updates: list[tuple[str, str]] = []
+    users = SimpleNamespace(
+        get_active_superuser=Mock(
+            return_value=SimpleNamespace(name="legacy-admin")
+        )
+    )
+    monkeypatch.setattr(
+        security_composition,
+        "get_runtime_setting",
+        lambda _key: "",
+    )
+
+    def update_setting(key: str, value: str) -> tuple[bool, str]:
+        """记录启动兼容修复写入。"""
+        updates.append((key, value))
+        return True, ""
+
+    monkeypatch.setattr(
+        security_composition,
+        "update_runtime_setting",
+        update_setting,
+    )
+
+    security_composition._backfill_superuser_setting(users)
+
+    assert updates == [("SUPERUSER", "legacy-admin")]
+    users.get_active_superuser.assert_called_once_with()
+
+
+def test_backfill_superuser_setting_preserves_explicit_configuration(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """显式 SUPERUSER 必须保持原值，不得被数据库顺序覆盖。"""
+    users = SimpleNamespace(get_active_superuser=Mock())
+    update = Mock()
+    monkeypatch.setattr(
+        security_composition,
+        "get_runtime_setting",
+        lambda _key: "configured-admin",
+    )
+    monkeypatch.setattr(security_composition, "update_runtime_setting", update)
+
+    security_composition._backfill_superuser_setting(users)
+
+    users.get_active_superuser.assert_not_called()
+    update.assert_not_called()

tests/test_user_repository.py

@@ -160,6 +160,21 @@ async def test_user_snapshots_are_detached_and_deeply_frozen(user_repository) ->
         public.permissions["features"]["search"] = False  # type: ignore[index]
 
 
+def test_user_repository_selects_first_active_superuser(user_repository) -> None:
+    """升级绑定只能选择启用管理员，并按主键保持确定顺序。"""
+    repository, sync_factory = user_repository
+    _insert_user(sync_factory, name="disabled-admin", is_active=False)
+    expected_id = _insert_user(sync_factory, name="first-admin")
+    _insert_user(sync_factory, name="second-admin")
+    _insert_user(sync_factory, name="member", is_superuser=False)
+
+    selected = repository.get_active_superuser()
+
+    assert selected is not None
+    assert selected.id == expected_id
+    assert selected.name == "first-admin"
+
+
 def test_auxiliary_create_commits_before_return(user_repository) -> None:
     """辅助认证创建成功返回时，新用户必须已对后续独立会话可见。"""
     repository, sync_factory = user_repository