feat: 租户管理API（创建向导消费出厂档案/blank保留jc墙/域名/停用/slug不可变）

POST /api/platform/tenants: profile ∈ {auto,jiancai,blank}——首版快照复用
FileTenantSource(seed).read_tenant(profile).snapshot(Task 4 审查指定);
blank=jiancai 档案 _blankify 去身份组但保留 wall/prompts(compile_wall/
PromptProvider 的硬要求)。建 Tenant+main 站点+版本1+租户管理员
(must_change_password=True),创建后 registry_reload(upsert)立即可用。

站点 CRUD(brand_overlay 整字典赋值,M-2a)/域名绑定解绑(全局唯一,
撞车 400 带现有归属)/PATCH 租户(name/status;slug 不可变 400;停用走
reload 路径 dispose 引擎)。列表/详情管理员概要显式挑字段,凭据永不外泄。
鉴权: 写操作+列表 require_super;GET 详情放宽 require_tenant(slug)。

测试: db_source_app 夹具(registry 用 DbTenantSource 生产形态指向临时
platform.db,跨循环纪律: 直查走独立 NullPool 连接),16 项全绿,全量 519。

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

app/main.py

@@ -81,9 +81,13 @@ from app.engine.routes.chat import router as chat_router
 from app.engine.routes.chat_rest import router as chat_rest_router
 from app.engine.routes.public import router as public_router
 from app.platform.routes.auth_routes import router as platform_auth_router
+from app.platform.routes.tenant_routes import router as platform_tenant_router
 app.include_router(chat_router)
 app.include_router(chat_rest_router)
 app.include_router(public_router)
+# 租户管理 API(Task 5):创建向导/站点/域名/停用——同样只依赖路由层认证
+# (require_super / require_tenant),Host 豁免面归 Task 3 的前缀规则覆盖。
+app.include_router(platform_tenant_router)
 # 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下
 # /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为
 # /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析);

app/platform/routes/tenant_routes.py

@@ -0,0 +1,516 @@
+"""租户管理 API(Phase 2 Task 5):/api/platform/tenants*——超管的创建向导/
+站点/域名/停用,GET 详情放宽到本租户管理员。
+
+创建向导的快照复用路径(Task 4 审查指定,勿手搓拼装):
+``FileTenantSource(SEED_DIR).read_tenant(profile).snapshot`` 直接产出首版
+快照——auto/jiancai 原样入版本1;**blank** = jiancai 档案经 _blankify 去
+身份组(brand 的 title/whatsapp/email/privacy_url、push.prefix、
+engine.context_seed_keys 置出厂默认),**wall 与 prompts 组保留**(取
+jiancai 档案)——compile_wall 要求 wall 全字段、PromptProvider 要求四件套,
+DEFAULT_CONFIG 满足不了。
+
+写操作后一律 registry_reload(slug)(Task 4 upsert 语义:新建租户未加载
+也直接插入,创建即可用);停用走 reload 路径顺带 dispose 旧 engine
+(引擎释放由 registry 负责,本层不碰)。
+
+鉴权:写操作与列表 require_super;GET /{slug} 详情放宽 require_tenant
+(slug)(租户管理员看自己概要,403 边界有测试钉住)。
+
+安全纪律:
+- 密码哈希/salt 绝不进响应——管理员序列化显式挑 {username, status}
+  (db.list_admin_users 的 M-5 警示);
+- 创建出的租户管理员一律 must_change_password=True(引导密码首登改掉)。
+
+M-2a 陷阱:Site.brand_overlay 是 ORM JSON 列——写时**整字典赋值*,不做
+按键局部更新(局部改不标记脏,静默丢写)。
+"""
+import copy
+import json
+import pathlib
+import re
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel, Field
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.platform.auth import (
+    UserContext,
+    hash_password,
+    platform_user,
+    require_super,
+    require_tenant,
+)
+from app.platform.db import AdminUser, Site, SiteDomain, Tenant, TenantConfigVersion
+from app.platform.defaults import DEFAULT_CONFIG
+from app.platform.registry import _normalize_host
+from app.platform.tenant_source import FileTenantSource
+
+router = APIRouter(prefix="/api/platform/tenants", tags=["platform-tenants"])
+
+# 出厂档案目录:与 app.main.SEED_DIR 同一锚定规则(仓库根/seed_profiles),
+# 但不 import app.main——main 模块级挂路由,反向导入制造循环依赖。
+SEED_DIR = pathlib.Path(__file__).resolve().parents[3] / "seed_profiles"
+
+PROFILES = ("auto", "jiancai", "blank")
+STATUSES = ("active", "disabled")
+SLUG_RE = re.compile(r"^[a-z0-9-]+$")
+
+# blank 身份组清理清单(brand 的这几个键置出厂默认;其余 brand 字段保留)
+_IDENTITY_BRAND_KEYS = ("title", "whatsapp", "email", "privacy_url")
+
+
+def _blankify(snapshot: dict) -> dict:
+    """出厂档案快照 → blank 形态:身份组置出厂默认,其余(含 wall/prompts)
+    原样保留。深拷贝后改——纯函数纪律,绝不动入参。"""
+    out = copy.deepcopy(snapshot)
+    brand = dict(out.get("brand") or {})
+    for key in _IDENTITY_BRAND_KEYS:
+        brand[key] = DEFAULT_CONFIG["brand"][key]
+    out["brand"] = brand
+    push = dict(out.get("push") or {})
+    push["prefix"] = DEFAULT_CONFIG["push"]["prefix"]
+    out["push"] = push
+    engine = dict(out.get("engine") or {})
+    engine["context_seed_keys"] = dict(
+        DEFAULT_CONFIG["engine"]["context_seed_keys"])
+    out["engine"] = engine
+    return out
+
+
+def _profile_snapshot(profile: str) -> dict:
+    """profile → 首版快照:auto/jiancai 读对应出厂档案原样;blank 读 jiancai
+    档案再去身份组(wall/prompts 保留——见模块 docstring)。"""
+    base_slug = "jiancai" if profile == "blank" else profile
+    snapshot = FileTenantSource(SEED_DIR).read_tenant(base_slug).snapshot
+    return _blankify(snapshot) if profile == "blank" else snapshot
+
+
+# ---- 校验助手 -------------------------------------------------------------------
+
+
+def _validate_slug(slug: str) -> str:
+    if not SLUG_RE.fullmatch(slug or ""):
+        raise HTTPException(
+            status_code=400, detail="slug 只允许小写字母、数字和连字符")
+    return slug
+
+
+def _normalize_domain(domain: str) -> str:
+    """入库前归一(小写/去端口/去尾点——与 resolve 的 host 归一同一事实源);
+    归一后为空 → 400。"""
+    normalized = _normalize_host(domain or "")
+    if not normalized:
+        raise HTTPException(status_code=400, detail="域名不能为空")
+    return normalized
+
+
+def _check_name(name) -> None:
+    if not isinstance(name, str) or not 1 <= len(name) <= 128:
+        raise HTTPException(status_code=400, detail="name 须为 1-128 字符")
+
+
+def _check_status(status) -> None:
+    if status not in STATUSES:
+        raise HTTPException(
+            status_code=400,
+            detail=f"status 只允许 {STATUSES}(实际 {status!r})")
+
+
+def _patch_fields(body: dict, *, allow: tuple[str, ...]) -> dict:
+    """PATCH 通用校验:slug 键一律 400(不可变);白名单外字段 400;显式
+    null 视为缺席;全空 → 400(无意义补丁)。返回只含出现且非 null 的字段。"""
+    if "slug" in body:
+        raise HTTPException(status_code=400, detail="slug 不可修改")
+    unknown = set(body) - set(allow)
+    if unknown:
+        raise HTTPException(
+            status_code=400,
+            detail=f"不可修改的字段: {', '.join(sorted(unknown))}")
+    fields = {k: v for k, v in body.items() if v is not None}
+    if not fields:
+        raise HTTPException(status_code=400, detail="没有可更新字段")
+    return fields
+
+
+async def _get_tenant_or_404(session: AsyncSession, slug: str) -> Tenant:
+    tenant = (await session.execute(
+        select(Tenant).where(Tenant.slug == slug))).scalar_one_or_none()
+    if tenant is None:
+        raise HTTPException(status_code=404, detail=f"租户不存在: {slug!r}")
+    return tenant
+
+
+async def _get_site_or_404(
+        session: AsyncSession, tenant_slug: str, site_slug: str) -> Site:
+    site = (await session.execute(
+        select(Site).join(Tenant, Site.tenant_id == Tenant.id)
+        .where(Tenant.slug == tenant_slug, Site.slug == site_slug)
+    )).scalar_one_or_none()
+    if site is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"站点不存在: {tenant_slug}/{site_slug}")
+    return site
+
+
+async def _assert_domain_free(session: AsyncSession, domain: str) -> None:
+    """域名全局唯一预检:撞车 400 带现有归属(租户+站点)——调用方能一眼
+    看到域名被谁占了。"""
+    row = (await session.execute(
+        select(SiteDomain.domain, Tenant.slug, Site.slug)
+        .join(Site, SiteDomain.site_id == Site.id)
+        .join(Tenant, Site.tenant_id == Tenant.id)
+        .where(SiteDomain.domain == domain))).first()
+    if row is not None:
+        raise HTTPException(
+            status_code=400,
+            detail=f"域名 {domain} 已绑定租户 {row[1]!r} 的站点 {row[2]!r}")
+
+
+async def _site_summary(session: AsyncSession, site: Site) -> dict:
+    domains = (await session.execute(
+        select(SiteDomain.domain).where(SiteDomain.site_id == site.id)
+        .order_by(SiteDomain.id))).scalars().all()
+    return {
+        "slug": site.slug, "name": site.name, "status": site.status,
+        "domains": list(domains),
+        "brand_overlay": dict(site.brand_overlay or {}),
+    }
+
+
+def _admin_summary(admin: AdminUser) -> dict:
+    # 显式挑字段——password_hash/salt 绝不进响应(M-5)
+    return {"username": admin.username, "status": admin.status}
+
+
+def _tenant_guard(slug: str, user: platform_user) -> UserContext:
+    """路径参数版 require_tenant:复用 auth 的依赖工厂,slug 取自路径
+    (super 或本租户管理员放行,409 门经 platform_user 连锁)。"""
+    return require_tenant(slug)(user)
+
+
+# ---- 请求模型 -------------------------------------------------------------------
+
+
+class CreateTenantRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=128)
+    slug: str = Field(min_length=1, max_length=64)
+    profile: str
+    admin_username: str = Field(min_length=1, max_length=64)
+    admin_password: str = Field(min_length=1, max_length=128)
+    domains: list[str] = Field(default_factory=list)
+
+
+class CreateSiteRequest(BaseModel):
+    slug: str = Field(min_length=1, max_length=64)
+    name: str = Field(default="", max_length=128)
+    brand_overlay: dict = Field(default_factory=dict)
+    domains: list[str] = Field(default_factory=list)
+
+
+class AddDomainRequest(BaseModel):
+    domain: str = Field(min_length=1, max_length=255)
+
+
+# ---- 租户:创建 / 列表 / 详情 / 改名改状态 ------------------------------------------
+
+
+@router.post("", status_code=201)
+async def create_tenant(
+        body: CreateTenantRequest, request: Request,
+        user=Depends(require_super)):
+    """创建向导:校验 slug/profile → 出厂档案快照 → 建租户+main 站点+
+    版本1+租户管理员(must_change_password=True)→ registry_reload(upsert,
+    创建即可用)。"""
+    _validate_slug(body.slug)
+    if body.profile not in PROFILES:
+        raise HTTPException(
+            status_code=400,
+            detail=f"未知 profile: {body.profile!r}(可选 {PROFILES})")
+    # 请求内去重(保序),免自撞唯一约束
+    domains = list(dict.fromkeys(
+        _normalize_domain(d) for d in body.domains))
+    try:
+        snapshot = _profile_snapshot(body.profile)
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=400, detail=f"出厂档案缺失: {exc}") from exc
+
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        if (await session.execute(
+                select(Tenant.id).where(Tenant.slug == body.slug))).first():
+            raise HTTPException(
+                status_code=400, detail=f"slug 已存在: {body.slug!r}")
+        if (await session.execute(
+                select(AdminUser.id).where(
+                    AdminUser.username == body.admin_username))).first():
+            raise HTTPException(
+                status_code=400,
+                detail=f"管理员用户名已被占用: {body.admin_username!r}")
+        for domain in domains:
+            await _assert_domain_free(session, domain)
+
+        tenant = Tenant(slug=body.slug, name=body.name, status="active")
+        session.add(tenant)
+        await session.flush()
+        site = Site(tenant_id=tenant.id, slug="main", name="main")
+        session.add(site)
+        await session.flush()
+        for domain in domains:
+            session.add(SiteDomain(domain=domain, site_id=site.id))
+        password_hash, salt = hash_password(body.admin_password)
+        session.add(AdminUser(
+            tenant_id=tenant.id,
+            username=body.admin_username,
+            password_hash=password_hash,
+            salt=salt,
+            must_change_password=True))
+        session.add(TenantConfigVersion(
+            tenant_id=tenant.id,
+            version=1,
+            config_json=json.dumps(snapshot, ensure_ascii=False),
+            updated_by=f"super#{user.user_id}",
+            comment=f"创建向导: profile={body.profile}"))
+        try:
+            await session.commit()
+        except IntegrityError as exc:
+            # 并发竞态兜底(预检与提交之间被人抢先):唯一约束落库层兜住
+            await session.rollback()
+            raise HTTPException(
+                status_code=409,
+                detail="并发冲突: slug/用户名/域名已被占用") from exc
+
+    await request.app.state.registry.registry_reload(body.slug)
+    return {
+        "slug": tenant.slug, "name": tenant.name, "status": tenant.status,
+        "profile": body.profile, "version": 1,
+        "site": {"slug": "main", "domains": domains},
+        "admin_username": body.admin_username,
+    }
+
+
+@router.get("")
+async def list_tenants(request: Request, user=Depends(require_super)):
+    """租户列表:slug/name/status/域名数/当前版本号/管理员概要。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenants = (await session.execute(
+            select(Tenant).order_by(Tenant.slug))).scalars().all()
+        domain_counts = dict((await session.execute(
+            select(Site.tenant_id, func.count(SiteDomain.id))
+            .join(SiteDomain, SiteDomain.site_id == Site.id)
+            .group_by(Site.tenant_id))).all())
+        versions = dict((await session.execute(
+            select(TenantConfigVersion.tenant_id,
+                   func.max(TenantConfigVersion.version))
+            .group_by(TenantConfigVersion.tenant_id))).all())
+        admins = (await session.execute(
+            select(AdminUser).where(AdminUser.tenant_id.is_not(None))
+            .order_by(AdminUser.id))).scalars().all()
+
+    admins_by_tenant: dict[int, list[dict]] = {}
+    for admin in admins:
+        admins_by_tenant.setdefault(admin.tenant_id, []).append(
+            _admin_summary(admin))
+    return {"items": [{
+        "slug": t.slug,
+        "name": t.name,
+        "status": t.status,
+        "domain_count": domain_counts.get(t.id, 0),
+        "current_version": versions.get(t.id),
+        "admins": admins_by_tenant.get(t.id, []),
+    } for t in tenants]}
+
+
+@router.get("/{slug}")
+async def tenant_detail(
+        slug: str, request: Request, user=Depends(_tenant_guard)):
+    """租户详情:sites(含 domains/brand_overlay)/管理员用户名/当前版本。
+    鉴权放宽 require_tenant(slug)——租户管理员看自己概要。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        sites = (await session.execute(
+            select(Site).where(Site.tenant_id == tenant.id)
+            .order_by(Site.id))).scalars().all()
+        domains = (await session.execute(
+            select(SiteDomain)
+            .where(SiteDomain.site_id.in_([s.id for s in sites]))
+            .order_by(SiteDomain.id))).scalars().all()
+        current_version = (await session.execute(
+            select(func.max(TenantConfigVersion.version))
+            .where(TenantConfigVersion.tenant_id == tenant.id))).scalar()
+        admins = (await session.execute(
+            select(AdminUser).where(AdminUser.tenant_id == tenant.id)
+            .order_by(AdminUser.id))).scalars().all()
+
+    domains_by_site: dict[int, list[str]] = {}
+    for d in domains:
+        domains_by_site.setdefault(d.site_id, []).append(d.domain)
+    return {
+        "slug": tenant.slug,
+        "name": tenant.name,
+        "status": tenant.status,
+        "current_version": current_version,
+        "sites": [{
+            "slug": s.slug, "name": s.name, "status": s.status,
+            "domains": list(domains_by_site.get(s.id, [])),
+            "brand_overlay": dict(s.brand_overlay or {}),
+        } for s in sites],
+        "admins": [_admin_summary(a) for a in admins],
+    }
+
+
+@router.patch("/{slug}")
+async def patch_tenant(
+        slug: str, body: dict, request: Request, user=Depends(require_super)):
+    """改名/停用复启用。slug 不可变(400);status≠active → registry reload
+    后 resolve 抛 TenantDisabledError,引擎释放由 reload 路径处理(dispose)。"""
+    fields = _patch_fields(body, allow=("name", "status"))
+    if "name" in fields:
+        _check_name(fields["name"])
+    if "status" in fields:
+        _check_status(fields["status"])
+
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        if "name" in fields:
+            tenant.name = fields["name"]
+        if "status" in fields:
+            tenant.status = fields["status"]
+        await session.commit()
+        result = {"slug": tenant.slug, "name": tenant.name,
+                  "status": tenant.status}
+
+    await request.app.state.registry.registry_reload(slug)
+    return result
+
+
+# ---- 站点:建站 / 改门面 -----------------------------------------------------------
+
+
+@router.post("/{slug}/sites", status_code=201)
+async def create_site(
+        slug: str, body: CreateSiteRequest, request: Request,
+        user=Depends(require_super)):
+    """建站:slug/name/brand_overlay/domains 可选。brand_overlay 是 ORM
+    JSON 列——整字典赋值(M-2a)。"""
+    _validate_slug(body.slug)
+    domains = list(dict.fromkeys(
+        _normalize_domain(d) for d in body.domains))
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        if (await session.execute(
+                select(Site.id).where(Site.tenant_id == tenant.id,
+                                      Site.slug == body.slug))).first():
+            raise HTTPException(
+                status_code=400,
+                detail=f"站点 slug 已存在: {body.slug!r}")
+        for domain in domains:
+            await _assert_domain_free(session, domain)
+
+        site = Site(tenant_id=tenant.id, slug=body.slug,
+                    name=body.name or body.slug,
+                    brand_overlay=dict(body.brand_overlay))
+        session.add(site)
+        await session.flush()
+        for domain in domains:
+            session.add(SiteDomain(domain=domain, site_id=site.id))
+        try:
+            await session.commit()
+        except IntegrityError as exc:
+            await session.rollback()
+            raise HTTPException(
+                status_code=409,
+                detail="并发冲突: 站点 slug 或域名已被占用") from exc
+        result = await _site_summary(session, site)
+
+    await request.app.state.registry.registry_reload(slug)
+    return result
+
+
+@router.patch("/{slug}/sites/{site_slug}")
+async def patch_site(
+        slug: str, site_slug: str, body: dict, request: Request,
+        user=Depends(require_super)):
+    """改站点:停用/改名/改 overlay(整字典赋值,旧键不残留)。站点 slug
+    同样不可变(400)。"""
+    fields = _patch_fields(body, allow=("name", "status", "brand_overlay"))
+    if "name" in fields:
+        _check_name(fields["name"])
+    if "status" in fields:
+        _check_status(fields["status"])
+    if "brand_overlay" in fields and not isinstance(
+            fields["brand_overlay"], dict):
+        raise HTTPException(
+            status_code=400, detail="brand_overlay 必须是 JSON 对象")
+
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        site = await _get_site_or_404(session, slug, site_slug)
+        if "name" in fields:
+            site.name = fields["name"]
+        if "status" in fields:
+            site.status = fields["status"]
+        if "brand_overlay" in fields:
+            site.brand_overlay = dict(fields["brand_overlay"])   # M-2a 整赋值
+        await session.commit()
+        result = await _site_summary(session, site)
+
+    await request.app.state.registry.registry_reload(slug)
+    return result
+
+
+# ---- 域名:绑定 / 解绑 ------------------------------------------------------------
+
+
+@router.post("/{slug}/sites/{site_slug}/domains", status_code=201)
+async def add_domain(
+        slug: str, site_slug: str, body: AddDomainRequest, request: Request,
+        user=Depends(require_super)):
+    """域名绑定:全局唯一(撞车 400 带现有归属);绑定后 registry reload
+    生效。"""
+    domain = _normalize_domain(body.domain)
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        site = await _get_site_or_404(session, slug, site_slug)
+        await _assert_domain_free(session, domain)
+        session.add(SiteDomain(domain=domain, site_id=site.id))
+        try:
+            await session.commit()
+        except IntegrityError as exc:
+            await session.rollback()
+            raise HTTPException(
+                status_code=409,
+                detail=f"并发冲突: 域名 {domain} 已被占用") from exc
+
+    await request.app.state.registry.registry_reload(slug)
+    return {"domain": domain, "site": site_slug}
+
+
+@router.delete("/{slug}/sites/{site_slug}/domains/{domain}")
+async def delete_domain(
+        slug: str, site_slug: str, domain: str, request: Request,
+        user=Depends(require_super)):
+    """域名解绑:只解本站点的该域名(不存在 404);解绑后 reload 生效。"""
+    domain = _normalize_domain(domain)
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        site = await _get_site_or_404(session, slug, site_slug)
+        row = (await session.execute(
+            select(SiteDomain).where(SiteDomain.site_id == site.id,
+                                     SiteDomain.domain == domain)
+        )).scalar_one_or_none()
+        if row is None:
+            raise HTTPException(
+                status_code=404, detail=f"域名不存在: {domain!r}")
+        await session.delete(row)
+        await session.commit()
+
+    await request.app.state.registry.registry_reload(slug)
+    return {"ok": True, "domain": domain, "site": site_slug}

tests/test_tenant_admin_api.py

@@ -0,0 +1,504 @@
+"""租户管理 API(Phase 2 Task 5):超管的创建向导/站点/域名/停用 + 403 边界。
+
+db_source_app 夹具:app.state.registry 用 DbTenantSource(生产形态)指向
+临时 platform.db——创建 API 写库后 registry_reload 走 DB 源闭环(Task 4
+的 upsert 语义:未加载的 slug 直接插入,创建即可用)。platform.db 起始
+为空(schema+bootstrap 超管由 lifespan 跑)——租户全部经创建 API 进入,
+registry 不做初始 load_all(生产装配的启动次序归 Task 11)。
+
+跨循环纪律:platform engine(pool=2 常驻连接)只在 TestClient portal
+循环内使用(lifespan+路由);测试线程直查 platform.db 一律走 _db_fetch
+(独立 NullPool engine,即用即弃,绝不复用 portal 循环的常驻连接)。
+"""
+import asyncio
+import json
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.engine.services.ai_service import PromptProvider
+from app.engine.wall_schema import compile_wall
+from app.main import app
+from app.platform.db import (
+    AdminUser,
+    Site,
+    SiteDomain,
+    Tenant,
+    TenantConfigVersion,
+    make_platform_engine,
+)
+from app.platform.defaults import DEFAULT_CONFIG
+from app.platform.registry import TenantDisabledError, TenantRegistry
+from app.platform.routes.tenant_routes import _blankify
+from app.platform.tenant_source import DbTenantSource, FileTenantSource
+from tests.conftest import SEED_DIR, app_test_client
+
+TENANTS = "/api/platform/tenants"
+LOGIN = "/api/platform/auth/login"
+CHANGE_PW = "/api/platform/auth/change-password"
+
+
+# ---- 夹具 ----------------------------------------------------------------------
+
+
+@pytest.fixture
+def db_source_app(tmp_path):
+    """app + 生产形态 registry(DbTenantSource 指向临时 platform.db)。
+
+    platform_engine 也指同一临时库(lifespan 在 portal 循环里 ensure_schema
+    + bootstrap 超管 admin/admin123)。teardown 复位 app.state(照抄
+    app_test_client 的关残留窗纪律;engine dispose 由 lifespan finally)。
+    """
+    engine = make_platform_engine(tmp_path / "platform.db")
+    app.state.registry = TenantRegistry(
+        DbTenantSource(engine), data_dir=tmp_path / "tenants")
+    app.state.platform_engine = engine
+    try:
+        yield app
+    finally:
+        app.state.registry = None
+        app.state.platform_engine = None
+
+
+@pytest.fixture
+def super_client(db_source_app, tmp_path):
+    """超管 client:lifespan bootstrap 的 admin/admin123 登录 + 首登改密
+    (过 must_change 409 门)。token 挂 client.platform_token。"""
+    with app_test_client(db_source_app, "http://testserver", tmp_path) as c:
+        r = c.post(LOGIN, json={"username": "admin", "password": "admin123"})
+        assert r.status_code == 200
+        c.platform_token = r.json()["token"]
+        r = c.post(CHANGE_PW, headers=_auth(c), json={
+            "old_password": "admin123", "new_password": "sup3r-pass-1"})
+        assert r.status_code == 200
+        yield c
+
+
+# ---- 助手 ----------------------------------------------------------------------
+
+
+def _auth(client):
+    return {"Authorization": f"Bearer {client.platform_token}"}
+
+
+def _bearer(token):
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _create(client, slug="demo", profile="jiancai", name=None, admin=None,
+            password=None, domains=None):
+    """创建租户的便捷 POST(admin/password 缺省按 slug 派生,免撞用户名)。"""
+    return client.post(TENANTS, headers=_auth(client), json={
+        "name": name or f"{slug} 租户",
+        "slug": slug,
+        "profile": profile,
+        "admin_username": admin or f"{slug}-admin",
+        "admin_password": password or f"{slug}-pass-1",
+        "domains": domains or [],
+    })
+
+
+def _tenant_admin_token(client, username, old_password, new_password="tn-pass-1"):
+    """租户管理员登录 + 首登改密(过 409 门),返回可用 token。"""
+    r = client.post(LOGIN, json={"username": username, "password": old_password})
+    assert r.status_code == 200
+    token = r.json()["token"]
+    r = client.post(CHANGE_PW, headers=_bearer(token), json={
+        "old_password": old_password, "new_password": new_password})
+    assert r.status_code == 200
+    return token
+
+
+def _db_fetch(root, stmt):
+    """测试线程直查 platform.db:独立 NullPool engine,同一 asyncio.run 内
+    查完即 dispose(跨循环纪律,见模块 docstring)。"""
+    engine = create_async_engine(
+        f"sqlite+aiosqlite:///{(root / 'platform.db').as_posix()}",
+        poolclass=NullPool)
+
+    async def _run():
+        try:
+            async with AsyncSession(engine) as session:
+                return list((await session.execute(stmt)).all())
+        finally:
+            await engine.dispose()
+
+    return asyncio.run(_run())
+
+
+def _current_snapshot(root, slug):
+    """该租户当前生效版本(MAX)的 config_json 解析。"""
+    rows = _db_fetch(root, select(TenantConfigVersion)
+                     .join(Tenant, TenantConfigVersion.tenant_id == Tenant.id)
+                     .where(Tenant.slug == slug)
+                     .order_by(TenantConfigVersion.version.desc()))
+    assert rows, f"租户 {slug} 没有配置版本"
+    return json.loads(rows[0][0].config_json)
+
+
+def _jc_snapshot():
+    return FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot
+
+
+# ---- 核心 6 项 ------------------------------------------------------------------
+
+
+def test_create_tenant_from_profile(super_client, db_source_app, tmp_path):
+    r = _create(super_client, slug="demo", profile="jiancai",
+                domains=["demo.example.com"])
+    assert r.status_code == 201
+    body = r.json()
+    assert body["slug"] == "demo"
+    assert body["status"] == "active"
+    assert body["version"] == 1
+    assert body["site"] == {"slug": "main", "domains": ["demo.example.com"]}
+
+    # platform.db: tenant + main 站点 + 版本1(config = seed 档案快照)
+    rows = _db_fetch(tmp_path, select(Tenant).where(Tenant.slug == "demo"))
+    assert len(rows) == 1
+    tenant = rows[0][0]
+    assert tenant.status == "active"
+    sites = _db_fetch(tmp_path, select(Site).where(Site.tenant_id == tenant.id))
+    assert [(row[0].slug, row[0].name) for row in sites] == [("main", "main")]
+    versions = _db_fetch(
+        tmp_path, select(TenantConfigVersion)
+        .where(TenantConfigVersion.tenant_id == tenant.id))
+    assert len(versions) == 1
+    assert versions[0][0].version == 1
+    assert json.loads(versions[0][0].config_json) == _jc_snapshot()
+    domains = _db_fetch(
+        tmp_path, select(SiteDomain).join(Site, SiteDomain.site_id == Site.id)
+        .where(Site.tenant_id == tenant.id))
+    assert [row[0].domain for row in domains] == ["demo.example.com"]
+
+    # 租户管理员可登录(role=tenant, tenant_slug=demo, 引导改密标志)
+    admins = _db_fetch(
+        tmp_path, select(AdminUser).where(AdminUser.tenant_id == tenant.id))
+    assert len(admins) == 1
+    admin = admins[0][0]
+    assert admin.username == "demo-admin"
+    assert admin.must_change_password is True
+    assert admin.password_hash.startswith("scrypt$")   # 哈希在库,不在响应
+    r = super_client.post(LOGIN, json={
+        "username": "demo-admin", "password": "demo-pass-1"})
+    assert r.status_code == 200
+    data = r.json()
+    assert data["role"] == "tenant"
+    assert data["tenant_slug"] == "demo"
+    assert data["must_change_password"] is True
+
+    # 创建后立即可用: reload upsert 后经中间件 resolve 命中(DB 源闭环)
+    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
+    assert r.status_code == 200
+    assert r.json()["slug"] == "demo"
+
+
+def test_create_blank_tenant(super_client, tmp_path):
+    r = _create(super_client, slug="blank1", profile="blank",
+                domains=["blank.example.com"])
+    assert r.status_code == 201
+    # 版本1快照: 身份组清零,wall/prompts 取 jiancai 档案
+    snapshot = _current_snapshot(tmp_path, "blank1")
+    jc = _jc_snapshot()
+    db_brand = DEFAULT_CONFIG["brand"]
+    assert snapshot["brand"]["title"] == db_brand["title"]
+    assert snapshot["brand"]["whatsapp"] == db_brand["whatsapp"] == ""
+    assert snapshot["brand"]["email"] == db_brand["email"] == ""
+    assert snapshot["brand"]["privacy_url"] == db_brand["privacy_url"] == ""
+    assert snapshot["push"]["prefix"] == ""
+    assert snapshot["engine"]["context_seed_keys"] == {}
+    # 非身份字段与 wall/prompts 保留(jiancai 档案)
+    assert snapshot["brand"]["color"] == jc["brand"]["color"]
+    assert snapshot["brand"]["topics"] == jc["brand"]["topics"]
+    assert snapshot["wall"] == jc["wall"]
+    assert snapshot["prompts"] == jc["prompts"]
+    # wall/prompts 可消费(compile_wall / PromptProvider 能过)
+    compile_wall(snapshot["wall"])
+    PromptProvider.from_config(snapshot["prompts"])
+    # E2E: blank 租户对外门面是出厂身份
+    r = super_client.get("/api/public/brand", headers={"host": "blank.example.com"})
+    assert r.status_code == 200
+    assert r.json()["title"] == db_brand["title"]
+    assert r.json()["whatsapp"] == ""
+
+
+def test_slug_immutable(super_client, tmp_path):
+    _create(super_client, slug="demo", name="原名")
+    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                           json={"slug": "demo2", "name": "新名"})
+    assert r.status_code == 400
+    assert "slug" in r.json()["detail"]
+    # 400 先于字段应用: 名字没被捎带改掉,库里也只有原租户
+    rows = _db_fetch(tmp_path, select(Tenant))
+    assert [(row[0].slug, row[0].name) for row in rows] == [("demo", "原名")]
+
+
+def test_add_domain_and_resolve(super_client):
+    _create(super_client, slug="demo")            # domains=[]
+    # 加域名前: 未知 Host 404
+    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
+    assert r.status_code == 404
+    r = super_client.post(f"{TENANTS}/demo/sites/main/domains",
+                          headers=_auth(super_client),
+                          json={"domain": "demo.example.com"})
+    assert r.status_code == 201
+    # registry reload 后 resolve 命中
+    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
+    assert r.status_code == 200
+    assert r.json()["slug"] == "demo"
+
+
+def test_disable_tenant_releases(super_client, db_source_app):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    assert super_client.get(
+        "/api/public/brand", headers={"host": "demo.example.com"}).status_code == 200
+    registry = db_source_app.state.registry
+    rt_old = registry._by_slug["demo"]
+    assert rt_old.engine is not None       # 懒建已发生(portal 循环内)
+
+    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                           json={"status": "disabled"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "disabled"
+    assert rt_old.engine is None           # reload 路径已 dispose 旧 engine
+
+    # resolve 抛 TenantDisabledError(停用检查先于懒建,无 engine 副作用)
+    with pytest.raises(TenantDisabledError):
+        asyncio.run(registry.resolve_site("demo.example.com"))
+    # 中间件映射 503 中性页
+    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
+    assert r.status_code == 503
+
+    # 复启用 → resolve 恢复(新 runtime 重新懒建)
+    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                           json={"status": "active"})
+    assert r.status_code == 200
+    assert super_client.get(
+        "/api/public/brand", headers={"host": "demo.example.com"}).status_code == 200
+
+
+def test_tenant_admin_cannot_create_tenant(super_client):
+    _create(super_client, slug="demo", admin="demo-admin",
+            password="demo-pass-1")
+    token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
+    r = super_client.post(TENANTS, headers=_bearer(token), json={
+        "name": "越权租户", "slug": "evil", "profile": "jiancai",
+        "admin_username": "evil-admin", "admin_password": "evil-pass-1"})
+    assert r.status_code == 403
+    # 对照: 超管可建
+    assert _create(super_client, slug="ok1").status_code == 201
+
+
+# ---- _blankify 单测 --------------------------------------------------------------
+
+
+def test_blankify_strips_identity_keeps_wall_prompts():
+    jc = _jc_snapshot()
+    out = _blankify(jc)
+    assert out is not jc                                  # 不改入参
+    assert jc["brand"]["title"] == "HESUPIPE Support"     # 入参原样
+    db_brand = DEFAULT_CONFIG["brand"]
+    for key in ("title", "whatsapp", "email", "privacy_url"):
+        assert out["brand"][key] == db_brand[key]
+    assert out["push"]["prefix"] == ""
+    assert out["engine"]["context_seed_keys"] == {}
+    # 非身份字段保留
+    assert out["brand"]["color"] == jc["brand"]["color"]
+    assert out["brand"]["welcome"] == jc["brand"]["welcome"]
+    assert out["brand"]["topics"] == jc["brand"]["topics"]
+    assert out["email"] == jc["email"]
+    # wall/prompts 完整可消费
+    assert out["wall"] == jc["wall"]
+    assert out["prompts"] == jc["prompts"]
+    compile_wall(out["wall"])
+    PromptProvider.from_config(out["prompts"])
+
+
+# ---- 校验与冲突 -----------------------------------------------------------------
+
+
+def test_slug_validation_and_collision(super_client, tmp_path):
+    assert _create(super_client, slug="Bad_Slug").status_code == 400
+    assert _create(super_client, slug="下划线_x").status_code == 400
+    assert _create(super_client, slug="demo").status_code == 201
+    r = _create(super_client, slug="demo")
+    assert r.status_code == 400                           # 撞现有租户
+    assert "demo" in r.json()["detail"]
+    # 管理员用户名全局唯一
+    r = _create(super_client, slug="other", admin="demo-admin")
+    assert r.status_code == 400
+    # 未知 profile
+    assert _create(super_client, slug="x9", profile="vip").status_code == 400
+
+
+def test_create_auto_profile(super_client, tmp_path):
+    r = _create(super_client, slug="auto2", profile="auto",
+                domains=["auto2.example.com"])
+    assert r.status_code == 201
+    assert _current_snapshot(tmp_path, "auto2") == FileTenantSource(
+        SEED_DIR).read_tenant("auto").snapshot
+    r = super_client.get("/api/public/brand", headers={"host": "auto2.example.com"})
+    assert r.status_code == 200
+    assert r.json()["title"] == "HESU Support"            # auto 档案身份
+
+
+def test_domain_collision_reports_owner(super_client):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    _create(super_client, slug="other")
+    # 大写归一后同域 → 400 带现有归属(tenant + site)
+    r = super_client.post(f"{TENANTS}/other/sites/main/domains",
+                          headers=_auth(super_client),
+                          json={"domain": "Demo.Example.com"})
+    assert r.status_code == 400
+    detail = r.json()["detail"]
+    assert "demo" in detail and "main" in detail
+    # 创建向导的初始域名同样预检
+    r = _create(super_client, slug="third", domains=["demo.example.com"])
+    assert r.status_code == 400
+
+
+def test_delete_domain(super_client):
+    _create(super_client, slug="demo",
+            domains=["a.example.com", "b.example.com"])
+    r = super_client.delete(f"{TENANTS}/demo/sites/main/domains/a.example.com",
+                            headers=_auth(super_client))
+    assert r.status_code == 200
+    assert super_client.get(
+        "/api/public/brand", headers={"host": "a.example.com"}).status_code == 404
+    assert super_client.get(
+        "/api/public/brand", headers={"host": "b.example.com"}).status_code == 200
+    # 重复删/不存在 → 404
+    assert super_client.delete(
+        f"{TENANTS}/demo/sites/main/domains/a.example.com",
+        headers=_auth(super_client)).status_code == 404
+    assert super_client.delete(
+        f"{TENANTS}/demo/sites/main/domains/nope.example.com",
+        headers=_auth(super_client)).status_code == 404
+
+
+# ---- 站点 CRUD / brand_overlay 整赋值 ---------------------------------------------
+
+
+def test_site_create_patch_overlay(super_client, tmp_path):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    r = super_client.post(f"{TENANTS}/demo/sites", headers=_auth(super_client),
+                          json={"slug": "es", "name": "ES 站",
+                                "brand_overlay": {"title": "Demo ES",
+                                                  "welcome": "Hola"},
+                                "domains": ["es.demo.example.com"]})
+    assert r.status_code == 201
+    # 域名直连该站点(reload 后 resolve 命中站点)
+    r = super_client.get("/api/public/brand", headers={"host": "es.demo.example.com"})
+    assert r.status_code == 200
+    assert r.json()["title"] == "Demo ES"
+
+    # PATCH 改名 + overlay 整字典赋值(M-2a: 换 dict,旧键不残留)
+    r = super_client.patch(f"{TENANTS}/demo/sites/es",
+                           headers=_auth(super_client),
+                           json={"name": "ES2",
+                                 "brand_overlay": {"title": "Demo ES v2"}})
+    assert r.status_code == 200
+    rows = _db_fetch(tmp_path, select(Site)
+                     .join(Tenant, Site.tenant_id == Tenant.id)
+                     .where(Tenant.slug == "demo", Site.slug == "es"))
+    site = rows[0][0]
+    assert site.name == "ES2"
+    assert site.brand_overlay == {"title": "Demo ES v2"}   # welcome 键没了
+    r = super_client.get("/api/public/brand", headers={"host": "es.demo.example.com"})
+    assert r.json()["title"] == "Demo ES v2"
+    # welcome 回退租户级 brand(覆盖层不再含它)
+    assert r.json()["welcome"] != "Hola"
+
+    # 站点 slug 不可改;站点内 slug 撞车 400;未知站点 404
+    assert super_client.patch(
+        f"{TENANTS}/demo/sites/es", headers=_auth(super_client),
+        json={"slug": "es2"}).status_code == 400
+    assert super_client.post(
+        f"{TENANTS}/demo/sites", headers=_auth(super_client),
+        json={"slug": "es"}).status_code == 400
+    assert super_client.patch(
+        f"{TENANTS}/demo/sites/ghost", headers=_auth(super_client),
+        json={"name": "x"}).status_code == 404
+
+    # 站点停用持久化(解析级语义后续任务接线,此处钉字段落库+reload 不炸)
+    r = super_client.patch(f"{TENANTS}/demo/sites/es",
+                           headers=_auth(super_client),
+                           json={"status": "disabled"})
+    assert r.status_code == 200
+    rows = _db_fetch(tmp_path, select(Site)
+                     .join(Tenant, Site.tenant_id == Tenant.id)
+                     .where(Tenant.slug == "demo", Site.slug == "es"))
+    assert rows[0][0].status == "disabled"
+
+
+def test_patch_tenant_name_and_validation(super_client, tmp_path):
+    _create(super_client, slug="demo", name="原名")
+    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                           json={"name": "新名"})
+    assert r.status_code == 200
+    rows = _db_fetch(tmp_path, select(Tenant).where(Tenant.slug == "demo"))
+    assert rows[0][0].name == "新名"
+    # 非法 status / 空补丁 / 未知租户
+    assert super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                              json={"status": "paused"}).status_code == 400
+    assert super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
+                              json={}).status_code == 400
+    assert super_client.patch(f"{TENANTS}/ghost", headers=_auth(super_client),
+                              json={"name": "x"}).status_code == 404
+
+
+# ---- 列表/详情/凭据不外泄/403 边界 -------------------------------------------------
+
+
+def test_list_tenants(super_client):
+    _create(super_client, slug="demo",
+            domains=["a.example.com", "b.example.com"])
+    _create(super_client, slug="other")
+    r = super_client.get(TENANTS, headers=_auth(super_client))
+    assert r.status_code == 200
+    items = {i["slug"]: i for i in r.json()["items"]}
+    assert set(items) == {"demo", "other"}
+    assert items["demo"]["status"] == "active"
+    assert items["demo"]["domain_count"] == 2
+    assert items["demo"]["current_version"] == 1
+    assert items["demo"]["admins"] == [
+        {"username": "demo-admin", "status": "active"}]
+    assert items["other"]["domain_count"] == 0
+
+
+def test_detail_shape_and_scope(super_client):
+    _create(super_client, slug="demo", admin="demo-admin",
+            password="demo-pass-1", domains=["demo.example.com"])
+    _create(super_client, slug="other")
+    r = super_client.get(f"{TENANTS}/demo", headers=_auth(super_client))
+    assert r.status_code == 200
+    d = r.json()
+    assert d["slug"] == "demo"
+    assert d["current_version"] == 1
+    assert [s["slug"] for s in d["sites"]] == ["main"]
+    assert d["sites"][0]["domains"] == ["demo.example.com"]
+    assert d["admins"] == [{"username": "demo-admin", "status": "active"}]
+
+    # 详情放宽到本租户管理员;看别的租户 403;列表(超管专属)403
+    token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
+    assert super_client.get(
+        f"{TENANTS}/demo", headers=_bearer(token)).status_code == 200
+    assert super_client.get(
+        f"{TENANTS}/other", headers=_bearer(token)).status_code == 403
+    assert super_client.get(
+        TENANTS, headers=_bearer(token)).status_code == 403
+    # 未认证 401
+    assert super_client.get(f"{TENANTS}/demo").status_code == 401
+
+
+def test_no_credentials_leak_in_responses(super_client):
+    """密码哈希/salt/明文密码绝不出现在列表与详情响应(M-5 红线)。"""
+    _create(super_client, slug="demo", admin="demo-admin",
+            password="demo-pass-1", domains=["demo.example.com"])
+    for r in (super_client.get(TENANTS, headers=_auth(super_client)),
+              super_client.get(f"{TENANTS}/demo", headers=_auth(super_client))):
+        assert r.status_code == 200
+        for needle in ("password_hash", "salt", "scrypt", "demo-pass-1"):
+            assert needle not in r.text, f"响应泄露凭据字段: {needle}"