fix: Registry 审查修复（JSON定位/校验测试/竞态窗口/TenantDisabledError/路径锚定/seed_dir）

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

app/platform/registry.py

@@ -7,12 +7,16 @@ platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子
 
 关键行为:
 - resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site);
-  未知 Host → UnknownTenantError(spec §4:404,不做兜底租户)。
+  未知 Host → UnknownTenantError(spec §4:404,不做兜底租户);租户停用 →
+  TenantDisabledError(Task 7 中间件映射为 503 中性"服务暂停"页,非 404)。
   resolve(host) 只是它的便捷形式——双 API 一层实现,防两套解析漂移。
 - 懒建:首次 resolve 才建 engine(WAL/pool=2)+ session_factory + ensure_schema,
-  per-slug asyncio.Lock 串行,双检防并发双建;建库失败不留半开 engine。
-- registry_reload(slug):重读磁盘→校验(坏 JSON/墙/域名冲突全在替换前抛,
-  旧条目不动)→原子替换 _by_slug 条目并重建域名索引→dispose 旧 engine
+  per-slug asyncio.Lock 串行,锁内双检防并发双建;持锁后现势检查——条目
+  已被 reload 换掉则放弃本次建库、按当前条目重走(旧 runtime 永不产生
+  孤儿 engine);建库失败 dispose 半开 engine,保持未就绪下次重试。
+- registry_reload(slug):磁盘读取/校验在锁外,swap+dispose 段与懒建/
+  dispose_all 取同一把 per-slug 锁。坏配置(无效 JSON/墙编译失败/零站点/
+  域名冲突)在替换前抛,旧条目原样服务;替换后 dispose 旧 engine
   (Windows 下不关=库文件句柄泄漏)。
 
 测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all()
@@ -40,6 +44,24 @@ class UnknownTenantError(LookupError):
     """Host/域名无法映射到任何租户(调用方语义:聊天/公开端点 404)。"""
 
 
+class TenantDisabledError(LookupError):
+    """租户 status != "active"(调用方语义:503 中性"服务暂停"页,非 404;
+    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
@@ -67,12 +89,18 @@ def _normalize_host(host: str) -> str:
 
 @dataclass
 class TenantRuntime:
-    """单租户运行时。config/db_path/compiled_wall 加载期就绪;engine/
-    session_factory 由 Registry 首次 resolve 懒建;manager 由装配线注入
-    (Task 5b),prompts 由 Task 6 注入——本任务不含。"""
+    """单租户运行时。
+
+    加载期就绪:config(TenantConfig:data=快照⊕出厂默认,含 sites)/
+    db_path(该租户 SQLite 文件路径)/seed_dir(seed_profiles/<slug>/,
+    Task 6 PromptProvider 从中读 prompts/*.md)/compiled_wall(墙编译产物)。
+    懒建:engine/session_factory(Registry 首次 resolve,ensure_schema 已跑)。
+    manager(Task 5b)由装配线另行构造注入,不在本类持有。
+    """
 
     config: TenantConfig
     db_path: pathlib.Path
+    seed_dir: pathlib.Path
     compiled_wall: CompiledWall
     engine: AsyncEngine | None = None
     session_factory: async_sessionmaker | None = None
@@ -103,8 +131,8 @@ class TenantRegistry:
         """seed_profiles/<slug>/tenant.json(+同目录 wall.json)→ Registry。
 
         加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir
-        路径拼错→静默空注册表→全量 404,必须炸)、slug 与目录名不一致、
-        零站点、wall 编译错误、域名重复绑定。
+        路径拼错→静默空注册表→全量 404,必须炸)、无效 JSON、slug 与目录
+        名不一致、零站点、wall 编译错误、域名重复绑定。
         """
         reg = cls()
         reg._seed_dir = pathlib.Path(seed_dir)
@@ -122,8 +150,7 @@ class TenantRegistry:
 
     def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime:
         slug = tenant_dir.name
-        tenant_json = json.loads(
-            (tenant_dir / "tenant.json").read_text(encoding="utf-8"))
+        tenant_json = _read_json(tenant_dir / "tenant.json")
         if tenant_json.get("slug") != slug:
             raise ValueError(
                 f"{tenant_dir / 'tenant.json'}: slug 字段 "
@@ -133,8 +160,7 @@ class TenantRegistry:
             raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)")
         # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/
         # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译)
-        wall_cfg = json.loads(
-            (tenant_dir / "wall.json").read_text(encoding="utf-8"))
+        wall_cfg = _read_json(tenant_dir / "wall.json")
         snapshot = dict(tenant_json.get("config") or {})
         snapshot["wall"] = wall_cfg
         config = TenantConfig.from_snapshot(
@@ -143,6 +169,7 @@ class TenantRegistry:
         return TenantRuntime(
             config=config,
             db_path=self._data_dir / f"{slug}.db",
+            seed_dir=tenant_dir,
             compiled_wall=compile_wall(wall_cfg))
 
     def _build_domain_index(
@@ -155,27 +182,34 @@ class TenantRegistry:
                     key = _normalize_host(domain)
                     if not key:
                         continue
-                    if index.get(key, (slug, site.slug)) != (slug, site.slug):
+                    if key in index and index[key] != (slug, site.slug):
                         raise ValueError(
-                            f"域名 {key} 同时绑定 {index[key][0]} 与 "
-                            f"{slug} 的站点(域名不可重复绑定)")
+                            f"域名 {key} 已绑定 (tenant={index[key][0]}, "
+                            f"site={index[key][1]}),不能同时绑定 "
+                            f"(tenant={slug}, site={site.slug})")
                     index[key] = (slug, site.slug)
         return index
 
     # ---- 解析(对外 API)--------------------------------------------------
 
     async def resolve_site(self, host: str) -> tuple[TenantRuntime, Site]:
-        """host → (TenantRuntime, Site)。首次 resolve 懒建该租户 engine。"""
+        """host → (TenantRuntime, Site)。未知 Host → UnknownTenantError;
+        租户停用 → TenantDisabledError;首次 resolve 懒建该租户 engine。
+        返回的 runtime 以 _ensure_ready 的返回值为准(等锁期间条目可能
+        已被 reload 换成新配置)。"""
         mapped = self._domain_index.get(_normalize_host(host))
         if mapped is None:
             raise UnknownTenantError(f"未知 Host: {host!r}")
         tenant_slug, site_slug = mapped
         rt = self._by_slug[tenant_slug]
-        await self._ensure_ready(rt)
+        if rt.config.status != "active":
+            raise TenantDisabledError(f"租户已停用: {tenant_slug!r}")
+        rt = await self._ensure_ready(rt)
         site = rt.site_by_slug(site_slug)
-        if site is None:        # 防御:索引与配置同源重建,正常不可达
+        if site is None:        # reload 竞态:域名命中后站点被换掉,视同未知
             raise UnknownTenantError(
-                f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}")
+                f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}"
+                f"(可能已被 reload 移除)")
         return rt, site
 
     async def resolve(self, host: str) -> TenantRuntime:
@@ -187,50 +221,71 @@ class TenantRegistry:
     async def _ensure_ready(self, rt: TenantRuntime) -> TenantRuntime:
         """懒建 engine + session_factory + ensure_schema,per-slug Lock 串行。
 
-        双检:抢到锁后重看 engine——等锁期间前一个协程可能已建好。
-        建库失败 dispose 半开 engine 并保持未就绪,下次 resolve 重试。
+        持锁后现势检查:_by_slug 条目已被 registry_reload 换掉 → 放弃本次
+        建库(否则旧 runtime 会产出无人 dispose 的孤儿 engine),按当前
+        条目重走;双检:等锁期间前一个协程可能已建好。建库失败 dispose
+        半开 engine 并保持未就绪,下次 resolve 重试。
         """
-        if rt.engine is not None:
-            return rt
-        lock = self._locks.setdefault(rt.config.slug, asyncio.Lock())
-        async with lock:
+        slug = rt.config.slug
+        while True:
             if rt.engine is not None:
                 return rt
-            self._data_dir.mkdir(parents=True, exist_ok=True)
-            engine = make_tenant_engine(rt.db_path)
-            try:
-                await ensure_schema(engine)
-            except Exception:
-                await engine.dispose()
-                raise
-            rt.engine = engine
-            rt.session_factory = async_sessionmaker(
-                engine, class_=AsyncSession, expire_on_commit=False)
-        return rt
+            lock = self._locks.setdefault(slug, asyncio.Lock())
+            async with lock:
+                current = self._by_slug.get(slug)
+                if current is None:
+                    raise UnknownTenantError(f"租户已不存在: {slug!r}")
+                if current is not rt:
+                    rt = current              # 条目已被 reload 换掉,按新条目重走
+                    continue
+                if rt.engine is not None:     # 双检:等锁期间已建好
+                    return rt
+                self._data_dir.mkdir(parents=True, exist_ok=True)
+                engine = make_tenant_engine(rt.db_path)
+                try:
+                    await ensure_schema(engine)
+                except Exception:
+                    await engine.dispose()
+                    raise
+                rt.engine = engine
+                rt.session_factory = async_sessionmaker(
+                    engine, class_=AsyncSession, expire_on_commit=False)
+            return rt
 
     async def registry_reload(self, slug: str) -> None:
         """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。
 
-        先全量校验(重读 tenant.json/wall.json + 候选域名索引)再替换——
-        坏配置抛异常时旧条目原样服务,不留半更新状态。
+        磁盘读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all
+        取同一把 per-slug 锁:懒建要么完整发生在替换前(dispose 得到旧
+        engine),要么持锁后发现条目已换而放弃——不存在"建到一半被换"
+        的孤儿窗口。坏配置在替换前抛,旧条目原样服务,不留半更新状态。
         """
         if slug not in self._by_slug:
             raise UnknownTenantError(f"租户不存在: {slug!r}")
-        old = self._by_slug[slug]
         rt = self._load_tenant(self._seed_dir / slug)
-        candidate = dict(self._by_slug)
-        candidate[slug] = rt
-        new_index = self._build_domain_index(candidate)
-        self._by_slug[slug] = rt                 # 原子替换(单事件循环内无 await)
-        self._domain_index = new_index
-        if old.engine is not None:
-            await old.engine.dispose()
-            old.engine = old.session_factory = None
+        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._domain_index = new_index
+            if old is not None and old.engine is not None:
+                await old.engine.dispose()
+                old.engine = old.session_factory = None
 
     async def dispose_all(self) -> None:
         """关闭全部 engine 并复位为未就绪(测试 teardown 用——Windows 文件锁;
-        复位后若再 resolve 会重新懒建,可安全复用)。"""
-        for rt in self._by_slug.values():
-            if rt.engine is not None:
-                await rt.engine.dispose()
-                rt.engine = rt.session_factory = None
+        复位后若再 resolve 会重新懒建,可安全复用)。
+
+        与懒建/reload 取同一把 per-slug 锁:dispose 期间该租户不会有并发
+        建库(否则新建的 engine 会被本次复位漏掉或反过来被关)。迭代前
+        快照 _locks——迭代中并发 resolve 可能 setdefault 新锁。
+        """
+        for slug, lock in list(self._locks.items()):
+            async with lock:
+                rt = self._by_slug.get(slug)
+                if rt is not None and rt.engine is not None:
+                    await rt.engine.dispose()
+                    rt.engine = rt.session_factory = None

seed_profiles/README.md

@@ -0,0 +1,30 @@
+# seed_profiles —— 出厂租户档案
+
+每个子目录是一个租户的出厂档案（Phase 1 的配置来源；Phase 2 换 platform.db 后作为"新建租户"的模板与"重置出厂"的基准）。
+
+## 目录结构
+
+```
+seed_profiles/<slug>/
+├── tenant.json        # 站点 + 品牌 + 引擎配置（本文件是租户装配入口）
+├── wall.json          # 安全墙词表/话术（Task 4 由老项目 AST 提取生成）
+├── prompts/           # 提示词 4 件套 md（Task 6 拷入）
+│   ├── system.md / company_facts.md / extractor.md / evaluator.md
+└── brand_values.json  # 品牌值机器提取草稿（人工归一参考，运行时不读）
+```
+
+## tenant.json 要点
+
+- **slug 不可变**：客户端存储 key（邀请卡状态、会话令牌）全部以 slug 命名，改动 = 重置所有客户页面记忆；site slug 同理。`slug` 字段必须与目录名一致（加载期校验）。
+- **站点模型**：域名挂在站点上，站点属于租户。每租户**至少一个站点**。站点两种接入方式（jiancai 的两个站点即示范）：
+  - **绑域名**：`main` 站 `"domains": ["chat2.agent.hnxinyouhe.com", ...]`，按请求 Host 自动识别（大小写/端口会先归一）；
+  - **空域名 + `data-site`**：`es` 站 `"domains": []`，与 main 共用挂件域名，嵌入代码以 `data-site="es"` 参数选择站点。
+- **`site.brand` 是覆盖层**：缺省项继承 `config.brand`（深合并），只覆盖门面（标题/欢迎语等）；AI 脑子与线索管道留在租户级。
+- **wall 不写在 tenant.json 里**：加载器会把同目录 `wall.json` **隐式并入** `config["wall"]`（原始词表，供后台编辑/回滚），编译产物挂 runtime 的 `compiled_wall` 供引擎消费——词表数据源只有 wall.json 一份。
+- **`config` 内不含 prompts**：提示词由目录内 `prompts/*.md` 提供（Task 6）。
+- 加载期显性失败（不流入运行期）：无效 JSON、slug 与目录名不一致、零站点、wall 编译错误、域名跨站点/跨租户重复绑定。
+
+## 纪律
+
+- 改动本目录后必须跑全量 pytest（出厂配置影响面大，见 CLAUDE.md）。
+- `brand_values.json` 是 Task 4 的机器提取草稿（`conflict=true` 处需人工归一），`tenant.json` 的品牌值应与其保持一致；它本身不被运行时读取。

tests/test_registry.py

@@ -1,23 +1,56 @@
-"""TenantRegistry 单元契约(Task 5a 文件加载器)。
+"""TenantRegistry 单元契约(Task 5a 文件加载器 + 审查修复)。
 
 seed_profiles/<slug>/tenant.json + 同目录 wall.json → Registry;
 Host(小写/去端口)→站点→租户;首次 resolve 懒建 engine(WAL/pool=2)+
-ensure_schema;reload 原子换条目并 dispose 旧 engine。
+ensure_schema;reload 原子换条目并 dispose 旧 engine;停用租户
+TenantDisabledError(503 语义);加载期校验显性炸。
 
 Windows 纪律:每个建过 engine 的用例结尾 await reg.dispose_all()——
 连接不关,文件锁着,tmp_path 清理 PermissionError。
 """
 import asyncio
+import json
+import shutil
+from pathlib import Path
 
 import pytest
 
+import app.platform.registry as registry_mod
 from app.platform.registry import (
+    TenantDisabledError,
     TenantRegistry,
     UnknownTenantError,
     make_tenant_engine,
 )
 
-SEED_DIR = "seed_profiles"
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SEED_DIR = REPO_ROOT / "seed_profiles"
+
+
+def _tenant(slug, domains=("a.example.com",), **extra):
+    """最小合法 tenant.json(dict 形态);extra 覆盖顶层键。"""
+    tenant = {"slug": slug, "name": slug, "status": "active",
+              "sites": [{"slug": "main", "domains": list(domains)}],
+              "config": {}}
+    tenant.update(extra)
+    return tenant
+
+
+def _write_seed(tmp_path, tenants):
+    """tmp 下造最小 seed 目录:每租户目录 = 入参 tenant.json + 拷贝真实
+    jiancai wall.json(墙数据不重复维护,compile_wall 直接可过)。"""
+    seed = tmp_path / "seed"
+    seed.mkdir()
+    for slug, tenant in tenants.items():
+        tdir = seed / slug
+        tdir.mkdir()
+        shutil.copyfile(SEED_DIR / "jiancai" / "wall.json", tdir / "wall.json")
+        (tdir / "tenant.json").write_text(
+            json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
+    return seed
+
+
+# ---- 基础解析 -------------------------------------------------------------
 
 
 async def test_resolve_by_host(tmp_path):
@@ -74,8 +107,6 @@ async def test_host_normalization_case_and_port(tmp_path):
 
 async def test_concurrent_first_resolve_builds_engine_once(tmp_path, monkeypatch):
     """两个协程同时首 resolve 同一租户:engine 只建一次(asyncio.Lock 串行)。"""
-    import app.platform.registry as registry_mod
-
     calls = []
     real_make = registry_mod.make_tenant_engine
 
@@ -105,3 +136,161 @@ async def test_make_tenant_engine_enables_wal(tmp_path):
         assert str(mode).lower() == "wal"
     finally:
         await engine.dispose()
+
+
+# ---- 加载期校验(I-4) ------------------------------------------------------
+
+
+def test_load_empty_seed_dir_raises(tmp_path):
+    seed = tmp_path / "seed"
+    seed.mkdir()
+    with pytest.raises(ValueError, match="没有任何"):
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+
+
+def test_load_invalid_json_raises_with_path(tmp_path):
+    """坏 JSON 必须带文件路径显性炸(哪份文件坏了一目了然)。"""
+    seed = tmp_path / "seed"
+    (seed / "jiancai").mkdir(parents=True)
+    (seed / "jiancai" / "tenant.json").write_text("{oops", encoding="utf-8")
+    with pytest.raises(ValueError, match=r"tenant\.json.*无效 JSON"):
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+
+
+def test_load_slug_dir_mismatch_raises(tmp_path):
+    seed = _write_seed(tmp_path, {"dirslug": _tenant("jsonslug")})
+    with pytest.raises(ValueError, match="不一致"):
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+
+
+def test_load_zero_sites_raises(tmp_path):
+    tenant = _tenant("jiancai")
+    tenant["sites"] = []
+    seed = _write_seed(tmp_path, {"jiancai": tenant})
+    with pytest.raises(ValueError, match="至少一个站点"):
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+
+
+def test_load_domain_conflict_across_tenants_raises(tmp_path):
+    seed = _write_seed(tmp_path, {
+        "t1": _tenant("t1", ["dup.example.com"]),
+        "t2": _tenant("t2", ["dup.example.com"])})
+    with pytest.raises(ValueError, match="dup.example.com"):
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+
+
+def test_load_domain_conflict_within_tenant_raises(tmp_path):
+    """同租户两个站点抢同一域名也要炸,报错含双方 (tenant, site)。"""
+    tenant = _tenant("t1", ["dup.example.com"])
+    tenant["sites"].append({"slug": "second", "domains": ["dup.example.com"]})
+    seed = _write_seed(tmp_path, {"t1": tenant})
+    with pytest.raises(ValueError) as excinfo:
+        TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+    msg = str(excinfo.value)
+    assert "main" in msg and "second" in msg and "t1" in msg
+
+
+# ---- reload / dispose 行为 -------------------------------------------------
+
+
+async def test_reload_bad_config_keeps_old_entry_serving(tmp_path):
+    """reload 遇坏配置必须先炸后不动:旧条目(含域名索引)原样服务。"""
+    tenant = _tenant("jiancai", ["a.example.com"])
+    tenant["config"] = {"push": {"prefix": "v1"}}
+    seed = _write_seed(tmp_path, {"jiancai": tenant})
+    reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+    rt1 = await reg.resolve("a.example.com")
+    assert rt1.config.data["push"]["prefix"] == "v1"
+
+    tenant["sites"] = []                      # 改坏:零站点
+    (seed / "jiancai" / "tenant.json").write_text(
+        json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
+    with pytest.raises(ValueError, match="至少一个站点"):
+        await reg.registry_reload("jiancai")
+
+    rt = await reg.resolve("a.example.com")   # 旧条目继续服务
+    assert rt is rt1 and rt.config.data["push"]["prefix"] == "v1"
+    await reg.dispose_all()
+
+
+async def test_dispose_all_then_resolve_rebuilds(tmp_path):
+    """dispose_all 复位为未就绪,再 resolve 重新懒建(可安全复用)。"""
+    reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
+    rt = await reg.resolve("chat2.agent.hnxinyouhe.com")
+    engine_before = rt.engine
+    await reg.dispose_all()
+    assert rt.engine is None
+    rt2 = await reg.resolve("chat2.agent.hnxinyouhe.com")
+    assert rt2 is rt
+    assert rt2.engine is not None and rt2.engine is not engine_before
+    await reg.dispose_all()
+
+
+async def test_resolve_waiting_through_reload_serves_new_entry(tmp_path):
+    """I-2 竞态:resolve 排队等锁期间条目被 reload 换掉——持锁后现势
+    检查必须放弃旧 runtime 按新条目建库(旧 runtime 永不产生孤儿 engine)。
+
+    构造:测试占住 per-slug 锁 → reload 与 resolve 先后排到锁等待
+    (FIFO:reload 先拿锁换条目)→ 释放后 resolve 持锁发现条目已换。
+    """
+    reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
+    rt_old = reg._by_slug["jiancai"]            # 单元测试:直查内部状态
+    lock = reg._locks.setdefault("jiancai", asyncio.Lock())
+    async with lock:
+        reloader = asyncio.create_task(reg.registry_reload("jiancai"))
+        resolver = asyncio.create_task(reg.resolve("chat2.agent.hnxinyouhe.com"))
+        await asyncio.sleep(0)                   # 两任务均跑到锁等待
+        await asyncio.sleep(0)
+    await asyncio.gather(reloader, resolver)
+
+    rt = resolver.result()
+    assert rt is reg._by_slug["jiancai"]         # 服务 reload 后的新条目
+    assert rt is not rt_old
+    assert rt_old.engine is None                 # 旧 runtime 没被建库(无孤儿句柄)
+    assert rt.engine is not None
+    await reg.dispose_all()
+
+
+# ---- 停用租户(I-3) ---------------------------------------------------------
+
+
+async def test_disabled_tenant_resolve_raises_and_never_builds_engine(
+        tmp_path, monkeypatch):
+    """status=disabled:resolve 抛 TenantDisabledError(503 语义,非 404),
+    且从不建 engine。"""
+    calls = []
+    real_make = registry_mod.make_tenant_engine
+
+    def counting_make(db_path):
+        calls.append(db_path)
+        return real_make(db_path)
+
+    monkeypatch.setattr(registry_mod, "make_tenant_engine", counting_make)
+
+    seed = _write_seed(tmp_path, {"jiancai": _tenant("jiancai", status="disabled")})
+    reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+    with pytest.raises(TenantDisabledError):
+        await reg.resolve("a.example.com")
+    with pytest.raises(TenantDisabledError):     # 再 resolve 同样拒绝
+        await reg.resolve("a.example.com")
+    assert calls == []                           # 从未建 engine
+    await reg.dispose_all()
+
+
+async def test_reload_to_disabled_releases_engine(tmp_path):
+    """active→disabled 热切换:旧 engine 随 reload dispose,resolve 转抛。"""
+    tenant = _tenant("jiancai", ["a.example.com"])
+    seed = _write_seed(tmp_path, {"jiancai": tenant})
+    reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
+    rt1 = await reg.resolve("a.example.com")
+    assert rt1.engine is not None
+
+    tenant["status"] = "disabled"
+    (seed / "jiancai" / "tenant.json").write_text(
+        json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
+    await reg.registry_reload("jiancai")
+
+    assert rt1.engine is None                    # 旧 engine 已释放
+    with pytest.raises(TenantDisabledError):
+        await reg.resolve("a.example.com")
+    await reg.dispose_all()