feat: platform.db 模型与超管引导（WAL/scrypt/幂等bootstrap）

Co-Authored-By: Claude Code <noreply@anthropic.com>

app/engine/config.py

@@ -38,6 +38,8 @@ class Settings(BaseSettings):
     PUSH_MAX_ATTEMPTS: int = 3
     PUSH_RETRY_BACKOFF_BASE: float = 1.0   # 尝试间隔 1s/2s(3 次尝试、2 次间隔)
 
+    # Phase 2 超管引导(lifespan bootstrap,空表才建):初始密码须首登改掉
+    ADMIN_USERNAME: str = "admin"
     ADMIN_PASSWORD: str = "admin123"
 
     model_config = {

app/main.py

@@ -5,7 +5,12 @@ import logging
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 
-from app.engine.config import STATIC_DIR
+from app.engine.config import STATIC_DIR, settings
+from app.platform.db import (
+    bootstrap_superuser,
+    ensure_platform_schema,
+    make_platform_engine,
+)
 from app.platform.host_middleware import TenantHostMiddleware
 from app.platform.registry import TenantRegistry
 
@@ -15,6 +20,7 @@ logger = logging.getLogger(__name__)
 BASE_DIR = Path(__file__).resolve().parents[1]
 SEED_DIR = BASE_DIR / "seed_profiles"
 TENANTS_DIR = BASE_DIR / "tenants"        # .gitignore 已忽略 tenants/*.db
+PLATFORM_DB = BASE_DIR / "platform.db"    # .gitignore 已忽略 platform.db*
 
 
 @asynccontextmanager
@@ -26,7 +32,22 @@ async def lifespan(app: FastAPI):
         registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=TENANTS_DIR)
         app.state.registry = registry
 
+    # 平台库测试缝(Phase 2 Task 1,同 registry 缝模式):预置过
+    # app.state.platform_engine(临时库)则沿用——不预置则每个 TestClient
+    # 测试都会在仓库根建出真实 platform.db。schema+bootstrap 无条件跑
+    # (两者幂等):预置引擎同样被建到可用,Task 2 起认证测试直接可用。
+    platform_engine = getattr(app.state, "platform_engine", None)
+    if platform_engine is None:
+        platform_engine = make_platform_engine(PLATFORM_DB)
+        app.state.platform_engine = platform_engine
+
     try:
+        await ensure_platform_schema(platform_engine)
+        await bootstrap_superuser(
+            platform_engine,
+            username=settings.ADMIN_USERNAME,
+            password=settings.ADMIN_PASSWORD)
+
         # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed。逐租户、
         # 用该租户自己的 session_factory;db 文件不存在的租户跳过——全新库
         # 无遗留行,保持 engine 懒建纪律(租户 engine 由 Host 中间件按请求
@@ -48,8 +69,10 @@ async def lifespan(app: FastAPI):
         yield
     finally:
         # 幂等于"更少的引擎":启动半途失败也把已建好的 engine 关掉
-        # (Windows 文件锁: 不关则库文件删不掉)
+        # (Windows 文件锁: 不关则库文件删不掉)。platform_engine 含预置的
+        # 测试引擎——dispose 后 engine 对象仍可重连,测试侧重置引用即可。
         await registry.dispose_all()
+        await platform_engine.dispose()
 
 
 app = FastAPI(title="HESUPIPE AI 客服", lifespan=lifespan)

app/platform/db.py

@@ -0,0 +1,223 @@
+"""platform.db——平台层单库(Phase 2 Task 1,spec §5.2)。
+
+PlatformBase 独立 MetaData,与租户库 Base 无关:ensure_platform_schema 只跑
+PlatformBase.metadata,engine/models.ensure_schema 只跑 Base.metadata——
+平台表与租户业务表互不误建,即使 engine 指错库也不会混表。
+
+模型要点(spec §5.2 / 计划 Task 1):
+- tenants/sites/site_domains/tenant_config_versions/admin_users/platform_settings
+- FK 级联:ondelete="CASCADE"——删租户级联 sites/site_domains/
+  tenant_config_versions/该租户的 admin_users,删站点级联其域名。
+  SQLite 默认不启用 FK enforcement,make_platform_engine 的 connect 钩子
+  每连接 PRAGMA foreign_keys=ON(与 WAL 同一钩子)。
+- tenant_config_versions:当前生效版 = MAX(version);历史不可变(回滚 =
+  旧版复制成新版本,Task 8);UNIQUE(tenant_id, version)。config_json 用
+  Text 存 JSON 字符串,读写方显式 json.loads/dumps(快照几十 KB,保持
+  直存直取,不经 ORM JSON 类型的隐式序列化)。
+- admin_users:tenant_id NULL = 超管。password_hash/salt 存 hex 文本;
+  hash = hashlib.scrypt(n=2**14, r=8, p=1)——偏差注记:spec §7.2 写
+  bcrypt/argon2,用 stdlib scrypt 为有意取舍(零新依赖,抗 GPU 性能同级);
+  verify 走 hmac.compare_digest 常量时间比较。
+
+测试/teardown 纪律:建过 platform engine 的测试结尾 await engine.dispose()
+(Windows 文件锁,同 registry)。
+"""
+import datetime
+import hashlib
+import hmac
+import pathlib
+import secrets
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    event,
+    select,
+)
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    create_async_engine,
+)
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+def _utcnow() -> datetime.datetime:
+    # 跟随租户库 models.py 既有约定:datetime.now(timezone.utc)(tz-aware UTC)
+    return datetime.datetime.now(datetime.timezone.utc)
+
+
+class PlatformBase(DeclarativeBase):
+    pass
+
+
+class Tenant(PlatformBase):
+    __tablename__ = "tenants"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    slug: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    name: Mapped[str] = mapped_column(String(128))
+    status: Mapped[str] = mapped_column(String(16), default="active")
+    created_at: Mapped[datetime.datetime] = mapped_column(default=_utcnow)
+
+
+class Site(PlatformBase):
+    """站点门面覆盖层(spec §5.2):brand_overlay 只覆盖缺省项,未覆盖项
+    继承租户 brand;DbTenantSource(Task 4)把它映射进站点 brand 合并链。"""
+
+    __tablename__ = "sites"
+    __table_args__ = (UniqueConstraint("tenant_id", "slug"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    tenant_id: Mapped[int] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), index=True)
+    slug: Mapped[str] = mapped_column(String(64))
+    name: Mapped[str] = mapped_column(String(128))
+    status: Mapped[str] = mapped_column(String(16), default="active")
+    brand_overlay: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
+
+class SiteDomain(PlatformBase):
+    """域名→站点(站点→租户随外键);domain 全局 UNIQUE(跨租户也不可
+    重复绑定,加载期/保存校验之外的数据库兜底)。"""
+
+    __tablename__ = "site_domains"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    domain: Mapped[str] = mapped_column(String(255), unique=True, index=True)
+    site_id: Mapped[int] = mapped_column(
+        ForeignKey("sites.id", ondelete="CASCADE"), index=True)
+
+
+class TenantConfigVersion(PlatformBase):
+    """租户配置版本:每次保存存整份快照(几十 KB,不做增量);
+    当前生效版 = MAX(version);UNIQUE(tenant_id, version)。"""
+
+    __tablename__ = "tenant_config_versions"
+    __table_args__ = (UniqueConstraint("tenant_id", "version"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    tenant_id: Mapped[int] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), index=True)
+    version: Mapped[int] = mapped_column(Integer)
+    config_json: Mapped[str] = mapped_column(Text)
+    updated_by: Mapped[str] = mapped_column(String(64), default="")
+    comment: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    created_at: Mapped[datetime.datetime] = mapped_column(default=_utcnow)
+
+
+class AdminUser(PlatformBase):
+    """后台账号:tenant_id NULL = 超管;must_change_password=True 表示
+    引导/重置出的初始密码,须首登改掉(Task 2 强制语义)。"""
+
+    __tablename__ = "admin_users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    tenant_id: Mapped[int | None] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True, index=True)
+    username: Mapped[str] = mapped_column(String(64), unique=True)
+    password_hash: Mapped[str] = mapped_column(String(256))   # scrypt 摘要 hex
+    salt: Mapped[str] = mapped_column(String(64))             # 随机盐 hex
+    must_change_password: Mapped[bool] = mapped_column(Boolean, default=False)
+    status: Mapped[str] = mapped_column(String(16), default="active")
+    created_at: Mapped[datetime.datetime] = mapped_column(default=_utcnow)
+
+
+class PlatformSetting(PlatformBase):
+    """平台级默认(集团 AI 凭据/默认 SMTP 等,spec §5.2)。"""
+
+    __tablename__ = "platform_settings"
+
+    key: Mapped[str] = mapped_column(String(64), primary_key=True)
+    value: Mapped[str] = mapped_column(Text, default="")
+
+
+def make_platform_engine(db_path: pathlib.Path) -> AsyncEngine:
+    """平台 SQLite engine——纪律照抄 registry.make_tenant_engine(pool=2 无
+    溢出/忙等 30s/URL as_posix/打开即 WAL,spec §6.1 读写不互斥);另开
+    PRAGMA foreign_keys=ON:平台库的级联删除(spec §5.2 删租户连带子行)
+    依赖 FK enforcement,SQLite 默认关闭,必须每连接显式开。"""
+    engine = create_async_engine(
+        f"sqlite+aiosqlite:///{db_path.as_posix()}",
+        pool_size=2, max_overflow=0,
+        connect_args={"timeout": 30})
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def _pragmas(dbapi_conn, _):                  # dbapi_conn = aiosqlite 同步适配层
+        dbapi_conn.execute("PRAGMA journal_mode=WAL")
+        dbapi_conn.execute("PRAGMA foreign_keys=ON")
+
+    return engine
+
+
+async def ensure_platform_schema(engine: AsyncEngine) -> None:
+    """create_all,只含 PlatformBase.metadata(与租户库 Base 互不误建)。
+    platform.db 是 Phase 2 新库,无旧库加列迁移负担。"""
+    async with engine.begin() as conn:
+        await conn.run_sync(PlatformBase.metadata.create_all)
+
+
+# ---- 密码哈希(scrypt,偏差注记见模块 docstring)---------------------------
+
+_SCRYPT_N = 2 ** 14
+_SCRYPT_R = 8
+_SCRYPT_P = 1
+
+
+def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
+    """scrypt(password, salt) → (password_hash, salt),均 hex 文本(便于
+    直读排查);salt 缺省每次随机 16 字节。显式传 salt 可复现(测试/verify)。"""
+    if salt is None:
+        salt = secrets.token_bytes(16)
+    digest = hashlib.scrypt(
+        password.encode("utf-8"), salt=salt,
+        n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P)
+    return digest.hex(), salt.hex()
+
+
+def verify_password(password: str, user: AdminUser) -> bool:
+    """常量时间比较(hmac.compare_digest)——防时序侧信道;坏 hex 按验败
+    处理而非炸(存储异常不该变成 500)。"""
+    try:
+        salt = bytes.fromhex(user.salt)
+        expected = bytes.fromhex(user.password_hash)
+    except ValueError:
+        return False
+    actual = hashlib.scrypt(
+        password.encode("utf-8"), salt=salt,
+        n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P)
+    return hmac.compare_digest(expected, actual)
+
+
+# ---- 超管引导 ---------------------------------------------------------------
+
+async def bootstrap_superuser(
+        engine: AsyncEngine, username: str, password: str) -> None:
+    """空表才建超管(tenant_id NULL, must_change_password=True);表非空
+    一律不动——不重复建、不改已有用户/密码(幂等:lifespan 每次启动安全
+    重入,运行期改密不会被启动覆盖)。"""
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        existing = (await session.execute(
+            select(AdminUser.id).limit(1))).first()
+        if existing is not None:
+            return
+        password_hash, salt = hash_password(password)
+        session.add(AdminUser(
+            tenant_id=None,
+            username=username,
+            password_hash=password_hash,
+            salt=salt,
+            must_change_password=True))
+        await session.commit()
+
+
+async def list_admin_users(engine: AsyncEngine) -> list[AdminUser]:
+    """全部后台账号(按 id 稳序)——测试/管理列表用。"""
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        result = await session.execute(select(AdminUser).order_by(AdminUser.id))
+        return list(result.scalars().all())

tests/conftest.py

@@ -18,6 +18,7 @@ from app.engine.models import Base
 # "先 import app.main 再 import chat_rest"的循环导入次序坑随全局单例一并
 # 消失(routes 不再反向 import app.main),直接导入即可。
 from app.main import app
+from app.platform.db import make_platform_engine
 from app.platform.registry import TenantRegistry
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -136,9 +137,20 @@ def build_app(seed_dir, data_dir):
     延迟取 registry),工厂只负责把 Registry 放进 app.state——lifespan 见
     已挂则沿用(测试注入缝),不再加载仓库 seed。调用方用完必须复位
     app.state.registry = None(见 app_test_client 的 finally)。
+
+    Phase 2 Task 1 起同时预置 platform_engine(data_dir 下的临时平台库,
+    同一注入缝)——不预置则 lifespan 会在仓库根建出真实 platform.db;
+    schema+bootstrap 由 lifespan 统一跑(幂等),dispose 也在 lifespan
+    finally(与 registry.dispose_all 同处)。
     """
     registry = TenantRegistry.load_from_dir(seed_dir, data_dir=data_dir)
     app.state.registry = registry
+    # data_dir 可能尚不存在(租户库由 _ensure_ready 按需 mkdir,平台库由
+    # lifespan 立即连接——这里先建目录,否则 sqlite "unable to open database
+    # file")
+    Path(data_dir).mkdir(parents=True, exist_ok=True)
+    app.state.platform_engine = make_platform_engine(
+        Path(data_dir) / "platform.db")
     return app
 
 
@@ -172,10 +184,10 @@ class TenantTestClient(TestClient):
 
 @contextmanager
 def app_test_client(app_instance, base_url, cleanup_root):
-    """TestClient 包装:统一退出纪律——lifespan finally 已 dispose_all
+    """TestClient 包装:统一退出纪律——lifespan finally 已 dispose_all +
-    (Windows 文件锁),再置空 app.state.registry(关残留窗:不让已 dispose
+    platform_engine.dispose(Windows 文件锁),再置空 app.state.registry /
-    的测试 Registry 被后续无夹具的 lifespan 误用),最后清 tmp 目录
+    platform_engine(关残留窗:不让已 dispose 的测试对象被后续无夹具的
-    (删不掉告警留痕,不静默吞)。"""
+    lifespan 误用),最后清 tmp 目录(删不掉告警留痕,不静默吞)。"""
     try:
         with TenantTestClient(
                 app_instance, base_url=base_url,
@@ -183,6 +195,7 @@ def app_test_client(app_instance, base_url, cleanup_root):
             yield c
     finally:
         app_instance.state.registry = None
+        app_instance.state.platform_engine = None
         _rmtree_warn(cleanup_root)
 
 

tests/test_platform_db.py

@@ -0,0 +1,253 @@
+"""platform.db 模型与超管引导(Phase 2 Task 1)。
+
+独立 PlatformBase(与租户库 Base 无关);WAL/pool=2 纪律照抄
+make_tenant_engine;scrypt 密码哈希 + 常量时间校验;bootstrap 空表才建
+(幂等);FK 级联删除(PAGMA foreign_keys=ON,删租户连带全部子行);
+main.py lifespan 装配缝(预置 platform_engine 则沿用)。
+
+Windows 纪律:每个建过 engine 的用例结尾 await engine.dispose()——
+连接不关,文件锁着,tmp_path 清理 PermissionError。
+"""
+import asyncio
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.platform.db import (
+    AdminUser,
+    PlatformSetting,
+    Site,
+    SiteDomain,
+    Tenant,
+    TenantConfigVersion,
+    bootstrap_superuser,
+    ensure_platform_schema,
+    hash_password,
+    list_admin_users,
+    make_platform_engine,
+    verify_password,
+)
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SEED_DIR = REPO_ROOT / "seed_profiles"
+
+
+async def _insert_fixture(engine):
+    """最小平台数据:租户 t1 + 站点 main + 域名 + 配置版本 1 + 租户管理员。"""
+    password_hash, salt = hash_password("tenant-pw")
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = Tenant(slug="t1", name="T1")
+        session.add(tenant)
+        await session.flush()
+        site = Site(tenant_id=tenant.id, slug="main", name="Main")
+        session.add(site)
+        await session.flush()
+        session.add(SiteDomain(domain="a.example.com", site_id=site.id))
+        session.add(TenantConfigVersion(
+            tenant_id=tenant.id, version=1, config_json="{}"))
+        session.add(AdminUser(
+            tenant_id=tenant.id, username="t1admin",
+            password_hash=password_hash, salt=salt))
+        await session.commit()
+        return tenant.id, site.id
+
+
+# ---- bootstrap 超管 ---------------------------------------------------------
+
+
+async def test_bootstrap_creates_superuser_once(tmp_path):
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        await bootstrap_superuser(engine, username="admin", password="s3cret!")
+        await bootstrap_superuser(engine, username="admin", password="other")  # 幂等:不重复建/不改密
+        users = await list_admin_users(engine)
+        assert len(users) == 1 and users[0].username == "admin"
+        assert users[0].tenant_id is None            # NULL = 超管
+        assert users[0].must_change_password is True  # 引导密码必须首登改掉
+        assert verify_password("s3cret!", users[0])
+        assert not verify_password("other", users[0])
+    finally:
+        await engine.dispose()
+
+
+async def test_bootstrap_noop_when_any_user_exists(tmp_path):
+    """空表才建:表非空时换用户名再来也不建不改(启动安全重入)。"""
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        await bootstrap_superuser(engine, username="admin", password="first")
+        await bootstrap_superuser(engine, username="root", password="second")
+        users = await list_admin_users(engine)
+        assert len(users) == 1 and users[0].username == "admin"
+        assert verify_password("first", users[0])
+    finally:
+        await engine.dispose()
+
+
+# ---- engine 纪律 -------------------------------------------------------------
+
+
+async def test_platform_db_wal(tmp_path):
+    """spec §6.1 硬纪律:平台库打开即 WAL(查法照抄 test_registry)。"""
+    from sqlalchemy import text
+
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        async with engine.connect() as conn:
+            mode = (await conn.execute(text("PRAGMA journal_mode"))).scalar()
+        assert str(mode).lower() == "wal"
+    finally:
+        await engine.dispose()
+
+
+# ---- 密码哈希 -----------------------------------------------------------------
+
+
+def test_hash_password_fresh_salt_per_call():
+    """同密码两次哈希:salt/摘要都不同(随机 salt,彩虹表防护);
+    显式传 salt 可复现(verify_password 内部路径)。"""
+    h1, s1 = hash_password("same")
+    h2, s2 = hash_password("same")
+    assert s1 != s2 and h1 != h2
+    h3, s3 = hash_password("same", salt=bytes.fromhex(s1))
+    assert h3 == h1 and s3 == s1
+
+
+# ---- 模型约束 -------------------------------------------------------------
+
+
+async def test_unique_constraints(tmp_path):
+    """UNIQUE 们:tenants.slug / sites(tenant_id, slug) /
+    site_domains.domain / tenant_config_versions(tenant_id, version) /
+    admin_users.username。"""
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        await _insert_fixture(engine)
+        async with AsyncSession(engine) as session:
+            session.add(Tenant(slug="t1", name="dup"))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+
+            # 取 id 列而非 ORM 实体:rollback 会 expire 实例,事后访问
+            # 属性将触发同步懒加载(async 下 MissingGreenlet)
+            t1_id = (await session.execute(
+                select(Tenant.id).filter_by(slug="t1"))).scalar_one()
+            session.add(Site(tenant_id=t1_id, slug="main", name="dup"))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+
+            site_id = (await session.execute(
+                select(Site.id).filter_by(slug="main"))).scalar_one()
+            session.add(SiteDomain(domain="a.example.com", site_id=site_id))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+
+            session.add(TenantConfigVersion(
+                tenant_id=t1_id, version=1, config_json="{}"))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+
+            session.add(AdminUser(
+                username="t1admin", password_hash="x", salt="y"))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+    finally:
+        await engine.dispose()
+
+
+async def test_delete_tenant_cascades_children(tmp_path):
+    """FK 级联(make_platform_engine 每连接 PRAGMA foreign_keys=ON):
+    删租户 → sites/site_domains/tenant_config_versions/该租户的 admin_users
+    级联消失;超管(tenant_id NULL)不受波及。"""
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        await _insert_fixture(engine)
+        password_hash, salt = hash_password("super-pw")
+        async with AsyncSession(engine) as session:
+            session.add(AdminUser(
+                username="super", password_hash=password_hash, salt=salt))
+            await session.commit()
+        async with AsyncSession(engine) as session:
+            t1 = (await session.execute(
+                select(Tenant).filter_by(slug="t1"))).scalar_one()
+            await session.delete(t1)
+            await session.commit()
+        async with AsyncSession(engine) as session:
+            assert (await session.scalar(
+                select(func.count()).select_from(Site))) == 0
+            assert (await session.scalar(
+                select(func.count()).select_from(SiteDomain))) == 0
+            assert (await session.scalar(
+                select(func.count()).select_from(TenantConfigVersion))) == 0
+            admins = (await session.execute(
+                select(AdminUser).order_by(AdminUser.id))).scalars().all()
+            assert [u.username for u in admins] == ["super"]
+    finally:
+        await engine.dispose()
+
+
+async def test_platform_setting_roundtrip(tmp_path):
+    engine = make_platform_engine(tmp_path / "platform.db")
+    try:
+        await ensure_platform_schema(engine)
+        async with AsyncSession(engine) as session:
+            session.add(PlatformSetting(
+                key="smtp.default.host", value="smtp.example.com"))
+            await session.commit()
+            got = await session.get(PlatformSetting, "smtp.default.host")
+            assert got is not None and got.value == "smtp.example.com"
+    finally:
+        await engine.dispose()
+
+
+# ---- main.py lifespan 装配缝 ---------------------------------------------------
+
+
+def test_lifespan_platform_engine_seam_and_bootstrap(tmp_path, monkeypatch):
+    """预置 app.state.platform_engine → lifespan 沿用该引擎(不在仓库根建
+    真实 platform.db),对它跑 schema+bootstrap(settings 凭据),退出 dispose;
+    落盘结果用独立引擎复验(跨循环纪律:asyncio.run 单循环内完成)。"""
+    from app.engine.config import settings
+    from app.main import app
+    from app.platform.registry import TenantRegistry
+
+    monkeypatch.setattr(settings, "ADMIN_USERNAME", "bootadmin")
+    monkeypatch.setattr(settings, "ADMIN_PASSWORD", "boot-pw")
+
+    engine = make_platform_engine(tmp_path / "platform.db")
+    app.state.registry = TenantRegistry.load_from_dir(
+        SEED_DIR, data_dir=tmp_path / "tenants")
+    app.state.platform_engine = engine
+    try:
+        with TestClient(app) as client:
+            assert app.state.platform_engine is engine   # 沿用,未被换掉
+            assert client.get("/health").status_code == 200
+        # 退出时 lifespan finally 已 dispose(Windows 文件锁释放,tmp 可清理)
+    finally:
+        app.state.registry = None
+        app.state.platform_engine = None
+
+    async def _probe():
+        check = make_platform_engine(tmp_path / "platform.db")
+        try:
+            return await list_admin_users(check)
+        finally:
+            await check.dispose()
+
+    users = asyncio.run(_probe())
+    assert [u.username for u in users] == ["bootadmin"]
+    assert verify_password("boot-pw", users[0])
+    assert users[0].must_change_password is True