feat: 配置API（分组校验管线/整份版本/回滚/重置出厂/试金石/平台设置）

- config_validation.validate_group: 九组校验管线(类型/键 allowlist/数值
  范围/prompts 大小上限 50KB·100KB/wall 全量交 compile_wall/push.prefix
  跨租户查重 spec E1),错误清单中文带字段路径
- config_routes: GET/PUT config(整组替换原样入版本,MAX+1,热 reload)/
  reset(按 Tenant.profile 取出厂档案组)/versions 列表·详情/rollback
  (旧版复制为新版本,历史不可变)/wall/test 试金石(逐句判定,不落库
  不 bump)/settings 平台设置 CRUD(仅超管,resolver 接线归 Phase 3)
- db.write_config_version: 版本写入公共路径(创建向导与配置 API 单一
  事实源);Tenant 落 profile 列(重置出厂按它取对应档案)
- _profile_snapshot/_blankify 迁 tenant_source(两路由模块共用);
  db_source_app/super_client 夹具收拢 conftest(两测试文件共用)
- 测试: 519 → 537 全绿

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

app/main.py

@@ -81,6 +81,10 @@ 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.config_routes import (
+    router as platform_config_router,
+    settings_router as platform_settings_router,
+)
 from app.platform.routes.tenant_routes import router as platform_tenant_router
 app.include_router(chat_router)
 app.include_router(chat_rest_router)
@@ -88,6 +92,11 @@ app.include_router(public_router)
 # 租户管理 API(Task 5):创建向导/站点/域名/停用——同样只依赖路由层认证
 # (require_super / require_tenant),Host 豁免面归 Task 3 的前缀规则覆盖。
 app.include_router(platform_tenant_router)
+# 配置 API(Task 6):分组校验管线/整份版本/回滚/试金石/重置出厂(鉴权
+# require_tenant,租户全开)+ 平台设置(require_super)——/api/platform/
+# 前缀的 Host 豁免由 Task 3 的中间件规则统一覆盖。
+app.include_router(platform_config_router)
+app.include_router(platform_settings_router)
 # 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下
 # /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为
 # /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析);

app/platform/config_validation.py

@@ -0,0 +1,256 @@
+"""配置校验管线(Phase 2 Task 6)——PUT /config/{group} 的分组校验。
+
+validate_group(group, payload, context) -> list[str]:错误清单(中文,带
+字段路径),空清单 = 通过;路由层把非空清单拼进 400 detail。
+
+纪律:
+- **键 allowlist**:各组未知键显性拒绝(与 wall_schema._KNOWN_FIELDS 同款
+  防呆——配置编辑器时代拼错键 = 深合并静默忽略,这里前移到保存期报错)。
+  wall 组例外:未知键交 compile_wall 管(它的报错自带 wall 侧字段路径)。
+- **单一事实源**:各组合法键从 DEFAULT_CONFIG 派生(prompts 除外——
+  DEFAULT 不含 prompts,取 ai_service.PROMPTS_KEYS 四件套)。
+- **wall 组不重复造校验**:深拷贝 payload 交 compile_wall,复用全部既有
+  校验(键 allowlist/占位符白名单/flags/正则编译/必填池)。
+- **prompts 组**:四键字符串 + 大小上限(company_facts ≤50KB 软上限,
+  spec §5.7——每轮整段拼入 system prompt,过大门线性放大 LLM 成本;
+  其余三件 ≤100KB)。字节口径按 UTF-8 编码长度。
+- **push.prefix 跨租户查重**(spec E1,external_key 前缀全局唯一防 CRM
+  线索空间串号):context.other_prefixes 由路由层从 DB 各租户当前版本
+  收集(排除自身;空串 = 未设置,不参战——出厂默认本就多租户同空)。
+- 数值字段(None = "未设置"哨兵,与 DEFAULT_CONFIG 同款纪律):None 或
+  范围内数值;bool 不是 int(isinstance 陷阱,显式排除)。
+"""
+import copy
+from dataclasses import dataclass, field
+
+from app.engine.services.ai_service import PROMPTS_KEYS
+from app.engine.wall_schema import WallConfigError, compile_wall
+from app.platform.defaults import DEFAULT_CONFIG
+
+# 各组合法键 = DEFAULT_CONFIG 同组键(单一事实源);prompts 特例见上。
+_ALLOWED_KEYS: dict[str, frozenset] = {
+    group: frozenset(keys) for group, keys in DEFAULT_CONFIG.items()}
+_ALLOWED_KEYS["prompts"] = frozenset(PROMPTS_KEYS)
+
+# 合法组名(排序输出进错误信息,稳定可测)
+CONFIG_GROUPS = frozenset(_ALLOWED_KEYS)
+
+# prompts 大小上限(字节,UTF-8 口径):company_facts 50KB 软上限,其余 100KB
+PROMPTS_SIZE_CAPS = {
+    name: 50 * 1024 if name == "company_facts" else 100 * 1024
+    for name in PROMPTS_KEYS}
+
+_BRAND_STR_KEYS = ("title", "color", "avatar", "welcome", "placeholder",
+                   "whatsapp", "email", "privacy_url")
+
+
+@dataclass(frozen=True)
+class ValidationContext:
+    """跨租户校验上下文(路由层按请求现势收集,校验管线不自查 DB)。"""
+    other_prefixes: frozenset = field(default_factory=frozenset)
+
+
+def _type_name(value) -> str:
+    return type(value).__name__
+
+
+def _want_str(value, path: str, errors: list) -> None:
+    if not isinstance(value, str):
+        errors.append(f"{path}: 须为字符串(实际 {_type_name(value)})")
+
+
+def _want_bool(value, path: str, errors: list) -> None:
+    if not isinstance(value, bool):
+        errors.append(f"{path}: 须为布尔(实际 {_type_name(value)})")
+
+
+def _want_int(value, path: str, errors: list) -> None:
+    # bool 是 int 的子类——True 会被 isinstance 放过,显式排除
+    if not isinstance(value, int) or isinstance(value, bool):
+        errors.append(f"{path}: 须为整数(实际 {_type_name(value)})")
+
+
+def _want_str_list(value, path: str, errors: list) -> None:
+    if not isinstance(value, list):
+        errors.append(f"{path}: 须为字符串列表(实际 {_type_name(value)})")
+        return
+    for i, item in enumerate(value):
+        if not isinstance(item, str):
+            errors.append(f"{path}[{i}]: 项须为字符串(实际 {_type_name(item)})")
+
+
+def _check_brand(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    for key in _BRAND_STR_KEYS:
+        if key in payload:
+            _want_str(payload[key], f"brand.{key}", errors)
+    if "topics" in payload:
+        _want_str_list(payload["topics"], "brand.topics", errors)
+
+
+def _check_wall(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    """全量交 compile_wall(深拷贝防意外 mutation):键 allowlist/占位符/
+    flags/正则编译/必填池全套复用,WallConfigError 转清单项。"""
+    try:
+        compile_wall(copy.deepcopy(payload))
+    except WallConfigError as exc:
+        errors.append(str(exc))
+
+
+def _check_prompts(payload: dict, ctx: ValidationContext,
+                   errors: list) -> None:
+    for key in PROMPTS_KEYS:
+        if key not in payload:
+            errors.append(f"prompts.{key}: 缺少必填键(四件套须整组提交)")
+            continue
+        text = payload[key]
+        if not isinstance(text, str):
+            errors.append(f"prompts.{key}: 须为字符串(实际 {_type_name(text)})")
+            continue
+        size = len(text.encode("utf-8"))
+        cap = PROMPTS_SIZE_CAPS[key]
+        if size > cap:
+            errors.append(
+                f"prompts.{key}: 内容 {size} 字节超过上限 {cap} 字节"
+                f"({cap // 1024}KB,spec §5.7——每轮拼入 system prompt,"
+                f"请精炼)")
+
+
+def _check_contact(payload: dict, ctx: ValidationContext,
+                   errors: list) -> None:
+    for key in ("cc_completion", "email_light_ask"):
+        if key in payload:
+            _want_bool(payload[key], f"contact.{key}", errors)
+    if "template_languages" in payload:
+        _want_str_list(payload["template_languages"],
+                       "contact.template_languages", errors)
+
+
+def _check_email(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    if "sender_name" in payload:
+        _want_str(payload["sender_name"], "email.sender_name", errors)
+    if "smtp" in payload and not (payload["smtp"] is None
+                                  or isinstance(payload["smtp"], dict)):
+        errors.append(f"email.smtp: 须为映射或 null(实际 "
+                      f"{_type_name(payload['smtp'])})")
+    if "notify_emails" in payload:
+        value = payload["notify_emails"]
+        if not isinstance(value, list):
+            errors.append(f"email.notify_emails: 须为列表(实际 "
+                          f"{_type_name(value)})")
+            return
+        for i, item in enumerate(value):
+            if not isinstance(item, str) or "@" not in item:
+                errors.append(f"email.notify_emails[{i}]: 项须为含 @ 的邮箱"
+                              f"地址(实际 {item!r})")
+
+
+def _check_push(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    if "enabled" in payload:
+        _want_bool(payload["enabled"], "push.enabled", errors)
+    for key in ("url", "api_key", "prefix"):
+        if key in payload:
+            _want_str(payload[key], f"push.{key}", errors)
+    if "timeout" in payload:
+        _want_positive_number(payload["timeout"], "push.timeout", errors)
+    if "retry_backoff_base" in payload:
+        _want_positive_number(payload["retry_backoff_base"],
+                              "push.retry_backoff_base", errors)
+    if "max_attempts" in payload:
+        value = payload["max_attempts"]
+        if value is not None and (not isinstance(value, int)
+                                  or isinstance(value, bool)
+                                  or not 1 <= value <= 10):
+            errors.append(f"push.max_attempts: 须为 1-10 整数或 null(实际 "
+                          f"{value!r})")
+    # spec E1: external_key 前缀全局唯一(空串 = 未设置,不参战)
+    prefix = payload.get("prefix")
+    if isinstance(prefix, str) and prefix and prefix in ctx.other_prefixes:
+        errors.append(f"push.prefix: {prefix!r} 已被其他租户占用"
+                      f"(external_key 前缀全局唯一)")
+
+
+def _want_positive_number(value, path: str, errors: list) -> None:
+    if value is None:
+        return
+    if (not isinstance(value, (int, float)) or isinstance(value, bool)
+            or value <= 0):
+        errors.append(f"{path}: 须为正数或 null(实际 {value!r})")
+
+
+def _check_ai(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    for key in ("provider", "model"):
+        if key in payload:
+            _want_str(payload[key], f"ai.{key}", errors)
+    for key in ("api_key", "base_url"):
+        if key in payload:
+            value = payload[key]
+            if value is not None and not isinstance(value, str):
+                errors.append(f"ai.{key}: 须为字符串或 null(实际 "
+                              f"{_type_name(value)})")
+
+
+def _check_engine(payload: dict, ctx: ValidationContext,
+                  errors: list) -> None:
+    for key in ("rate_limit_per_hour", "ws_conn_limit"):
+        if key in payload:
+            value = payload[key]
+            _want_int(value, f"engine.{key}", errors)
+            if isinstance(value, int) and not isinstance(value, bool) \
+                    and value < 1:
+                errors.append(f"engine.{key}: 须为正整数(实际 {value})")
+    if "context_seed_keys" in payload:
+        value = payload["context_seed_keys"]
+        if not isinstance(value, dict):
+            errors.append(f"engine.context_seed_keys: 须为映射(实际 "
+                          f"{_type_name(value)})")
+            return
+        for k, v in value.items():
+            if not isinstance(v, str):
+                errors.append(f"engine.context_seed_keys.{k}: 须为字符串"
+                              f"(实际 {_type_name(v)})")
+
+
+def _check_misc(payload: dict, ctx: ValidationContext, errors: list) -> None:
+    if "name_to_region_extras" in payload:
+        value = payload["name_to_region_extras"]
+        if not isinstance(value, dict):
+            errors.append(f"misc.name_to_region_extras: 须为映射(实际 "
+                          f"{_type_name(value)})")
+            return
+        for k, v in value.items():
+            if not isinstance(v, str):
+                errors.append(f"misc.name_to_region_extras.{k}: 须为字符串"
+                              f"(实际 {_type_name(v)})")
+
+
+_GROUP_CHECKERS = {
+    "brand": _check_brand,
+    "wall": _check_wall,
+    "contact": _check_contact,
+    "email": _check_email,
+    "push": _check_push,
+    "ai": _check_ai,
+    "engine": _check_engine,
+    "misc": _check_misc,
+    "prompts": _check_prompts,
+}
+
+
+def validate_group(group: str, payload, context: ValidationContext) -> list:
+    """校验单个配置组的 payload(整组替换语义:只查出现键的类型/形状,
+    缺省键由消费侧深合并出厂默认补)。返回错误清单(中文,带字段路径),
+    空清单 = 通过。未知组名/非对象 payload 也走清单(路由统一转 400)。"""
+    checker = _GROUP_CHECKERS.get(group)
+    if checker is None:
+        return [f"未知配置组: {group!r}(合法: "
+                f"{', '.join(sorted(CONFIG_GROUPS))})"]
+    if not isinstance(payload, dict):
+        return [f"{group}: 须为 JSON 对象(实际 {_type_name(payload)})"]
+    errors: list = []
+    if group != "wall":    # wall 的未知键交 compile_wall(报错自带其字段路径)
+        unknown = set(payload) - _ALLOWED_KEYS[group]
+        if unknown:
+            errors.append(f"{group}: 未知键 {', '.join(sorted(unknown))}"
+                          f"(合法: {', '.join(sorted(_ALLOWED_KEYS[group]))})")
+    checker(payload, context, errors)
+    return errors

app/platform/db.py

@@ -26,6 +26,7 @@ PlatformBase.metadata,engine/models.ensure_schema 只跑 Base.metadata——
 (Windows 文件锁,同 registry)。
 """
 import datetime
+import json
 import pathlib
 
 from sqlalchemy import (
@@ -38,6 +39,7 @@ from sqlalchemy import (
     Text,
     UniqueConstraint,
     event,
+    func,
     select,
 )
 from sqlalchemy.exc import IntegrityError
@@ -69,6 +71,10 @@ class Tenant(PlatformBase):
     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")
+    # 创建向导选定的出厂档案(Task 6 起):重置出厂按它取对应档案组
+    # (auto/jiancai 档案 / blank = jiancai 去身份组)。缺省 blank——
+    # 正常路径一律经创建向导写入,缺省只是手工插行的兜底语义。
+    profile: Mapped[str] = mapped_column(String(16), default="blank")
     created_at: Mapped[datetime.datetime] = mapped_column(default=_utcnow)
 
 
@@ -176,6 +182,29 @@ async def ensure_platform_schema(engine: AsyncEngine) -> None:
 
 # ---- 密码哈希:定义在 auth.py(见文件头 re-export)---------------------------
 
+# ---- 配置版本写入(Task 6 抽公共:创建向导与配置 API 的单一版本写路径) --------
+
+
+async def write_config_version(
+        session: AsyncSession, tenant_id: int, snapshot: dict, *,
+        updated_by: str = "", comment: str | None = None) -> int:
+    """写新配置版本:MAX(version)+1,整份快照直存(spec §5.2 历史不可变,
+    当前生效 = MAX)。Task 5 创建向导(版本 1)与 Task 6 配置 API(PUT 分组
+    编辑/rollback 复制旧版/reset 重置出厂)的公共写入路径——两处各自
+    MAX+1 会漂移,收敛为单一事实源。调用方负责 commit;返回新版本号。"""
+    current = (await session.execute(
+        select(func.max(TenantConfigVersion.version))
+        .where(TenantConfigVersion.tenant_id == tenant_id))).scalar()
+    version = (current or 0) + 1
+    session.add(TenantConfigVersion(
+        tenant_id=tenant_id,
+        version=version,
+        config_json=json.dumps(snapshot, ensure_ascii=False),
+        updated_by=updated_by,
+        comment=comment))
+    return version
+
+
 # ---- 超管引导 ---------------------------------------------------------------
 
 async def bootstrap_superuser(

app/platform/routes/config_routes.py

@@ -0,0 +1,433 @@
+"""配置 API(Phase 2 Task 6):/api/platform/tenants/{slug}/config* 与
+/api/platform/settings。
+
+能力面(校验管线见 config_validation.py):
+- GET  /config:当前生效快照(MAX 版本)+ version。返回**原样快照**——
+  DEFAULT_CONFIG 不入版本(出厂模板演进不被旧版本锁死,spec E4),消费侧
+  (registry → TenantConfig.from_snapshot)深合并;
+- PUT  /config/{group}:校验 → 读当前快照 → **整组替换** → write_config_version
+  (MAX+1)→ registry_reload 热生效。updated_by 取认证主体(super#id /
+  tenant#id),comment 走查询参数(可选);
+- POST /config/{group}/reset:重置出厂 = 代码库当前模板(spec §5.6,随平台
+  升级演进)——按 Tenant.profile 取 _profile_snapshot 的对应组,档案缺的组
+  取 DEFAULT_CONFIG 整组;
+- GET  /config/versions(列表:版本/时间/作者/备注,不含 config 本体)与
+  /config/versions/{n}(详情含整份);
+- POST /config/rollback {version}:目标版本复制为新版本(历史不可变),
+  reload 生效;
+- POST /wall/test(试金石):body 可带 wall 覆盖(先 compile_wall 校验)或
+  缺省用当前墙——逐句调 boundary_filter 纯函数给判定,**不落库不 bump 版本**
+  (reset/编辑器的预览入口)。判定优先级与 conversation.py 执行侧一致:
+  价格墙 → 零号码墙 → 再索要墙(仅 has_contact);
+- GET/PUT /api/platform/settings:PlatformSetting CRUD,仅超管。本阶段只落
+  存储与接口(value 以 JSON Text 直存,读回还原类型);resolver 的平台默认
+  回退链接入归 Phase 3。
+
+鉴权矩阵(spec §3):config 系 = require_tenant(slug)(租户全开——本租户
+管理员可自助改配置);settings = require_super。409 门经 platform_user 连锁。
+
+版本写入一律走 db.write_config_version(与 Task 5 创建向导同一 MAX+1 路径)。
+"""
+import copy
+import json
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.engine.services import boundary_filter
+from app.engine.wall_schema import WallConfigError, compile_wall
+from app.platform.auth import UserContext, require_super
+from app.platform.config_validation import (
+    CONFIG_GROUPS,
+    ValidationContext,
+    validate_group,
+)
+from app.platform.db import (
+    PlatformSetting,
+    Tenant,
+    TenantConfigVersion,
+    write_config_version,
+)
+from app.platform.defaults import DEFAULT_CONFIG
+from app.platform.routes.tenant_routes import _get_tenant_or_404, _tenant_guard
+from app.platform.tenant_source import _profile_snapshot
+
+router = APIRouter(prefix="/api/platform/tenants", tags=["platform-config"])
+settings_router = APIRouter(
+    prefix="/api/platform/settings", tags=["platform-settings"])
+
+
+# ---- 助手 -----------------------------------------------------------------------
+
+
+async def _json_body(request: Request) -> dict:
+    """手动解析 JSON 对象请求体:非 JSON/非对象一律 400(带实际类型),
+    不让 FastAPI 的 422 抢答——配置 API 的错误契约是"400 带字段路径"。"""
+    try:
+        body = await request.json()
+    except ValueError as exc:
+        raise HTTPException(status_code=400,
+                            detail="请求体必须是合法 JSON 对象") from exc
+    if not isinstance(body, dict):
+        raise HTTPException(
+            status_code=400,
+            detail=f"请求体必须是 JSON 对象(实际 {type(body).__name__})")
+    return body
+
+
+async def _current_version_row(
+        session: AsyncSession, tenant_id: int) -> TenantConfigVersion | None:
+    """当前生效版本行(= MAX(version),db.py 纪律)。"""
+    return (await session.execute(
+        select(TenantConfigVersion)
+        .where(TenantConfigVersion.tenant_id == tenant_id)
+        .order_by(TenantConfigVersion.version.desc())
+        .limit(1))).scalar_one_or_none()
+
+
+async def _load_current(
+        session: AsyncSession, tenant_id: int, slug: str
+) -> tuple[TenantConfigVersion, dict]:
+    """当前生效版本行 + 解析后的快照。无版本(防御——创建向导必写版本 1)
+    → 404;坏 JSON(库层数据损坏)→ 500 显性报,不静默。"""
+    row = await _current_version_row(session, tenant_id)
+    if row is None:
+        raise HTTPException(
+            status_code=404, detail=f"租户 {slug!r} 没有任何配置版本")
+    try:
+        snapshot = json.loads(row.config_json)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=500,
+            detail=f"租户 {slug!r} 版本 {row.version} 的 config_json 损坏"
+        ) from exc
+    return row, snapshot
+
+
+async def _other_prefixes(
+        session: AsyncSession, exclude_tenant_id: int) -> frozenset:
+    """各租户**当前版本**的 push.prefix 集合(spec E1 查重数据源,排除自身;
+    空/坏数据不参战)。MAX(version) 子查询定位每租户当前版。"""
+    max_versions = (
+        select(TenantConfigVersion.tenant_id.label("tid"),
+               func.max(TenantConfigVersion.version).label("mv"))
+        .group_by(TenantConfigVersion.tenant_id).subquery())
+    rows = (await session.execute(
+        select(TenantConfigVersion.tenant_id, TenantConfigVersion.config_json)
+        .join(max_versions, (TenantConfigVersion.tenant_id == max_versions.c.tid)
+              & (TenantConfigVersion.version == max_versions.c.mv)))).all()
+    prefixes = set()
+    for tenant_id, config_json in rows:
+        if tenant_id == exclude_tenant_id:
+            continue
+        try:
+            snapshot = json.loads(config_json)
+        except ValueError:
+            continue
+        push = snapshot.get("push") if isinstance(snapshot, dict) else None
+        prefix = push.get("prefix") if isinstance(push, dict) else None
+        if isinstance(prefix, str) and prefix:
+            prefixes.add(prefix)
+    return frozenset(prefixes)
+
+
+def _updated_by(user: UserContext) -> str:
+    return f"{user.role}#{user.user_id}"
+
+
+# ---- 配置快照 / 分组编辑 ----------------------------------------------------------
+
+
+@router.get("/{slug}/config")
+async def get_config(slug: str, request: Request,
+                     user=Depends(_tenant_guard)):
+    """当前生效快照(MAX 版本)+ version。原样快照,非合并视图——缺省键
+    由消费侧(registry)深合并 DEFAULT_CONFIG。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        row, snapshot = await _load_current(session, tenant.id, slug)
+    return {"slug": slug, "version": row.version, "config": snapshot}
+
+
+@router.put("/{slug}/config/{group}")
+async def put_config_group(
+        slug: str, group: str, request: Request, comment: str | None = None,
+        user=Depends(_tenant_guard)):
+    """分组编辑:校验(config_validation 管线)→ 整组替换当前快照该组 →
+    MAX+1 写新版本 → registry_reload 热生效。快照存用户配置**原样组**
+    (组内缺省键消费侧深合并出厂默认补)。"""
+    body = await _json_body(request)
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        context = ValidationContext(
+            other_prefixes=await _other_prefixes(session, tenant.id))
+        errors = validate_group(group, body, context)
+        if errors:
+            raise HTTPException(status_code=400, detail="; ".join(errors))
+        _, current = await _load_current(session, tenant.id, slug)
+        new_snapshot = dict(current)
+        new_snapshot[group] = copy.deepcopy(body)
+        version = await write_config_version(
+            session, tenant.id, new_snapshot,
+            updated_by=_updated_by(user), comment=comment)
+        await session.commit()
+
+    await request.app.state.registry.registry_reload(slug)
+    return {"slug": slug, "group": group, "version": version,
+            "config": new_snapshot[group]}
+
+
+@router.post("/{slug}/config/{group}/reset")
+async def reset_config_group(
+        slug: str, group: str, request: Request, user=Depends(_tenant_guard)):
+    """重置出厂(spec §5.6 = 代码库当前模板,非创建时的版本 1 快照):
+    按租户创建 profile 取 _profile_snapshot 对应组;档案快照缺的组
+    (contact/misc/ai)取 DEFAULT_CONFIG 整组。产新版本 + reload。"""
+    if group not in CONFIG_GROUPS:
+        raise HTTPException(
+            status_code=400,
+            detail=f"未知配置组: {group!r}(合法: "
+                   f"{', '.join(sorted(CONFIG_GROUPS))})")
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        try:
+            factory = _profile_snapshot(tenant.profile)
+        except LookupError as exc:
+            raise HTTPException(
+                status_code=400, detail=f"出厂档案缺失: {exc}") from exc
+        if group in factory:
+            value = copy.deepcopy(factory[group])
+        else:
+            value = copy.deepcopy(DEFAULT_CONFIG[group])
+        _, current = await _load_current(session, tenant.id, slug)
+        new_snapshot = dict(current)
+        new_snapshot[group] = value
+        version = await write_config_version(
+            session, tenant.id, new_snapshot,
+            updated_by=_updated_by(user),
+            comment=f"重置出厂: {group}")
+        await session.commit()
+
+    await request.app.state.registry.registry_reload(slug)
+    return {"slug": slug, "group": group, "version": version, "config": value}
+
+
+# ---- 版本历史 / 回滚 ---------------------------------------------------------------
+
+
+@router.get("/{slug}/config/versions")
+async def list_config_versions(
+        slug: str, request: Request, user=Depends(_tenant_guard)):
+    """版本列表(升序):版本/时间/作者/备注——不含 config_json 本体
+    (列表几十 KB × N 版本会白白撑爆响应)。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        rows = (await session.execute(
+            select(TenantConfigVersion)
+            .where(TenantConfigVersion.tenant_id == tenant.id)
+            .order_by(TenantConfigVersion.version))).scalars().all()
+    return {"slug": slug, "items": [{
+        "version": row.version,
+        "created_at": row.created_at.isoformat(),
+        "updated_by": row.updated_by,
+        "comment": row.comment,
+    } for row in rows]}
+
+
+@router.get("/{slug}/config/versions/{version}")
+async def get_config_version(
+        slug: str, version: int, request: Request, user=Depends(_tenant_guard)):
+    """单版本详情(含整份 config)。不存在 → 404。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        row = (await session.execute(
+            select(TenantConfigVersion)
+            .where(TenantConfigVersion.tenant_id == tenant.id,
+                   TenantConfigVersion.version == version)
+        )).scalar_one_or_none()
+        if row is None:
+            raise HTTPException(
+                status_code=404, detail=f"版本不存在: {version}")
+    return {
+        "slug": slug, "version": row.version,
+        "created_at": row.created_at.isoformat(),
+        "updated_by": row.updated_by, "comment": row.comment,
+        "config": json.loads(row.config_json),
+    }
+
+
+@router.post("/{slug}/config/rollback")
+async def rollback_config(
+        slug: str, request: Request, user=Depends(_tenant_guard)):
+    """回滚:目标版本整份复制为新版本(历史不可变——旧版本行永不改写),
+    reload 生效。"""
+    body = await _json_body(request)
+    target = body.get("version")
+    if not isinstance(target, int) or isinstance(target, bool) or target < 1:
+        raise HTTPException(
+            status_code=400, detail="version: 须为正整数(实际 "
+                                    f"{target!r})")
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        row = (await session.execute(
+            select(TenantConfigVersion)
+            .where(TenantConfigVersion.tenant_id == tenant.id,
+                   TenantConfigVersion.version == target)
+        )).scalar_one_or_none()
+        if row is None:
+            raise HTTPException(
+                status_code=404, detail=f"版本不存在: {target}")
+        snapshot = json.loads(row.config_json)
+        new_version = await write_config_version(
+            session, tenant.id, snapshot,
+            updated_by=_updated_by(user),
+            comment=f"回滚自版本 {target}")
+        await session.commit()
+
+    await request.app.state.registry.registry_reload(slug)
+    return {"slug": slug, "rollback_to": target, "version": new_version,
+            "config": snapshot}
+
+
+# ---- 试金石(wall/test) -----------------------------------------------------------
+
+
+_WALL_TEST_KEYS = frozenset({"sentences", "wall", "has_contact", "source"})
+
+
+def _judge_sentence(sentence: str, wall, *, has_contact: bool,
+                    source: str) -> dict:
+    """单句判定——优先级与 conversation.py 执行侧逐句门控一致:
+    价格墙(掐断+替换)→ 零号码墙(替换)→ 再索要墙(仅已留资,静默丢弃)。
+    替换话术同款取自墙词表:deflection_for(首句形态,emitted 为空的分支)/
+    no_contact 按 has_contact·source 分支 / 再索要无替换(丢句不掐断)。"""
+    if boundary_filter.violates_boundary(sentence, wall=wall):
+        return {"sentence": sentence, "action": "blocked", "reason": "boundary",
+                "replacement": boundary_filter.deflection_for(
+                    sentence, has_contact=has_contact, wall=wall)}
+    if boundary_filter.no_contact_value(sentence, wall=wall):
+        replacement = (
+            wall.no_contact["already"] if has_contact
+            else wall.no_contact["app"] if source == "app"
+            else wall.no_contact["web"])
+        return {"sentence": sentence, "action": "replaced",
+                "reason": "no_contact", "replacement": replacement}
+    if has_contact and boundary_filter.asks_contact(sentence, wall=wall):
+        return {"sentence": sentence, "action": "blocked", "reason": "reask",
+                "replacement": None}
+    return {"sentence": sentence, "action": "pass", "reason": None,
+            "replacement": None}
+
+
+@router.post("/{slug}/wall/test")
+async def test_wall(slug: str, request: Request, user=Depends(_tenant_guard)):
+    """试金石:逐句墙判定预览。body:
+    - sentences: 非空字符串列表(必填);
+    - wall: 覆盖墙(先 compile_wall 校验,坏墙 400 带字段路径)——缺省用
+      当前生效墙(reset/编辑器改墙前的预览入口);
+    - has_contact: 客户已留联系方式(默认 False——启用再索要墙/换话术池);
+    - source: "web" | "app"(默认 web——零号码墙替换句分支)。
+    纯内存判定:不落库、不 bump 版本、不 reload。"""
+    body = await _json_body(request)
+    unknown = set(body) - _WALL_TEST_KEYS
+    if unknown:
+        raise HTTPException(
+            status_code=400,
+            detail=f"未知字段: {', '.join(sorted(unknown))}(合法: "
+                   f"{', '.join(sorted(_WALL_TEST_KEYS))})")
+    sentences = body.get("sentences")
+    if not isinstance(sentences, list) or not sentences:
+        raise HTTPException(status_code=400,
+                            detail="sentences: 须为非空字符串列表")
+    for i, sentence in enumerate(sentences):
+        if not isinstance(sentence, str):
+            raise HTTPException(
+                status_code=400,
+                detail=f"sentences[{i}]: 项须为字符串(实际 "
+                       f"{type(sentence).__name__})")
+    has_contact = body.get("has_contact", False)
+    if not isinstance(has_contact, bool):
+        raise HTTPException(
+            status_code=400,
+            detail=f"has_contact: 须为布尔(实际 {type(has_contact).__name__})")
+    source = body.get("source", "web")
+    if source not in ("web", "app"):
+        raise HTTPException(
+            status_code=400, detail=f"source: 只允许 web/app(实际 {source!r})")
+
+    engine = request.app.state.platform_engine
+    wall_override = body.get("wall")
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        if wall_override is not None:
+            wall_source = "override"
+            candidate = wall_override
+        else:
+            wall_source = "current"
+            _, snapshot = await _load_current(session, tenant.id, slug)
+            candidate = snapshot.get("wall")
+    try:
+        compiled = compile_wall(copy.deepcopy(candidate))
+    except WallConfigError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    results = [_judge_sentence(s, compiled, has_contact=has_contact,
+                               source=source) for s in sentences]
+    return {"slug": slug, "wall_source": wall_source, "results": results}
+
+
+# ---- 平台设置(仅超管;resolver 回退链接入归 Phase 3) -----------------------------
+
+
+def _decode_setting(raw: str):
+    """Text 列还原类型:json.loads 失败(手工 SQL 塞的裸文本)回退原串。"""
+    try:
+        return json.loads(raw)
+    except ValueError:
+        return raw
+
+
+@settings_router.get("")
+async def list_settings(request: Request, user=Depends(require_super)):
+    """平台级默认(集团 AI 凭据/默认 SMTP 等,spec §5.2)——本阶段只落
+    存储与接口;消费(resolver 平台默认回退链)归 Phase 3。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        rows = (await session.execute(
+            select(PlatformSetting).order_by(PlatformSetting.key))
+        ).scalars().all()
+    return {"items": [{"key": row.key, "value": _decode_setting(row.value)}
+                      for row in rows]}
+
+
+@settings_router.put("")
+async def put_setting(request: Request, user=Depends(require_super)):
+    """upsert 单个设置项:key 1-64 字符;value 任意 JSON(json.dumps 入
+    Text,读回还原)。同 key 再写 = 整值替换。"""
+    body = await _json_body(request)
+    key = body.get("key")
+    if not isinstance(key, str) or not 1 <= len(key) <= 64:
+        raise HTTPException(
+            status_code=400, detail="key: 须为 1-64 字符字符串")
+    if "value" not in body:
+        raise HTTPException(status_code=400, detail="value: 缺少(必填)")
+    encoded = json.dumps(body["value"], ensure_ascii=False)
+
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        row = (await session.execute(
+            select(PlatformSetting).where(PlatformSetting.key == key)
+        )).scalar_one_or_none()
+        if row is None:
+            session.add(PlatformSetting(key=key, value=encoded))
+        else:
+            row.value = encoded
+        await session.commit()
+    return {"key": key, "value": body["value"]}

app/platform/routes/tenant_routes.py

@@ -23,10 +23,12 @@ DEFAULT_CONFIG 满足不了。
 
 M-2a 陷阱:Site.brand_overlay 是 ORM JSON 列——写时**整字典赋值*,不做
 按键局部更新(局部改不标记脏,静默丢写)。
+
+Task 6 增量:出厂档案快照助手(_profile_snapshot/_blankify)迁至
+tenant_source.py(配置 API 的重置出厂同样消费);版本 1 写入改走
+db.write_config_version(与配置 API 的 PUT/rollback/reset 同一版本写入
+路径,单一事实源);Tenant 落 profile 列(重置出厂按它取对应档案组)。
 """
-import copy
-import json
-import pathlib
 import re
 
 from fastapi import APIRouter, Depends, HTTPException, Request
@@ -42,50 +44,23 @@ from app.platform.auth import (
     require_super,
     require_tenant,
 )
-from app.platform.db import AdminUser, Site, SiteDomain, Tenant, TenantConfigVersion
-from app.platform.defaults import DEFAULT_CONFIG
+from app.platform.db import (
+    AdminUser,
+    Site,
+    SiteDomain,
+    Tenant,
+    TenantConfigVersion,
+    write_config_version,
+)
 from app.platform.registry import _normalize_host
-from app.platform.tenant_source import FileTenantSource
+from app.platform.tenant_source import _profile_snapshot
 
 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
-
 
 # ---- 校验助手 -------------------------------------------------------------------
 
@@ -253,7 +228,8 @@ async def create_tenant(
         for domain in domains:
             await _assert_domain_free(session, domain)
 
-        tenant = Tenant(slug=body.slug, name=body.name, status="active")
+        tenant = Tenant(slug=body.slug, name=body.name, status="active",
+                        profile=body.profile)
         session.add(tenant)
         await session.flush()
         site = Site(tenant_id=tenant.id, slug="main", name="main")
@@ -268,12 +244,13 @@ async def create_tenant(
             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),
+        # 版本写入走公共 helper(Task 6:与配置 API 的 PUT/rollback/reset
+        # 同一条 MAX(version)+1 路径;新租户无历史版本 → 落版本 1)
+        version = await write_config_version(
+            session, tenant.id, snapshot,
             updated_by=f"super#{user.user_id}",
-            comment=f"创建向导: profile={body.profile}"))
+            comment=f"创建向导: profile={body.profile}")
+        assert version == 1
         try:
             await session.commit()
         except IntegrityError as exc:

app/platform/tenant_source.py

@@ -23,6 +23,7 @@ registry 只认本模块的 TenantSource 协议(async load_all / load_tenant),
 - 文件源数据坏:坏 JSON/slug 与目录不一致/零站点 → ValueError 带路径,
   wall/prompts 文件缺失 → FileNotFoundError 带路径(read_text 原样)。
 """
+import copy
 import json
 import pathlib
 from dataclasses import dataclass
@@ -33,6 +34,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 
 from app.engine.services.ai_service import PROMPTS_KEYS
 from app.platform.db import Site, SiteDomain, Tenant, TenantConfigVersion
+from app.platform.defaults import DEFAULT_CONFIG
 
 
 @dataclass(frozen=True)
@@ -224,3 +226,45 @@ class DbTenantSource:
             sites=sites,
             snapshot=snapshot,
             seed_dir=None)
+
+
+# ---- 出厂档案快照(Task 5 创建向导/Task 6 重置出厂的公共路径) ----------------------
+# 自 tenant_routes 迁入(Task 6 微审注记:配置 API 的 reset 也要用,挪公共位置
+# 供两个路由模块消费——此处已是文件源的地盘,零新增依赖)。
+
+# 出厂档案目录:与 app.main.SEED_DIR 同一锚定规则(仓库根/seed_profiles),
+# 但不 import app.main——main 模块级挂路由,反向导入制造循环依赖。
+SEED_DIR = pathlib.Path(__file__).resolve().parents[2] / "seed_profiles"
+
+# 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 保留——见 Task 5 模块 docstring)。
+
+    重置出厂(spec §5.6"代码库当前模板,随平台升级演进")与创建向导共用
+    本函数:每次调用现读 seed 目录,不缓存。未知 profile → LookupError
+    (read_tenant 的文件源契约,调用方映射 400)。"""
+    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

tests/conftest.py

@@ -143,6 +143,52 @@ AUTO_HOST = "chat.agent.hnxinyouhe.com"
 # Task 4 收拢到 conftest 单一定义)
 NOPE_HOST = "nope.example.com"
 
+# ---- Phase 2 Task 6: DB 源 app/超管 client 夹具 -----------------------------------
+# 自 test_tenant_admin_api 收拢到 conftest(Task 5 时代该文件私有,Task 6 的
+# 配置 API/试金石测试与租户管理测试共用同一套 DB 源闭环环境)。
+# 跨循环纪律见 db_source_app docstring 与 test_tenant_admin_api 模块头。
+
+
+@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)。
+    """
+    from app.platform.tenant_source import DbTenantSource
+
+    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("/api/platform/auth/login",
+                   json={"username": "admin", "password": "admin123"})
+        assert r.status_code == 200
+        c.platform_token = r.json()["token"]
+        r = c.post("/api/platform/auth/change-password", headers=_bearer_of(c),
+                   json={"old_password": "admin123",
+                         "new_password": "sup3r-pass-1"})
+        assert r.status_code == 200
+        yield c
+
+
+def _bearer_of(client):
+    return {"Authorization": f"Bearer {client.platform_token}"}
+
 
 def build_app(seed_dir, data_dir):
     """app 工厂(Task 7): 建 Registry → 预置 app.state.registry → 返回 app。

tests/test_config_api.py

@@ -0,0 +1,431 @@
+"""配置 API(Phase 2 Task 6):分组校验管线/整份版本/回滚/重置出厂/平台设置。
+
+夹具 db_source_app/super_client 自 conftest(DB 源闭环);创建租户等助手
+复用 test_tenant_admin_api(_create/_auth/_tenant_admin_token)。
+
+鉴权矩阵(spec §3):config GET/PUT/reset/versions/rollback/wall-test =
+require_tenant(slug)(租户全开);settings = require_super。
+"""
+from app.platform.tenant_source import FileTenantSource
+from tests.conftest import SEED_DIR
+from tests.test_tenant_admin_api import (
+    _auth,
+    _bearer,
+    _create,
+    _tenant_admin_token,
+)
+
+CONFIG = "/api/platform/tenants/{slug}/config"
+SETTINGS = "/api/platform/settings"
+LOGIN = "/api/platform/auth/login"
+
+
+def _cfg(slug):
+    return CONFIG.format(slug=slug)
+
+
+def _get_config(client, slug):
+    r = client.get(_cfg(slug), headers=_auth(client))
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def _put_group(client, slug, group, payload, comment=None):
+    url = f"{_cfg(slug)}/{group}"
+    if comment is not None:
+        url += f"?comment={comment}"
+    return client.put(url, headers=_auth(client), json=payload)
+
+
+def _jc_snapshot():
+    return FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot
+
+
+# ---- 核心 7 项 ------------------------------------------------------------------
+
+
+def test_put_group_bumps_version_and_hot_reloads(super_client, db_source_app):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    # 旧值可见(顺带在 client 循环里懒建 engine——之后断言 reload dispose 它)
+    r = super_client.get("/api/public/brand",
+                         headers={"host": "demo.example.com"})
+    assert r.status_code == 200
+    assert r.json()["color"] == "#066fc0"            # jc 档案出厂色
+    registry = db_source_app.state.registry
+    rt_old = registry._by_slug["demo"]
+    assert rt_old.engine is not None
+
+    r = _put_group(super_client, "demo", "brand", {"color": "#ff0000"})
+    assert r.status_code == 200, r.text
+    assert r.json()["version"] == 2
+
+    # 热生效:旧 runtime 已被 reload dispose,新 runtime 服务新值
+    assert rt_old.engine is None
+    r = super_client.get("/api/public/brand",
+                         headers={"host": "demo.example.com"})
+    assert r.status_code == 200
+    assert r.json()["color"] == "#ff0000"
+    # 整组替换语义:快照存用户配置原样组,缺省键由消费侧深合并出厂默认
+    assert r.json()["title"] == "AI Support"
+
+    # GET config:新值 + version(原样快照,不是合并视图)
+    body = _get_config(super_client, "demo")
+    assert body["version"] == 2
+    assert body["config"]["brand"] == {"color": "#ff0000"}
+
+
+def test_put_wall_bad_regex_400_with_field(super_client):
+    _create(super_client, slug="demo")
+    wall = _get_config(super_client, "demo")["config"]["wall"]
+    wall["price_patterns"] = ["([unclosed"]
+    r = _put_group(super_client, "demo", "wall", wall)
+    assert r.status_code == 400
+    detail = r.json()["detail"]
+    assert "price_patterns[0]" in detail          # 字段路径
+    assert "regex" in detail                      # regex 字样
+    # 校验失败版本不涨
+    assert _get_config(super_client, "demo")["version"] == 1
+
+
+def test_prompts_size_cap(super_client):
+    _create(super_client, slug="demo")
+    prompts = dict(_get_config(super_client, "demo")["config"]["prompts"])
+    # company_facts 50KB 软上限(spec §5.7):超一节 → 400 带字段路径与上限
+    fat = dict(prompts, company_facts="x" * (50 * 1024 + 1))
+    r = _put_group(super_client, "demo", "prompts", fat)
+    assert r.status_code == 400
+    assert "company_facts" in r.json()["detail"]
+    assert "50" in r.json()["detail"]
+    # 其余三件 100KB 上限
+    fat = dict(prompts, system="x" * (100 * 1024 + 1))
+    r = _put_group(super_client, "demo", "prompts", fat)
+    assert r.status_code == 400
+    assert "system" in r.json()["detail"]
+    # 恰在上限内 → 通过(company_facts 正好 50KB)
+    ok = dict(prompts, company_facts="x" * (50 * 1024))
+    r = _put_group(super_client, "demo", "prompts", ok)
+    assert r.status_code == 200
+    assert r.json()["version"] == 2
+    # prompts 传字符串(Task 4 守卫兜过的形态)→ 400,不是 500/422
+    r = super_client.put(f"{_cfg('demo')}/prompts", headers=_auth(super_client),
+                         json="just a string")
+    assert r.status_code == 400
+    # 缺键 → 400
+    r = _put_group(super_client, "demo", "prompts", {"system": "x"})
+    assert r.status_code == 400
+    assert "company_facts" in r.json()["detail"]
+
+
+def test_prefix_uniqueness_across_tenants(super_client):
+    _create(super_client, slug="jc1", profile="jiancai")   # prefix ai-kefu-jc
+    _create(super_client, slug="au1", profile="auto")      # prefix ai-kefu
+    _create(super_client, slug="bl1", profile="blank")     # prefix ""
+    # auto 租户改成 jc1 的前缀 → 400(spec E1 跨租户查重)
+    r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu-jc"})
+    assert r.status_code == 400
+    assert "prefix" in r.json()["detail"]
+    assert "ai-kefu-jc" in r.json()["detail"]
+    # 新建的 blank 租户(空 prefix)同样撞车 → 400
+    r = _put_group(super_client, "bl1", "push", {"prefix": "ai-kefu-jc"})
+    assert r.status_code == 400
+    # 自身当前前缀重写 → 200(不与自己撞);版本照涨
+    r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu"})
+    assert r.status_code == 200
+    # 换新前缀后,别人再撞新前缀也 400(查重集合随 DB 当前版本刷新)
+    r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu-new"})
+    assert r.status_code == 200
+    r = _put_group(super_client, "bl1", "push", {"prefix": "ai-kefu-new"})
+    assert r.status_code == 400
+    # 空串(未设置)不参战:多个租户可同为 ""
+    r = _put_group(super_client, "bl1", "push", {"prefix": ""})
+    assert r.status_code == 200
+
+
+def test_rollback_restores(super_client):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    assert _put_group(super_client, "demo", "brand",
+                      {"color": "#ff0000"}).status_code == 200    # v2
+
+    # 版本列表:元数据齐全,不含 config 本体
+    r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client))
+    assert r.status_code == 200
+    items = r.json()["items"]
+    assert [i["version"] for i in items] == [1, 2]
+    assert "config_json" not in r.text and '"config"' not in r.text
+    assert items[0]["updated_by"].startswith("super#")
+    assert items[0]["comment"] == "创建向导: profile=jiancai"
+
+    # 回滚到 v1 → 生效值回 v1,产生新版本 v3(历史不可变)
+    r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client),
+                          json={"version": 1})
+    assert r.status_code == 200
+    assert r.json()["version"] == 3
+    body = _get_config(super_client, "demo")
+    assert body["version"] == 3
+    assert body["config"]["brand"]["color"] == "#066fc0"          # v1 的值
+    # 公开门面也回到 v1 色(reload 已生效)
+    r = super_client.get("/api/public/brand",
+                         headers={"host": "demo.example.com"})
+    assert r.json()["color"] == "#066fc0"
+
+    # v2 历史仍在(整份可查),回滚目标不存在 → 404
+    r = super_client.get(f"{_cfg('demo')}/versions/2", headers=_auth(super_client))
+    assert r.status_code == 200
+    assert r.json()["config"]["brand"]["color"] == "#ff0000"
+    r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client),
+                          json={"version": 99})
+    assert r.status_code == 404
+
+
+def test_tenant_admin_scoped_config(super_client):
+    _create(super_client, slug="jc2", admin="jc2-admin",
+            password="jc2-pass-1")
+    _create(super_client, slug="auto2", profile="auto")
+    token = _tenant_admin_token(super_client, "jc2-admin", "jc2-pass-1")
+
+    # 本租户全开:GET/PUT/versions/rollback/reset/wall-test
+    assert super_client.get(
+        _cfg("jc2"), headers=_bearer(token)).status_code == 200
+    r = super_client.put(f"{_cfg('jc2')}/brand", headers=_bearer(token),
+                         json={"color": "#123456"})
+    assert r.status_code == 200
+    assert r.json()["version"] == 2
+    assert super_client.get(
+        f"{_cfg('jc2')}/versions", headers=_bearer(token)).status_code == 200
+    r = super_client.post(f"{_cfg('jc2')}/rollback", headers=_bearer(token),
+                          json={"version": 1})
+    assert r.status_code == 200
+    r = super_client.post(f"{_cfg('jc2')}/brand/reset", headers=_bearer(token))
+    assert r.status_code == 200
+    r = super_client.post("/api/platform/tenants/jc2/wall/test",
+                          headers=_bearer(token), json={"sentences": ["ok."]})
+    assert r.status_code == 200
+
+    # 改别的租户 → 403
+    assert super_client.put(
+        f"{_cfg('auto2')}/brand", headers=_bearer(token),
+        json={"color": "#000000"}).status_code == 403
+    assert super_client.get(
+        _cfg("auto2"), headers=_bearer(token)).status_code == 403
+    assert super_client.post(
+        "/api/platform/tenants/auto2/wall/test", headers=_bearer(token),
+        json={"sentences": ["ok."]}).status_code == 403
+    # 平台设置超管专属
+    assert super_client.get(
+        SETTINGS, headers=_bearer(token)).status_code == 403
+    assert super_client.put(
+        SETTINGS, headers=_bearer(token),
+        json={"key": "k", "value": 1}).status_code == 403
+    # 未认证 401
+    assert super_client.get(_cfg("jc2")).status_code == 401
+
+
+# ---- 校验管线:类型/未知键/范围 ---------------------------------------------------
+
+
+def test_validation_type_errors(super_client):
+    _create(super_client, slug="demo")
+    cases = [
+        # (group, payload, detail 必含片段)
+        ("nope", {"x": 1}, "未知配置组"),
+        ("brand", {"unknown_key": 1}, "未知键"),
+        ("brand", {"color": 123}, "brand.color"),
+        ("brand", {"topics": ["a", 2]}, "brand.topics"),
+        ("contact", {"cc_completion": "yes"}, "cc_completion"),
+        ("contact", {"template_languages": "en"}, "template_languages"),
+        ("email", {"notify_emails": ["no-at-sign"]}, "notify_emails"),
+        ("email", {"smtp": "host"}, "smtp"),
+        ("push", {"timeout": 0}, "timeout"),
+        ("push", {"max_attempts": 11}, "max_attempts"),
+        ("push", {"retry_backoff_base": -1}, "retry_backoff_base"),
+        ("push", {"enabled": 1}, "enabled"),
+        ("ai", {"model": 42}, "model"),
+        ("engine", {"rate_limit_per_hour": "30"}, "rate_limit_per_hour"),
+        # bool 不是 int(isinstance 陷阱)
+        ("engine", {"rate_limit_per_hour": True}, "rate_limit_per_hour"),
+        ("engine", {"ws_conn_limit": 0}, "ws_conn_limit"),
+        ("engine", {"context_seed_keys": {"k": 1}}, "context_seed_keys.k"),
+        ("misc", {"name_to_region_extras": "x"}, "name_to_region_extras"),
+    ]
+    for group, payload, needle in cases:
+        r = _put_group(super_client, "demo", group, payload)
+        assert r.status_code == 400, f"{group}/{payload}: {r.text}"
+        assert needle in r.json()["detail"], f"{group}/{payload}: {r.text}"
+    # 全军覆没后版本仍是 1
+    assert _get_config(super_client, "demo")["version"] == 1
+
+
+def test_valid_groups_pass(super_client):
+    _create(super_client, slug="demo")
+    ok_cases = [
+        ("contact", {"cc_completion": False, "email_light_ask": True,
+                     "template_languages": ["en", "zh"]}),
+        ("email", {"sender_name": "Demo", "notify_emails": ["a@b.com"],
+                   "smtp": {"host": "smtp.example.com", "port": 465}}),
+        ("email", {"smtp": None}),
+        ("push", {"timeout": 5.5, "max_attempts": 3,
+                  "retry_backoff_base": 2}),
+        ("push", {"timeout": None, "max_attempts": None}),
+        ("ai", {"api_key": None, "base_url": None, "model": "glm-4-flash"}),
+        ("engine", {"rate_limit_per_hour": 60, "ws_conn_limit": 2,
+                    "context_seed_keys": {"k": "v"}}),
+        ("misc", {"name_to_region_extras": {"US": "加州"}}),
+    ]
+    version = 1
+    for group, payload in ok_cases:
+        r = _put_group(super_client, "demo", group, payload)
+        assert r.status_code == 200, f"{group}: {r.text}"
+        version += 1
+        assert r.json()["version"] == version
+        assert r.json()["config"] == payload
+
+
+def test_put_unknown_tenant_and_misc(super_client):
+    _create(super_client, slug="demo")
+    assert super_client.get(
+        _cfg("ghost"), headers=_auth(super_client)).status_code == 404
+    r = super_client.put(f"{_cfg('ghost')}/brand", headers=_auth(super_client),
+                         json={"color": "#000"})
+    assert r.status_code == 404
+    # rollback 坏 body → 400(非 422/500)
+    r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client),
+                          json={"version": "one"})
+    assert r.status_code == 400
+    r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client),
+                          json={})
+    assert r.status_code == 400
+    # PUT comment 落版本元数据
+    _put_group(super_client, "demo", "brand", {"color": "#000000"},
+               comment="换主题色")
+    items = _get_config(super_client, "demo")
+    r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client))
+    assert r.json()["items"][-1]["comment"] == "换主题色"
+
+
+# ---- 重置出厂 --------------------------------------------------------------------
+
+
+def test_reset_group_to_factory(super_client):
+    _create(super_client, slug="demo", domains=["demo.example.com"])
+    _put_group(super_client, "demo", "brand", {"color": "#ff0000"})   # v2
+    r = super_client.post(f"{_cfg('demo')}/brand/reset",
+                          headers=_auth(super_client))
+    assert r.status_code == 200
+    assert r.json()["version"] == 3
+    # 回到 jc 档案的整组 brand(重置出厂 = 代码库当前模板,spec §5.6)
+    assert r.json()["config"] == _jc_snapshot()["brand"]
+    body = _get_config(super_client, "demo")
+    assert body["config"]["brand"]["color"] == "#066fc0"
+    r = super_client.get("/api/public/brand",
+                         headers={"host": "demo.example.com"})
+    assert r.json()["color"] == "#066fc0"
+    assert r.json()["title"] == "HESUPIPE Support"
+    # 未知组 reset → 400
+    r = super_client.post(f"{_cfg('demo')}/nope/reset",
+                          headers=_auth(super_client))
+    assert r.status_code == 400
+
+
+def test_reset_blank_tenant_groups(super_client):
+    """blank 租户:wall/prompts 重置取 jiancai 档案;档案没有的组取出厂默认。"""
+    _create(super_client, slug="bl1", profile="blank")
+    _put_group(super_client, "bl1", "brand", {"color": "#ff0000"})
+    # brand → blankified 档案(身份键出厂默认,其余 jc)
+    r = super_client.post(f"{_cfg('bl1')}/brand/reset",
+                          headers=_auth(super_client))
+    assert r.status_code == 200
+    brand = r.json()["config"]
+    assert brand["title"] == "AI Support"       # 身份键 = DEFAULT
+    assert brand["whatsapp"] == ""
+    assert brand["color"] == "#066fc0"          # 非身份键 = jc 档案
+    # wall → jc 档案
+    r = super_client.post(f"{_cfg('bl1')}/wall/reset", headers=_auth(super_client))
+    assert r.status_code == 200
+    assert r.json()["config"] == _jc_snapshot()["wall"]
+    # 档案缺的组 → DEFAULT_CONFIG 整组
+    from app.platform.defaults import DEFAULT_CONFIG
+    r = super_client.post(f"{_cfg('bl1')}/contact/reset",
+                          headers=_auth(super_client))
+    assert r.status_code == 200
+    assert r.json()["config"] == DEFAULT_CONFIG["contact"]
+
+
+# ---- 平台设置(仅超管;本阶段只落存储与接口) ----------------------------------------
+
+
+def test_platform_settings_crud(super_client):
+    r = super_client.get(SETTINGS, headers=_auth(super_client))
+    assert r.status_code == 200
+    assert r.json() == {"items": []}
+    # 写入结构化值(JSON Text 存储,读回还原类型)
+    value = {"host": "smtp.example.com", "port": 465, "tls": True}
+    r = super_client.put(SETTINGS, headers=_auth(super_client),
+                         json={"key": "default_smtp", "value": value})
+    assert r.status_code == 200
+    assert r.json() == {"key": "default_smtp", "value": value}
+    r = super_client.get(SETTINGS, headers=_auth(super_client))
+    assert r.json()["items"] == [{"key": "default_smtp", "value": value}]
+    # 同 key 再写 = 更新(upsert,不重复行)
+    r = super_client.put(SETTINGS, headers=_auth(super_client),
+                         json={"key": "default_smtp", "value": "second"})
+    assert r.status_code == 200
+    r = super_client.get(SETTINGS, headers=_auth(super_client))
+    assert r.json()["items"] == [{"key": "default_smtp", "value": "second"}]
+    # 标量值也可
+    r = super_client.put(SETTINGS, headers=_auth(super_client),
+                         json={"key": "glm_timeout", "value": 30})
+    assert r.status_code == 200
+    # 坏 key/缺 value → 400
+    assert super_client.put(
+        SETTINGS, headers=_auth(super_client),
+        json={"value": 1}).status_code == 400
+    assert super_client.put(
+        SETTINGS, headers=_auth(super_client),
+        json={"key": "", "value": 1}).status_code == 400
+    # 未认证 401
+    assert super_client.get(SETTINGS).status_code == 401
+
+
+# ---- 版本列表形态 -----------------------------------------------------------------
+
+
+def test_versions_listing_shape(super_client, tmp_path):
+    _create(super_client, slug="demo")
+    _put_group(super_client, "demo", "brand", {"color": "#ff0000"})
+    r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client))
+    items = r.json()["items"]
+    assert len(items) == 2
+    for item in items:
+        assert set(item) == {"version", "created_at", "updated_by", "comment"}
+        assert item["created_at"]  # ISO 时间戳
+    # 详情 404
+    r = super_client.get(f"{_cfg('demo')}/versions/9", headers=_auth(super_client))
+    assert r.status_code == 404
+
+
+def test_wall_group_round_trip_and_body_hygiene(super_client):
+    """整墙组原样 PUT 回去必须通过(compile_wall 全量校验的 happy path);
+    坏 JSON 原文/组名撞 versions 字面量一律 400;未改密租户管理员 409 门
+    经 platform_user 连锁生效。"""
+    _create(super_client, slug="demo")
+    wall = _get_config(super_client, "demo")["config"]["wall"]
+    r = _put_group(super_client, "demo", "wall", wall)
+    assert r.status_code == 200
+    assert r.json()["config"] == wall
+
+    # 组名撞 versions 字面量 → 走未知组 400(不与版本路由混淆)
+    r = _put_group(super_client, "demo", "versions", {"a": 1})
+    assert r.status_code == 400
+    assert "未知配置组" in r.json()["detail"]
+    # 坏 JSON 原文 → 400(手动解析,不让 422/500 抢答)
+    r = super_client.put(f"{_cfg('demo')}/brand", headers=_auth(super_client),
+                         content=b"{not json")
+    assert r.status_code == 400
+
+    # 首登未改密的租户管理员 → 409(与登录/改密之外的受保护端点同款门)
+    _create(super_client, slug="t2", admin="t2-admin", password="t2-pass-1")
+    token = super_client.post(
+        LOGIN, json={"username": "t2-admin", "password": "t2-pass-1"}
+    ).json()["token"]
+    r = super_client.get(_cfg("t2"), headers=_bearer(token))
+    assert r.status_code == 409

tests/test_tenant_admin_api.py

@@ -1,7 +1,8 @@
 """租户管理 API(Phase 2 Task 5):超管的创建向导/站点/域名/停用 + 403 边界。
 
-db_source_app 夹具:app.state.registry 用 DbTenantSource(生产形态)指向
-临时 platform.db——创建 API 写库后 registry_reload 走 DB 源闭环(Task 4
+db_source_app/super_client 夹具(Task 6 起收拢到 conftest,与配置 API 测试
+共用):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)。
@@ -20,62 +21,23 @@ 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
+from app.platform.registry import TenantDisabledError
+from app.platform.tenant_source import FileTenantSource, _blankify
+from tests.conftest import SEED_DIR
 
 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
-
-
 # ---- 助手 ----------------------------------------------------------------------
 
 

tests/test_touchstone.py

@@ -0,0 +1,136 @@
+"""试金石(wall/test,Phase 2 Task 6):逐句墙判定预览。
+
+复用 test_wall_execution_tenant 的同款姿势——boundary_filter 纯函数 +
+jc/auto 两墙的词表差异做判别。判定优先级与 conversation.py 执行侧一致:
+价格墙 → 零号码墙 → 再索要墙(仅已留资)。
+
+红线:不落库、不 bump 版本(reset/编辑器共用的预览入口)。
+"""
+from app.engine.services import boundary_filter
+from app.platform.tenant_source import FileTenantSource
+from tests.conftest import SEED_DIR
+from tests.test_tenant_admin_api import _auth, _create
+
+WALL_TEST = "/api/platform/tenants/{slug}/wall/test"
+
+# 判别句选取依据(两墙 diff 实证,勿随手改文案——判别力依赖词表差异):
+# - PRICE 句撞 jc 独有的 [€£]\s?\d 组(auto 词表无此组);
+# - EMAIL 句撞零号码墙 any_email(替换句 = jc no_contact["web"]);
+# - CLEAN 句三墙全放行。
+PRICE = "The total is €500 for one set."
+EMAIL = "Sure, you can also email me at sales@example.com anytime."
+CLEAN = "We ship worldwide every day."
+REASK = "What's your WhatsApp number?"
+
+AUTO_WALL = FileTenantSource(SEED_DIR).read_tenant("auto").snapshot["wall"]
+JC_WALL = FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot["wall"]
+
+
+def _post(client, slug, payload):
+    return client.post(WALL_TEST.format(slug=slug),
+                       headers=_auth(client), json=payload)
+
+
+def _config_version(client, slug):
+    r = client.get(f"/api/platform/tenants/{slug}/config",
+                   headers=_auth(client))
+    assert r.status_code == 200
+    return r.json()["version"]
+
+
+def test_touchstone_verdicts(super_client):
+    _create(super_client, slug="demo")
+    version_before = _config_version(super_client, "demo")
+
+    r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, CLEAN]})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["wall_source"] == "current"
+    results = body["results"]
+    assert len(results) == 3
+
+    # 价格墙: blocked + 替换话术来自 jc deflections 池(en 首句)
+    assert results[0]["action"] == "blocked"
+    assert results[0]["reason"] == "boundary"
+    assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN[0]
+    # 零号码墙: replaced + 替换句精确等于 jc no_contact["web"]
+    assert results[1]["action"] == "replaced"
+    assert results[1]["reason"] == "no_contact"
+    assert results[1]["replacement"] == boundary_filter.NO_CONTACT_WEB
+    # 干净句: pass,无替换
+    assert results[2]["action"] == "pass"
+    assert results[2]["reason"] is None
+    assert results[2]["replacement"] is None
+
+    # 不落库不 bump 版本
+    assert _config_version(super_client, "demo") == version_before
+
+
+def test_touchstone_has_contact_and_source_variants(super_client):
+    """已留资/app 渠道两开关照搬执行侧语义:替换句换池/换变体,再索要墙启用。"""
+    _create(super_client, slug="demo")
+    # has_contact=True: 价格墙替换句来自 deflections_contact 池
+    r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, REASK],
+                                     "has_contact": True})
+    results = r.json()["results"]
+    assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN_CONTACT[0]
+    # 零号码墙替换句换 already 变体(已留资不再索要)
+    assert results[1]["replacement"] == boundary_filter.NO_CONTACT_ALREADY
+    # 再索要句被丢(has_contact 才启用): blocked 无替换
+    assert results[2]["action"] == "blocked"
+    assert results[2]["reason"] == "reask"
+    assert results[2]["replacement"] is None
+
+    # source=app: 零号码墙替换句用 app 变体
+    r = _post(super_client, "demo", {"sentences": [EMAIL], "source": "app"})
+    assert r.json()["results"][0]["replacement"] == boundary_filter.NO_CONTACT_APP
+    # has_contact=False(默认): 再索要句放行(墙未启用)
+    r = _post(super_client, "demo", {"sentences": [REASK]})
+    assert r.json()["results"][0]["action"] == "pass"
+
+
+def test_touchstone_wall_override(super_client):
+    """body 带 wall 覆盖 = reset/编辑器的预览:同句换墙判定翻转,不动库。"""
+    _create(super_client, slug="demo")
+    version_before = _config_version(super_client, "demo")
+
+    # 覆盖为 auto 墙(无 € 组): 价格句放行
+    r = _post(super_client, "demo",
+              {"sentences": [PRICE], "wall": AUTO_WALL})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["wall_source"] == "override"
+    assert body["results"][0]["action"] == "pass"
+
+    # 覆盖墙坏 regex → 400 带字段路径与 regex 字样
+    bad = dict(JC_WALL)
+    bad["price_patterns"] = ["([unclosed"]
+    r = _post(super_client, "demo", {"sentences": [PRICE], "wall": bad})
+    assert r.status_code == 400
+    assert "price_patterns[0]" in r.json()["detail"]
+    assert "regex" in r.json()["detail"]
+
+    # 覆盖实验全程不 bump 版本
+    assert _config_version(super_client, "demo") == version_before
+
+
+def test_touchstone_input_validation(super_client):
+    _create(super_client, slug="demo")
+    # sentences 必须是非空字符串列表
+    assert _post(super_client, "demo", {"sentences": []}).status_code == 400
+    assert _post(super_client, "demo",
+                 {"sentences": "not a list"}).status_code == 400
+    assert _post(super_client, "demo",
+                 {"sentences": ["ok", 42]}).status_code == 400
+    assert _post(super_client, "demo", {}).status_code == 400
+    # source 只认 web/app;has_contact 必须 bool;未知键拒绝
+    assert _post(super_client, "demo",
+                 {"sentences": ["ok."], "source": "sms"}).status_code == 400
+    assert _post(super_client, "demo",
+                 {"sentences": ["ok."],
+                  "has_contact": "yes"}).status_code == 400
+    assert _post(super_client, "demo",
+                 {"sentences": ["ok."], "surprise": 1}).status_code == 400
+    # 未知租户 404
+    assert _post(super_client, "ghost",
+                 {"sentences": ["ok."]}).status_code == 404