feat: TenantSource 抽象——File源(测试/档案)与DB源(生产)，registry 数据源可切换+reload upsert

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

app/engine/routes/public.py

@@ -81,7 +81,10 @@ async def avatar(request: Request):
     # 任意文件。显式拒绝分隔符与目录点,命中即回退静态兜底。
     if name in {".", ".."} or "/" in name or "\\" in name:
         name = ""
-    candidate = runtime.seed_dir / "assets" / name if name else None
+    # seed_dir=None = DB 源租户(Task 4;资产目录尚无 DB 侧存储)——直接
+    # 落静态兜底,不让 None 进路径拼接炸 500
+    candidate = (runtime.seed_dir / "assets" / name
+                 if name and runtime.seed_dir is not None else None)
     if candidate is not None and candidate.is_file():
         return FileResponse(
             candidate, headers={"Access-Control-Allow-Origin": "*"})

app/engine/services/ai_service.py

@@ -47,6 +47,11 @@ def _format_deflection_pool(deflections) -> str:
     )
 
 
+# 四件套键名(单一事实源): PromptProvider 构造校验 / FileTenantSource 读盘
+# / DbTenantSource 快照组形态, 三处消费同一清单
+PROMPTS_KEYS = ("system", "company_facts", "extractor", "evaluator")
+
+
 @dataclass(frozen=True)
 class PromptProvider:
     """租户提示词四件套持有者(Task 6, R1 话术单一事实源)。
@@ -54,9 +59,12 @@ class PromptProvider:
     frozen(Task 9 收口, Task 6 审查遗留): 实例跨租户运行时/模块级
     DEFAULT_PROVIDER 共享, 不可变防共享单例被毒化。
 
-    - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md,
-      存实例字段——提示词属于租户(seed_profiles/<slug>/prompts/),不再有
-      代码级 app/engine/prompts 副本;
+    - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md
+      委托 from_config——提示词属于租户(seed_profiles/<slug>/prompts/),
+      不再有代码级 app/engine/prompts 副本;
+    - from_config(md: dict)(Phase 2 Task 4): 快照 prompts 组(dict 四键)
+      直接构造——registry 对文件/DB 两源的统一入口(文件源读盘入快照,
+      DB 源嵌 config_json);缺键/值非字符串 → 显性 ValueError 带键名;
     - build_system(wall_deflections): system.md 的 $DEFLECTION_POOL 占位符
       替换为按语言枚举的话术清单 + "\\n\\n" + company_facts(行为 prompt
       在前、事实卡居中、当轮提醒最后拼接——拼接位在 AIService 内);
@@ -68,15 +76,28 @@ class PromptProvider:
     extractor: str
     evaluator: str
 
+    @classmethod
+    def from_config(cls, md: dict) -> "PromptProvider":
+        """快照 prompts 组 → PromptProvider。缺键/非字符串显性炸带键名
+        (DB 快照坏数据不留到 build_system 才炸)。"""
+        bad = [k for k in PROMPTS_KEYS if not isinstance((md or {}).get(k), str)]
+        if bad:
+            raise ValueError(f"prompts 配置缺少(或非字符串)键: {', '.join(bad)}")
+        return cls(
+            system=md["system"],
+            company_facts=md["company_facts"],
+            extractor=md["extractor"],
+            evaluator=md["evaluator"])
+
     @classmethod
     def from_dir(cls, prompts_dir) -> "PromptProvider":
         d = Path(prompts_dir)
-        return cls(
-            system=(d / "system.md").read_text(encoding="utf-8"),
-            company_facts=(d / "company_facts.md").read_text(encoding="utf-8"),
-            extractor=(d / "extractor.md").read_text(encoding="utf-8"),
-            evaluator=(d / "evaluator.md").read_text(encoding="utf-8"),
-        )
+        return cls.from_config({
+            "system": (d / "system.md").read_text(encoding="utf-8"),
+            "company_facts": (d / "company_facts.md").read_text(encoding="utf-8"),
+            "extractor": (d / "extractor.md").read_text(encoding="utf-8"),
+            "evaluator": (d / "evaluator.md").read_text(encoding="utf-8"),
+        })
 
     def build_system(self, wall_deflections=None) -> str:
         system = self.system

app/platform/registry.py

@@ -1,9 +1,12 @@
-"""TenantRegistry——进程内租户注册表(Phase 1 文件加载器)。
+"""TenantRegistry——进程内租户注册表(Phase 1 文件加载;Task 4 起数据源
+抽象化)。
 
-数据源:seed_profiles/<slug>/tenant.json(站点/品牌/引擎配置)+ 同目录
-wall.json(并入 snapshot["wall"],compile_wall 编译期显性炸)。Phase 2 换
-platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子替换条目,
-进行中的会话下一条消息即用新配置)。
+数据源经 TenantSource 协议注入(app/platform/tenant_source.py):
+FileTenantSource(seed_profiles/,测试/出厂档案)与 DbTenantSource
+(platform.db,生产 Task 11 起装配)输出同构 LoadedTenant(snapshot =
+配置组 ⊕ wall ⊕ prompts),本模块对源差异无感知。load_from_dir 保持
+为文件源便捷构造(既有调用零改动)。spec §5.3 的热加载语义不变:后台
+保存→原子替换条目,进行中的会话下一条消息即用新配置。
 
 关键行为:
 - resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site);
@@ -14,16 +17,17 @@ platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子
   per-slug asyncio.Lock 串行,锁内双检防并发双建;持锁后现势检查——条目
   已被 reload 换掉则放弃本次建库、按当前条目重走(旧 runtime 永不产生
   孤儿 engine);建库失败 dispose 半开 engine,保持未就绪下次重试。
-- registry_reload(slug):磁盘读取/校验在锁外,swap+dispose 段与懒建/
-  dispose_all 取同一把 per-slug 锁。坏配置(无效 JSON/墙编译失败/零站点/
-  域名冲突)在替换前抛,旧条目原样服务;替换后 dispose 旧 engine
-  (Windows 下不关=库文件句柄泄漏)。
+- registry_reload(slug):经 source 重读(锁外),swap+dispose 段与懒建/
+  dispose_all 取同一把 per-slug 锁。upsert 语义(Task 4):slug 未加载但
+  source 已知(DB/文件新建租户)→ 载入插入;source 也不知道 →
+  UnknownTenantError。坏配置(无效 JSON/墙编译失败/零站点/域名冲突)在
+  替换前抛,旧条目原样服务;替换后 dispose 旧 engine(Windows 下不关=
+  库文件句柄泄漏)。
 
 测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all()
 (Windows 文件锁,tmp 目录删不掉)。
 """
 import asyncio
-import json
 import pathlib
 from dataclasses import dataclass
 
@@ -39,6 +43,11 @@ from app.engine.models import ensure_schema
 from app.engine.services.ai_service import PromptProvider
 from app.engine.wall_schema import CompiledWall, compile_wall
 from app.platform.tenant_config import Site, TenantConfig
+from app.platform.tenant_source import (
+    FileTenantSource,
+    LoadedTenant,
+    TenantSource,
+)
 
 
 class UnknownTenantError(LookupError):
@@ -50,19 +59,6 @@ class TenantDisabledError(LookupError):
     Task 7 中间件负责映射)。"""
 
 
-def _read_json(path: pathlib.Path):
-    """读 JSON 文件;读不了/解析不了都带路径抛 ValueError——哪份文件
-    坏了一目了然(裸 FileNotFoundError/JSONDecodeError 不含路径上下文)。"""
-    try:
-        text = path.read_text(encoding="utf-8")
-    except OSError as exc:
-        raise ValueError(f"{path}: 无法读取: {exc}") from exc
-    try:
-        return json.loads(text)
-    except ValueError as exc:                     # JSONDecodeError 等
-        raise ValueError(f"{path}: 无效 JSON: {exc}") from exc
-
-
 def make_tenant_engine(db_path: pathlib.Path) -> AsyncEngine:
     """每租户 SQLite engine:pool=2 无溢出(单租户低并发),忙等 30s
     (对齐 jc 现状:后台推送连接可能撞 SMTP 窗口写锁),打开即 WAL
@@ -93,16 +89,17 @@ class TenantRuntime:
     """单租户运行时。
 
     加载期就绪:config(TenantConfig:data=快照⊕出厂默认,含 sites)/
-    db_path(该租户 SQLite 文件路径)/seed_dir(seed_profiles/<slug>/)/
-    compiled_wall(墙编译产物)/prompts(Task 6 PromptProvider,from_dir
-    (seed_dir/prompts) 四件套——$DEFLECTION_POOL 话术池注入的单一消费口)。
+    db_path(该租户 SQLite 文件路径)/seed_dir(文件源=租户目录;DB 源
+    None——avatar 解析链对其回退静态兜底)/compiled_wall(墙编译产物)/
+    prompts(PromptProvider.from_config(snapshot["prompts"]),两源统一
+    入口——$DEFLECTION_POOL 话术池注入的单一消费口)。
     懒建:engine/session_factory(Registry 首次 resolve,ensure_schema 已跑)。
     manager(Task 5b)由装配线另行构造注入,不在本类持有。
     """
 
     config: TenantConfig
     db_path: pathlib.Path
-    seed_dir: pathlib.Path
+    seed_dir: pathlib.Path | None
     compiled_wall: CompiledWall
     prompts: PromptProvider
     engine: AsyncEngine | None = None
@@ -120,9 +117,9 @@ class TenantRegistry:
     """tenant slug → TenantRuntime;域名 → (tenant_slug, site_slug) 索引。
     reload 重建索引而非增量维护——租户数是个位数,整建最简且无漂移。"""
 
-    def __init__(self):
-        self._seed_dir: pathlib.Path | None = None
-        self._data_dir: pathlib.Path | None = None
+    def __init__(self, source: TenantSource, data_dir):
+        self._source = source
+        self._data_dir = pathlib.Path(data_dir)
         self._by_slug: dict[str, TenantRuntime] = {}
         self._domain_index: dict[str, tuple[str, str]] = {}
         self._locks: dict[str, asyncio.Lock] = {}
@@ -131,52 +128,52 @@ class TenantRegistry:
 
     @classmethod
     def load_from_dir(cls, seed_dir, data_dir) -> "TenantRegistry":
-        """seed_profiles/<slug>/tenant.json(+同目录 wall.json)→ Registry。
+        """文件源便捷构造(既有调用零改动):seed_profiles/<slug>/
+        tenant.json + wall.json + prompts/ → Registry,同步完成。
 
         加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir
         路径拼错→静默空注册表→全量 404,必须炸)、无效 JSON、slug 与目录
         名不一致、零站点、wall 编译错误、域名重复绑定、prompts 读取失败
-        (Task 6: prompts/ 目录缺失或四件套不齐 → PromptProvider.from_dir
-        的 FileNotFoundError,加载即炸,不流入运行期)。
+        (prompts/ 目录缺失或四件套不齐 → read_text 的 FileNotFoundError,
+        加载即炸,不流入运行期)。
         """
-        reg = cls()
-        reg._seed_dir = pathlib.Path(seed_dir)
-        reg._data_dir = pathlib.Path(data_dir)
-        tenant_jsons = sorted(reg._seed_dir.glob("*/tenant.json"))
-        if not tenant_jsons:
-            raise ValueError(f"{reg._seed_dir} 下没有任何 <slug>/tenant.json")
-        for tenant_json in tenant_jsons:
-            rt = reg._load_tenant(tenant_json.parent)
-            if rt.config.slug in reg._by_slug:
-                raise ValueError(f"租户 slug 重复: {rt.config.slug}")
-            reg._by_slug[rt.config.slug] = rt
-        reg._domain_index = reg._build_domain_index(reg._by_slug)
+        source = FileTenantSource(seed_dir)
+        reg = cls(source, data_dir)
+        reg._install(source.read_all())
         return reg
 
-    def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime:
-        slug = tenant_dir.name
-        tenant_json = _read_json(tenant_dir / "tenant.json")
-        if tenant_json.get("slug") != slug:
-            raise ValueError(
-                f"{tenant_dir / 'tenant.json'}: slug 字段 "
-                f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致")
-        sites = tenant_json.get("sites") or []
-        if not sites:
-            raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)")
-        # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/
-        # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译)
-        wall_cfg = _read_json(tenant_dir / "wall.json")
-        snapshot = dict(tenant_json.get("config") or {})
-        snapshot["wall"] = wall_cfg
+    async def load_all(self) -> None:
+        """初始异步加载(DB 源等 async 源的装配入口;文件源走
+        load_from_dir 同步路径,不经过这里)。空源由 source 自己显性炸。"""
+        self._install(await self._source.load_all())
+
+    def _install(self, loaded: list[LoadedTenant]) -> None:
+        """LoadedTenant 列表 → 全量替换 _by_slug + 域名索引(整体重建,
+        无半更新状态;冲突在赋值前抛,旧状态原样)。"""
+        if not loaded:
+            raise ValueError(f"数据源未返回任何租户: {self._source!r}")
+        by_slug: dict[str, TenantRuntime] = {}
+        for lt in loaded:
+            rt = self._build_runtime(lt)
+            if rt.config.slug in by_slug:
+                raise ValueError(f"租户 slug 重复: {rt.config.slug}")
+            by_slug[rt.config.slug] = rt
+        self._by_slug = by_slug
+        self._domain_index = self._build_domain_index(by_slug)
+
+    def _build_runtime(self, lt: LoadedTenant) -> TenantRuntime:
+        """LoadedTenant → TenantRuntime(两源同构快照的单一消费点:
+        from_snapshot 出厂默认合并 / compile_wall 墙编译 /
+        PromptProvider.from_config 四件套)。"""
         config = TenantConfig.from_snapshot(
-            slug, tenant_json.get("name", slug), snapshot,
-            sites=sites, status=tenant_json.get("status", "active"))
+            lt.slug, lt.name, lt.snapshot,
+            sites=lt.sites, status=lt.status)
         return TenantRuntime(
             config=config,
-            db_path=self._data_dir / f"{slug}.db",
-            seed_dir=tenant_dir,
-            compiled_wall=compile_wall(wall_cfg),
-            prompts=PromptProvider.from_dir(tenant_dir / "prompts"))
+            db_path=self._data_dir / f"{lt.slug}.db",
+            seed_dir=lt.seed_dir,
+            compiled_wall=compile_wall(lt.snapshot.get("wall")),
+            prompts=PromptProvider.from_config(lt.snapshot.get("prompts")))
 
     def _build_domain_index(
             self, by_slug: dict[str, TenantRuntime]
@@ -272,23 +269,31 @@ class TenantRegistry:
             return rt
 
     async def registry_reload(self, slug: str) -> None:
-        """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。
+        """经 source 重读单租户,原子替换条目;旧 runtime 若已建 engine,
+        dispose 掉。
 
-        磁盘读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all
+        upsert 语义(Task 4,评审修订):slug 不在 _by_slug 但 source 已知
+        (新建租户已落库/落盘)→ 载入插入——插入同样走 per-slug 锁 + 整建
+        域名索引,与替换完全同路;source 也不知道(slug 拼错/未保存)→
+        UnknownTenantError。
+
+        source 读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all
         取同一把 per-slug 锁:懒建要么完整发生在替换前(dispose 得到旧
         engine),要么持锁后发现条目已换而放弃——不存在"建到一半被换"
         的孤儿窗口。坏配置在替换前抛,旧条目原样服务,不留半更新状态。
         """
-        if slug not in self._by_slug:
-            raise UnknownTenantError(f"租户不存在: {slug!r}")
-        rt = self._load_tenant(self._seed_dir / slug)
+        try:
+            loaded = await self._source.load_tenant(slug)
+        except LookupError as exc:
+            raise UnknownTenantError(f"租户不存在: {slug!r}") from exc
+        rt = self._build_runtime(loaded)
         lock = self._locks.setdefault(slug, asyncio.Lock())
         async with lock:
             old = self._by_slug.get(slug)
             candidate = dict(self._by_slug)
             candidate[slug] = rt
             new_index = self._build_domain_index(candidate)  # 冲突在此抛,状态未动
-            self._by_slug[slug] = rt                 # 原子替换(锁内,无 await)
+            self._by_slug[slug] = rt                 # 原子替换/插入(锁内,无 await)
             self._domain_index = new_index
             if old is not None and old.engine is not None:
                 await old.engine.dispose()

app/platform/tenant_source.py

@@ -0,0 +1,224 @@
+"""TenantSource——租户数据源抽象(Phase 2 Task 4)。
+
+registry 只认本模块的 TenantSource 协议(async load_all / load_tenant),
+不再直连文件系统:测试/出厂档案走 FileTenantSource,生产(Task 11 起装配)
+走 DbTenantSource(platform.db)。两源输出同构 LoadedTenant:
+
+- sites: raw 站点列表,形态 = 文件源 tenant.json 的 sites 数组
+  ({slug, name, domains, brand})——DB 源把 sites.brand_overlay 列映射进
+  "brand" 键(TenantConfig.from_snapshot 的站点覆盖合并链消费同一形态,
+  未覆盖项继承租户级 brand);
+- snapshot: 租户配置快照 dict = 配置组(brand/push/ai/...) ⊕ wall(墙
+  词表)⊕ prompts(提示词四件套文本)——两源键形态一致,registry 侧
+  单一消费路径(compile_wall / PromptProvider.from_config)。
+
+文件源行为增量(本任务,等价迁移):prompts 四件套并入 snapshot["prompts"]
+——旧路径 PromptProvider.from_dir 直读盘、不进快照;对齐后 DB 源(prompts
+嵌 config_json)与文件源输出形态一致,registry 不再感知源差异。
+
+错误契约(两源一致,registry 依赖):
+- 未知 slug → LookupError(registry_reload 映射为 UnknownTenantError);
+- 数据坏(缺版本/坏 JSON/零站点/缺 wall·prompts 组)→ ValueError 带
+  租户名(加载期显性失败,不流入运行期)。
+"""
+import json
+import pathlib
+from dataclasses import dataclass
+from typing import Protocol
+
+from sqlalchemy import select
+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
+
+
+@dataclass(frozen=True)
+class LoadedTenant:
+    """单租户的源侧加载产物(两源同构)。
+
+    snapshot 含 wall/prompts 键(见模块 docstring);sites 为 raw dict
+    列表(TenantConfig.from_snapshot 直接消费);seed_dir 文件源专属
+    (租户目录,DB 源 None——avatar 解析链对 None 回退静态兜底)。
+    """
+
+    slug: str
+    name: str
+    status: str
+    sites: list[dict]
+    snapshot: dict
+    seed_dir: pathlib.Path | None = None
+
+
+class TenantSource(Protocol):
+    """registry 的数据源接口(纯读,无缓存——每次现读)。"""
+
+    async def load_all(self) -> list[LoadedTenant]:
+        """全量加载(启动/整表重读)。空源 → ValueError(路径拼错静默
+        空注册表 = 全量 404,必须炸)。"""
+        ...
+
+    async def load_tenant(self, slug: str) -> LoadedTenant:
+        """单租户重读(registry_reload 用)。未知 slug → LookupError。"""
+        ...
+
+
+def _read_json(path: pathlib.Path):
+    """读 JSON 文件;读不了/解析不了都带路径抛 ValueError——哪份文件
+    坏了一目了然(裸 FileNotFoundError/JSONDecodeError 不含路径上下文)。
+    (自 registry 原样迁入,报错文案逐字等价。)"""
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        raise ValueError(f"{path}: 无法读取: {exc}") from exc
+    try:
+        return json.loads(text)
+    except ValueError as exc:                     # JSONDecodeError 等
+        raise ValueError(f"{path}: 无效 JSON: {exc}") from exc
+
+
+class FileTenantSource:
+    """文件源:seed_profiles/<slug>/(测试/出厂档案)。
+
+    行为自 registry._load_tenant 逐字迁入(含校验与报错文案);唯一增量
+    = prompts 四件套读入 snapshot["prompts"](两源形态对齐)。读盘是纯
+    同步快操作——同步本体 read_all/read_tenant 供 load_from_dir 便捷构造
+    同步调用;async load_all/load_tenant 为 TenantSource 协议适配
+    (registry 的统一 async 消费面,直接返回本体结果)。
+    """
+
+    def __init__(self, seed_dir):
+        self.seed_dir = pathlib.Path(seed_dir)
+
+    # ---- 同步本体 ---------------------------------------------------------
+
+    def read_all(self) -> list[LoadedTenant]:
+        tenant_jsons = sorted(self.seed_dir.glob("*/tenant.json"))
+        if not tenant_jsons:
+            raise ValueError(f"{self.seed_dir} 下没有任何 <slug>/tenant.json")
+        return [self.read_tenant(p.parent.name) for p in tenant_jsons]
+
+    def read_tenant(self, slug: str) -> LoadedTenant:
+        tenant_json_path = self.seed_dir / slug / "tenant.json"
+        if not tenant_json_path.is_file():
+            raise LookupError(f"文件源没有租户 {slug!r}(缺 {tenant_json_path})")
+        tenant_json = _read_json(tenant_json_path)
+        if tenant_json.get("slug") != slug:
+            raise ValueError(
+                f"{tenant_json_path}: slug 字段 "
+                f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致")
+        sites = tenant_json.get("sites") or []
+        if not sites:
+            raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)")
+        # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/
+        # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译)
+        snapshot = dict(tenant_json.get("config") or {})
+        snapshot["wall"] = _read_json(self.seed_dir / slug / "wall.json")
+        prompts_dir = self.seed_dir / slug / "prompts"
+        snapshot["prompts"] = {
+            name: (prompts_dir / f"{name}.md").read_text(encoding="utf-8")
+            for name in PROMPTS_KEYS
+        }
+        return LoadedTenant(
+            slug=slug,
+            name=tenant_json.get("name", slug),
+            status=tenant_json.get("status", "active"),
+            sites=list(sites),
+            snapshot=snapshot,
+            seed_dir=self.seed_dir / slug)
+
+    # ---- 协议适配(registry 的 async 消费面)--------------------------------
+
+    async def load_all(self) -> list[LoadedTenant]:
+        return self.read_all()
+
+    async def load_tenant(self, slug: str) -> LoadedTenant:
+        return self.read_tenant(slug)
+
+
+class DbTenantSource:
+    """platform.db 源(生产,Task 11 起装配;本任务先交付能力)。
+
+    tenants JOIN sites(brand_overlay → 站点 brand 覆盖层)/site_domains
+    + 当前生效版本(MAX(version),db.py 纪律:历史不可变,当前版取最大)
+    的 config_json 解析。版本快照由 Task 5 创建向导写入(config 组 ⊕ wall
+    ⊕ prompts 同构嵌套)。
+
+    已知边界:site.status 列暂不参与映射(全量输出)——Task 5 站点
+    CRUD/停用接线时定语义。
+    """
+
+    def __init__(self, engine: AsyncEngine):
+        self._engine = engine
+
+    async def load_all(self) -> list[LoadedTenant]:
+        async with AsyncSession(self._engine) as session:
+            tenants = (await session.execute(
+                select(Tenant).order_by(Tenant.slug))).scalars().all()
+            if not tenants:
+                raise ValueError(
+                    "platform.db 没有任何租户(tenants 表为空)——"
+                    "库路径拼错会静默空注册表(全量 404),必须炸")
+            return [await self._load_one(session, t) for t in tenants]
+
+    async def load_tenant(self, slug: str) -> LoadedTenant:
+        async with AsyncSession(self._engine) as session:
+            tenant = (await session.execute(
+                select(Tenant).where(Tenant.slug == slug))).scalar_one_or_none()
+            if tenant is None:
+                raise LookupError(f"platform.db 中不存在租户: {slug!r}")
+            return await self._load_one(session, tenant)
+
+    async def _load_one(
+            self, session: AsyncSession, tenant: Tenant) -> LoadedTenant:
+        sites_rows = (await session.execute(
+            select(Site).where(Site.tenant_id == tenant.id)
+            .order_by(Site.id))).scalars().all()
+        if not sites_rows:
+            raise ValueError(f"{tenant.slug}: 每租户至少一个站点(sites 为空)")
+        domains = (await session.execute(
+            select(SiteDomain)
+            .where(SiteDomain.site_id.in_([s.id for s in sites_rows]))
+            .order_by(SiteDomain.id))).scalars().all()
+        domains_by_site: dict[int, list[str]] = {}
+        for d in domains:
+            domains_by_site.setdefault(d.site_id, []).append(d.domain)
+
+        version = (await session.execute(
+            select(TenantConfigVersion)
+            .where(TenantConfigVersion.tenant_id == tenant.id)
+            .order_by(TenantConfigVersion.version.desc())
+            .limit(1))).scalar_one_or_none()
+        if version is None:
+            raise ValueError(
+                f"租户 {tenant.name!r}({tenant.slug}) 没有任何配置版本"
+                f"(tenant_config_versions 无行)——先在后台保存配置出版本 1")
+        try:
+            snapshot = json.loads(version.config_json)
+        except ValueError as exc:
+            raise ValueError(
+                f"租户 {tenant.name!r}({tenant.slug}) 版本 "
+                f"{version.version} 的 config_json 无效 JSON: {exc}") from exc
+        if not isinstance(snapshot, dict):
+            raise ValueError(
+                f"租户 {tenant.name!r}({tenant.slug}) 版本 "
+                f"{version.version} 的 config_json 不是 JSON 对象"
+                f"(实际 {type(snapshot).__name__})")
+        for key in ("wall", "prompts"):
+            if not isinstance(snapshot.get(key), dict):
+                raise ValueError(
+                    f"租户 {tenant.name!r}({tenant.slug}) 版本 "
+                    f"{version.version} 的 config_json 缺少 {key!r} 配置组"
+                    f"(或其不是对象)")
+
+        sites = [{"slug": s.slug, "name": s.name,
+                  "domains": list(domains_by_site.get(s.id, [])),
+                  "brand": dict(s.brand_overlay or {})}
+                 for s in sites_rows]
+        return LoadedTenant(
+            slug=tenant.slug,
+            name=tenant.name,
+            status=tenant.status,
+            sites=sites,
+            snapshot=snapshot,
+            seed_dir=None)

tests/conftest.py

@@ -139,6 +139,9 @@ def build_auto_runtime():
 # (与 5b 之前 default_runtime 兜底的语义一致,既有断言零改动)
 JIANCAI_HOST = "chat2.agent.hnxinyouhe.com"
 AUTO_HOST = "chat.agent.hnxinyouhe.com"
+# 未知 Host 哨兵(原 test_admin_mounting/test_host_routing 各自定义,
+# Task 4 收拢到 conftest 单一定义)
+NOPE_HOST = "nope.example.com"
 
 
 def build_app(seed_dir, data_dir):

tests/test_admin_mounting.py

@@ -9,7 +9,7 @@
 # 的 test_tenant_middleware_is_outermost_user_middleware 覆盖
 # (user_middleware[0] 断言)——全量回归确认仍绿,此处不重复建。
 
-NOPE_HOST = "nope.example.com"
+from tests.conftest import NOPE_HOST
 
 
 def test_platform_api_bypasses_tenant_resolution(app_client):

tests/test_db_tenant_source.py

@@ -0,0 +1,356 @@
+"""TenantSource 抽象与 DB 源(Phase 2 Task 4)。
+
+两源输出同构 LoadedTenant:sites(tenant.json sites 形态,DB 源把
+sites.brand_overlay 列映射进站点 "brand" 键)+ snapshot(配置组 ⊕ wall
+⊕ prompts 四件套文本——文件源的行为增量:prompts 并入快照,两源对齐)。
+
+registry 构造收 source:load_from_dir 退化为 File 源便捷构造(既有调用
+零改动,既有 test_registry 全绿即证文件源行为等价);registry_reload 走
+source 并支持 upsert——source 已知但未加载的 slug 载入插入(Task 5
+创建租户后 reload 依赖此语义),source 也不知道 → UnknownTenantError。
+
+Windows 纪律:建过 engine 的用例结尾 await engine.dispose()/dispose_all()
+(文件锁,tmp 目录才删得掉)。
+"""
+import json
+import shutil
+from pathlib import Path
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.engine.services.ai_service import PromptProvider
+from app.platform.db import (
+    Site,
+    SiteDomain,
+    Tenant,
+    TenantConfigVersion,
+    ensure_platform_schema,
+    make_platform_engine,
+)
+from app.platform.registry import UnknownTenantError, TenantRegistry
+from app.platform.tenant_source import DbTenantSource, FileTenantSource
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SEED_DIR = REPO_ROOT / "seed_profiles"
+JC_DOMAIN = "db.jc.example.com"      # DB 源测试专属域名(不与文件源种子撞)
+
+_PROMPTS_KEYS = ("system", "company_facts", "extractor", "evaluator")
+
+
+def _jiancai_snapshot(prefix="ai-kefu-jc"):
+    """版本 config_json 构造助手:seed jiancai 的 config 组 + wall.json +
+    prompts 四件套合并(Task 5 创建向导将产出同构内容)。"""
+    tenant = json.loads(
+        (SEED_DIR / "jiancai" / "tenant.json").read_text(encoding="utf-8"))
+    prompts_dir = SEED_DIR / "jiancai" / "prompts"
+    snapshot = dict(tenant["config"])
+    snapshot["push"] = dict(snapshot.get("push") or {}, prefix=prefix)
+    snapshot["wall"] = json.loads(
+        (SEED_DIR / "jiancai" / "wall.json").read_text(encoding="utf-8"))
+    snapshot["prompts"] = {
+        name: (prompts_dir / f"{name}.md").read_text(encoding="utf-8")
+        for name in _PROMPTS_KEYS
+    }
+    return snapshot
+
+
+async def _make_engine(tmp_path):
+    engine = make_platform_engine(tmp_path / "platform.db")
+    await ensure_platform_schema(engine)
+    return engine
+
+
+async def _add_jiancai(engine, prefix="ai-kefu-jc", version=1,
+                       config_json=None, with_version=True):
+    """jiancai 租户 + main(JC_DOMAIN)+ es(brand_overlay)+ 版本行。"""
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = Tenant(slug="jiancai", name="HESUPIPE 建材", status="active")
+        session.add(tenant)
+        await session.flush()
+        main = Site(tenant_id=tenant.id, slug="main", name="main")
+        es = Site(tenant_id=tenant.id, slug="es", name="es",
+                  brand_overlay={"title": "HESUPIPE Soporte"})
+        session.add_all([main, es])
+        await session.flush()
+        session.add(SiteDomain(domain=JC_DOMAIN, site_id=main.id))
+        if with_version:
+            session.add(TenantConfigVersion(
+                tenant_id=tenant.id, version=version,
+                config_json=config_json if config_json is not None
+                else json.dumps(_jiancai_snapshot(prefix), ensure_ascii=False)))
+        await session.commit()
+        return tenant.id
+
+
+async def _add_version(engine, tenant_id, version, prefix):
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        session.add(TenantConfigVersion(
+            tenant_id=tenant_id, version=version,
+            config_json=json.dumps(_jiancai_snapshot(prefix),
+                                   ensure_ascii=False)))
+        await session.commit()
+
+
+async def _add_tenant(engine, slug, domain, prefix):
+    """极简单租户(无 es 覆盖)——upsert 测试用。"""
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = Tenant(slug=slug, name=slug)
+        session.add(tenant)
+        await session.flush()
+        site = Site(tenant_id=tenant.id, slug="main", name="main")
+        session.add(site)
+        await session.flush()
+        session.add(SiteDomain(domain=domain, site_id=site.id))
+        session.add(TenantConfigVersion(
+            tenant_id=tenant.id, version=1,
+            config_json=json.dumps(_jiancai_snapshot(prefix),
+                                   ensure_ascii=False)))
+        await session.commit()
+
+
+@pytest.fixture
+async def platform_db_with_tenant(tmp_path):
+    """tmp platform.db + jiancai(main+es 含 brand_overlay)+域名+版本1。"""
+    engine = await _make_engine(tmp_path)
+    await _add_jiancai(engine)
+    yield engine
+    await engine.dispose()
+
+
+# ---- DB 源加载 ---------------------------------------------------------------
+
+
+async def test_db_source_loads_tenant(platform_db_with_tenant):
+    src = DbTenantSource(platform_db_with_tenant)
+    loaded = await src.load_all()
+    assert [lt.slug for lt in loaded] == ["jiancai"]
+    lt = loaded[0]
+    assert lt.name == "HESUPIPE 建材" and lt.status == "active"
+    assert lt.snapshot["push"]["prefix"] == "ai-kefu-jc"    # 快照组直读
+    assert "system" in lt.snapshot["prompts"]               # prompts 嵌 config_json
+    assert lt.seed_dir is None                              # DB 源无种子目录
+    main = [s for s in lt.sites if s["slug"] == "main"][0]
+    es = [s for s in lt.sites if s["slug"] == "es"][0]
+    assert main["domains"] == [JC_DOMAIN]
+    assert es["brand"]["title"] == "HESUPIPE Soporte"       # brand_overlay 生效
+    assert es["domains"] == []
+
+
+async def test_db_source_load_tenant_single_and_unknown(
+        platform_db_with_tenant):
+    src = DbTenantSource(platform_db_with_tenant)
+    lt = await src.load_tenant("jiancai")
+    assert lt.slug == "jiancai"
+    with pytest.raises(LookupError, match="nope"):
+        await src.load_tenant("nope")
+
+
+async def test_db_source_empty_db_raises(tmp_path):
+    engine = await _make_engine(tmp_path)
+    try:
+        with pytest.raises(ValueError, match="没有任何租户"):
+            await DbTenantSource(engine).load_all()
+    finally:
+        await engine.dispose()
+
+
+async def test_db_source_no_version_raises(tmp_path):
+    """有租户无版本 → 显性 ValueError 带租户名(哪个租户缺版本一目了然)。"""
+    engine = await _make_engine(tmp_path)
+    try:
+        await _add_jiancai(engine, with_version=False)
+        with pytest.raises(ValueError) as excinfo:
+            await DbTenantSource(engine).load_all()
+        assert "jiancai" in str(excinfo.value) and "没有任何配置版本" in str(
+            excinfo.value)
+    finally:
+        await engine.dispose()
+
+
+async def test_db_source_bad_json_raises(tmp_path):
+    """config_json 无效 JSON → 带租户名与版本号的清晰错误。
+
+    CHECK(json_valid) 挡住正常写入方,坏行只能来自约束启用前的历史/
+    外部直改——经 PRAGMA ignore_check_constraints 直插构造,防御路径
+    仍要显性报错。
+    """
+    engine = await _make_engine(tmp_path)
+    try:
+        tid = await _add_jiancai(engine)
+        async with engine.connect() as conn:
+            await conn.exec_driver_sql("PRAGMA ignore_check_constraints=ON")
+            await conn.execute(
+                text("INSERT INTO tenant_config_versions "
+                     "(tenant_id, version, config_json, updated_by, created_at) "
+                     "VALUES (:t, :v, :c, 'test', '2026-01-01 00:00:00')"),
+                {"t": tid, "v": 2, "c": "{oops"})
+            await conn.commit()
+        with pytest.raises(ValueError) as excinfo:
+            await DbTenantSource(engine).load_tenant("jiancai")
+        msg = str(excinfo.value)
+        assert "jiancai" in msg and "版本 2" in msg and "JSON" in msg
+    finally:
+        await engine.dispose()
+
+
+async def test_db_source_non_object_json_raises(tmp_path):
+    """合法 JSON 但非对象(如数组)→ 显性错误带租户名与版本号。"""
+    engine = await _make_engine(tmp_path)
+    try:
+        await _add_jiancai(engine, config_json="[1,2]")
+        with pytest.raises(ValueError) as excinfo:
+            await DbTenantSource(engine).load_all()
+        msg = str(excinfo.value)
+        assert "jiancai" in msg and "版本 1" in msg and "不是 JSON 对象" in msg
+    finally:
+        await engine.dispose()
+
+
+async def test_db_source_missing_prompts_group_raises(tmp_path):
+    """快照缺 wall/prompts 组 → 显性错误带租户名与版本号(不留给
+    compile_wall/from_config 出无上下文的错)。"""
+    engine = await _make_engine(tmp_path)
+    try:
+        snapshot = _jiancai_snapshot()
+        del snapshot["prompts"]
+        await _add_jiancai(engine, config_json=json.dumps(
+            snapshot, ensure_ascii=False))
+        with pytest.raises(ValueError) as excinfo:
+            await DbTenantSource(engine).load_all()
+        msg = str(excinfo.value)
+        assert "jiancai" in msg and "prompts" in msg and "版本 1" in msg
+    finally:
+        await engine.dispose()
+
+
+# ---- 文件源(直测;等价性由既有 test_registry 全绿证明)-----------------------
+
+
+async def test_file_source_loads_same_snapshot_shape():
+    src = FileTenantSource(SEED_DIR)
+    loaded = await src.load_all()
+    by_slug = {lt.slug: lt for lt in loaded}
+    assert set(by_slug) == {"jiancai", "auto"}
+    jc = by_slug["jiancai"]
+    assert jc.snapshot["push"]["prefix"] == "ai-kefu-jc"
+    assert "system" in jc.snapshot["prompts"]     # 行为增量:四件套并入快照
+    assert jc.seed_dir == SEED_DIR / "jiancai"
+    es = [s for s in jc.sites if s["slug"] == "es"][0]
+    assert es["brand"]["title"] == "HESUPIPE Soporte"
+    # 快照 prompts 与 from_dir 读盘逐字一致(等价迁移直证)
+    assert (PromptProvider.from_config(jc.snapshot["prompts"])
+            == PromptProvider.from_dir(SEED_DIR / "jiancai" / "prompts"))
+
+
+def test_file_source_read_tenant_unknown_slug_raises(tmp_path):
+    """reload upsert 的语义地基:文件源对未知 slug 报 LookupError
+    (registry 映射回 UnknownTenantError,不与坏数据的 ValueError 混)。"""
+    src = FileTenantSource(tmp_path)      # 空目录:一切 slug 未知
+    with pytest.raises(LookupError, match="ghost"):
+        src.read_tenant("ghost")
+
+
+def test_file_source_read_tenant_rereads_disk(tmp_path):
+    """read_tenant 每次现读盘(reload 换配置的数据通路)。"""
+    tdir = tmp_path / "t1"
+    tdir.mkdir()
+    shutil.copyfile(SEED_DIR / "jiancai" / "wall.json", tdir / "wall.json")
+    shutil.copytree(SEED_DIR / "jiancai" / "prompts", tdir / "prompts")
+    (tdir / "tenant.json").write_text(json.dumps({
+        "slug": "t1", "name": "t1", "status": "active",
+        "sites": [{"slug": "main", "domains": ["a.example.com"]}],
+        "config": {"push": {"prefix": "v1"}}}, ensure_ascii=False), encoding="utf-8")
+
+    src = FileTenantSource(tmp_path)
+    assert src.read_tenant("t1").snapshot["push"]["prefix"] == "v1"
+    tenant = json.loads((tdir / "tenant.json").read_text(encoding="utf-8"))
+    tenant["config"]["push"]["prefix"] = "v2"
+    (tdir / "tenant.json").write_text(
+        json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
+    assert src.read_tenant("t1").snapshot["push"]["prefix"] == "v2"
+
+
+# ---- registry × DB 源(reload / upsert)--------------------------------------
+
+
+async def test_registry_reload_via_db_source(tmp_path):
+    engine = await _make_engine(tmp_path)
+    try:
+        tid = await _add_jiancai(engine, prefix="v1")
+        reg = TenantRegistry(
+            DbTenantSource(engine), data_dir=tmp_path / "tenants")
+        await reg.load_all()
+        old = await reg.resolve(JC_DOMAIN)
+        assert old.config.data["push"]["prefix"] == "v1"
+        assert old.prompts.system                      # prompts 经 from_config 接线
+
+        await _add_version(engine, tid, 2, prefix="v2")
+        await reg.registry_reload("jiancai")
+        rt = await reg.resolve(JC_DOMAIN)
+        assert rt is not old
+        assert rt.config.data["push"]["prefix"] == "v2"   # MAX(version) 生效
+        assert old.engine is None                     # 旧 engine 已 dispose
+        await reg.dispose_all()
+    finally:
+        await engine.dispose()
+
+
+async def test_registry_reload_upsert_new_tenant_db_source(tmp_path):
+    """source 已知但未加载的 slug(DB 新建租户)→ reload 载入插入不炸;
+    source 也不知道的 slug → UnknownTenantError(原语义保留)。"""
+    engine = await _make_engine(tmp_path)
+    try:
+        await _add_jiancai(engine)
+        reg = TenantRegistry(
+            DbTenantSource(engine), data_dir=tmp_path / "tenants")
+        await reg.load_all()
+
+        await _add_tenant(engine, "demo", "demo.example.com", prefix="dm1")
+        await reg.registry_reload("demo")             # 未加载的 slug:插入
+        rt = await reg.resolve("demo.example.com")
+        assert rt.config.slug == "demo"
+        assert rt.config.data["push"]["prefix"] == "dm1"
+
+        with pytest.raises(UnknownTenantError):
+            await reg.registry_reload("ghost")
+        await reg.dispose_all()
+    finally:
+        await engine.dispose()
+
+
+def _write_min_seed(tmp_path, tenants):
+    """最小 seed 目录(拷真实 jiancai wall/prompts,免数据重复维护)。
+    幂等:第二次调用往同一 seed 下加新租户目录。"""
+    seed = tmp_path / "seed"
+    seed.mkdir(exist_ok=True)
+    for slug, tenant in tenants.items():
+        tdir = seed / slug
+        tdir.mkdir(exist_ok=True)
+        shutil.copyfile(SEED_DIR / "jiancai" / "wall.json", tdir / "wall.json")
+        shutil.copytree(SEED_DIR / "jiancai" / "prompts", tdir / "prompts",
+                        dirs_exist_ok=True)
+        (tdir / "tenant.json").write_text(
+            json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
+    return seed
+
+
+async def test_registry_reload_upsert_new_tenant_file_source(tmp_path):
+    """文件源同款 upsert 语义:新租户目录落盘后 reload 插入可 resolve
+    (Task 5 创建向导 + File 源部署形态的数据通路)。"""
+    _write_min_seed(tmp_path, {"t1": {
+        "slug": "t1", "name": "t1", "status": "active",
+        "sites": [{"slug": "main", "domains": ["a.example.com"]}],
+        "config": {}}})
+    reg = TenantRegistry.load_from_dir(
+        tmp_path / "seed", data_dir=tmp_path / "data")
+
+    _write_min_seed(tmp_path, {"t2": {
+        "slug": "t2", "name": "t2", "status": "active",
+        "sites": [{"slug": "main", "domains": ["b.example.com"]}],
+        "config": {}}})
+    await reg.registry_reload("t2")
+    rt = await reg.resolve("b.example.com")
+    assert rt.config.slug == "t2"
+    await reg.dispose_all()

tests/test_host_routing.py

@@ -16,13 +16,12 @@ from app.platform.host_middleware import TenantHostMiddleware
 from tests.conftest import (
     AUTO_HOST,
     JIANCAI_HOST,
+    NOPE_HOST,
     SEED_DIR,
     app_test_client,
     build_app,
 )
 
-NOPE_HOST = "nope.example.com"
-
 
 # ---- 中间件挂载顺序守护 -------------------------------------------------------
 

tests/test_prompt_provider.py

@@ -136,6 +136,30 @@ def test_build_extractor_handles_none_info():
     assert "{}" in out                           # existing_info or {} 语义保留
 
 
+# ---- from_config(Phase 2 Task 4: 快照 prompts 组构造)----------------------
+
+def test_from_config_missing_key_raises():
+    # 缺一键 → 显性 ValueError 带键名(不留到 build_system 才炸)
+    with pytest.raises(ValueError, match="company_facts"):
+        PromptProvider.from_config(
+            {"system": "s", "extractor": "e", "evaluator": "v"})
+
+
+def test_from_config_non_string_key_raises():
+    # 值非字符串(JSON null 等)同样显性拒绝
+    with pytest.raises(ValueError, match="system"):
+        PromptProvider.from_config(
+            {"system": None, "company_facts": "c",
+             "extractor": "e", "evaluator": "v"})
+
+
+def test_from_config_equals_from_dir():
+    d = JC_PROMPTS
+    cfg = {name: (d / f"{name}.md").read_text(encoding="utf-8")
+           for name in ("system", "company_facts", "extractor", "evaluator")}
+    assert PromptProvider.from_config(cfg) == PromptProvider.from_dir(JC_PROMPTS)
+
+
 # ---- runtime 接线 --------------------------------------------------------
 
 def test_registry_runtime_carries_prompts(tmp_path):