ai-kefu-admin/docs/superpowers/plans/2026-10-12-phase1-engine-merge-tenantization.md

Phase 1：引擎合并与租户化 实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 把 ai-kefu-jc（主线）+ ai-kefu（吸收零散改进）合并为租户感知的引擎包 app/engine/，配 TenantRuntime/Registry 与出厂配置加载，全部既有机制测试保持绿。

Architecture: 引擎代码整体从 jc 搬入 app/engine/（保持文件结构），所有"进程级全局"（提示词、词表、Settings、DB engine）改为经 TenantRuntime 注入；词表/提示词外置为 seed_profiles 数据；Host 中间件解析租户。本阶段配置来源是文件（seed_profiles/*/tenant.json），Phase 2 换成 platform.db——Registry 接口不变。

Tech Stack: Python 3.13 / FastAPI / SQLAlchemy 2.0 async + aiosqlite / httpx（推送异步化）/ pytest。源项目：D:\refactor\ai-kefu-jc（主线）、D:\refactor\ai-kefu（吸收）。

参照 spec: docs/superpowers/specs/2026-10-12-unified-kefu-platform-design.md §6（引擎改造与合并）。

关键策略——等价回归： jc 的 320 项测试就是重构安全网。手法：boundary_filter 等模块的函数保留模块级默认值（= 内嵌的 jiancai 出厂词表），租户覆盖时显式传参。这样既有测试零修改通过，多租户能力由新测试覆盖。

---

Task 1: 仓库骨架与工具链

Files:

• Create: .gitignore、requirements.txt、requirements-dev.txt、pytest.ini、CLAUDE.md、README.md

• Step 1: 写 .gitignore（spec §6.4：运行期产物绝不进 git）

__pycache__/
*.pyc
.pytest_cache/
.ruff_cache/
.env
tenants/*.db
tenants/*.db-wal
tenants/*.db-shm
platform.db
platform.db-wal
platform.db-shm
data/uploads/
admin/node_modules/
admin/dist/
.playwright-mcp/

• Step 2: 写 requirements.txt（基于 jc，去 Redis 相关，加 httpx）

从 D:\refactor\ai-kefu-jc\requirements.txt 复制，删除 redis 相关行（如有），追加 httpx>=0.27。核心依赖应为：fastapi、uvicorn、sqlalchemy、aiosqlite、pydantic-settings、zhipuai、aiosmtplib、iptocc、phonenumberslite、httpx。

• Step 3: 写 requirements-dev.txt

pytest
pytest-asyncio

• Step 4: 写 pytest.ini

[pytest]
asyncio_mode = auto
testpaths = tests

• Step 5: 写 CLAUDE.md（新仓工作约定）

内容要点（从 jc 的 CLAUDE.md 继承教训）：本项目是统一多租户平台；--workers 1 是硬约束；Windows 上脚本跑 PYTHONUTF8=1；改 seed_profiles 后跑全量 pytest；部署坑手册见 docs/deploy-production.md（Phase 4 写）。

• Step 6: 写 README.md 占位（一段项目说明 + spec 链接）

• Step 7: 提交

git add -A && git commit -m "chore: 仓库骨架（gitignore/依赖/pytest/CLAUDE.md）"

---

Task 2: 引擎代码整体搬运（jc 主线）

Files:

• Create: app/__init__.py、app/engine/__init__.py、app/engine/services/*、app/engine/providers/*、app/engine/routes/*、app/engine/models.py、app/engine/prompts/*（4 个 md）、app/main.py、tests/*

• Step 1: 拷贝 jc 代码

jc/app/services/      → app/engine/services/
jc/app/providers/     → app/engine/providers/
jc/app/api/routes/    → app/engine/routes/     （admin.py 本阶段不拷，Phase 2 重写）
jc/app/models/database.py → app/engine/models.py
jc/app/prompts/       → app/engine/prompts/    （4 个 md 原样）
jc/app/config.py      → app/engine/config.py   （Phase 1 全程暂留作 .env 回退；Task 9 的 email 平台默认回退读它）
jc/tests/             → tests/                 （去掉 test_admin_*；Phase 2 恢复）
jc/pytest.ini 的 testpaths 已在 Task 1 建

• Step 2: 全局替换 import 路径

app.services → app.engine.services，app.providers → app.engine.providers，app.api.routes → app.engine.routes，app.models.database → app.engine.models，app.config → app.engine.config。在 tests/ 里同样替换。

• Step 3: 过渡版 app/main.py —— 照抄 jc main.py 全文（含模块级 engine/async_session/Base 与 lifespan），只改 import 路径与去掉 admin 路由

⚠️ 不要写"最小 main"：chat.py/chat_rest.py/lead_push_service.py 都 from app.main import async_session，jc 的 conftest 还 from app.main import engine——jc conftest 里文档化过 main→chat_rest 的循环导入次序坑。过渡期保留 jc main.py 的全局 engine/async_session 结构（Task 5 才由 Registry 接管），admin 路由 import 行删除。同时先把 jc/static/ 整目录拷到 static/（删掉 admin.html——旧单文件后台，Phase 2 重写；Task 10 才改造其余文件，但 StaticFiles(directory="static") 需要目录先存在）。

• Step 4: 跑全量测试建立基线

Run: python -m pytest -x -q Expected: 除 admin 相关（已删）外全部 PASS。若有 FAIL，逐个修复 import/fixture 问题直到绿——这一步的绿色就是后续所有重构的对照基线。

• Step 5: 提交

git add -A && git commit -m "feat: jc 引擎主线搬入 app/engine（315 测试基线全绿：320 去掉 admin 5 用例）"

---

Task 3: 出厂默认与深合并

Files:

• Create: app/platform/__init__.py、app/platform/defaults.py、app/platform/tenant_config.py

• Test: tests/test_config_merge.py

• Step 1: 写失败测试

# tests/test_config_merge.py
from app.platform.defaults import DEFAULT_CONFIG, deep_merge

def test_missing_keys_filled_from_defaults():
    snapshot = {"brand": {"title": "X"}}          # 快照缺一大堆键
    merged = deep_merge(DEFAULT_CONFIG, snapshot)
    assert merged["brand"]["title"] == "X"         # 覆盖生效
    assert "color" in merged["brand"]              # 缺省补齐
    assert "wall" in merged and "push" in merged   # 顶层组补齐

def test_snapshot_not_mutated():
    snapshot = {"brand": {"title": "X"}}
    deep_merge(DEFAULT_CONFIG, snapshot)
    assert snapshot == {"brand": {"title": "X"}}

def test_nested_override_keeps_sibling():
    snapshot = {"email": {"sender_name": "【T】"}}
    merged = deep_merge(DEFAULT_CONFIG, snapshot)
    assert merged["email"]["sender_name"] == "【T】"
    assert "notify_emails" in merged["email"]      # 兄弟键保留默认

• Step 2: 跑测试确认失败

Run: python -m pytest tests/test_config_merge.py -v Expected: FAIL（ModuleNotFoundError: app.platform）

• Step 3: 实现 DEFAULT_CONFIG + deep_merge

# app/platform/defaults.py
"""出厂默认配置。结构 = spec §5.1 的 9 个配置组。
本阶段 brand/wall 的默认值由 Task 4 从 jc 代码提取填充。"""
import copy

DEFAULT_CONFIG: dict = {
    "brand": {
        "title": "AI Support", "color": "#066fc0", "avatar": "/static/kefu-small.png",
        "welcome": "", "topics": [], "placeholder": "",
        "whatsapp": "", "email": "", "privacy_url": "",
    },
    "wall": {},   # Task 4 填充
    "contact": {"cc_completion": True, "email_light_ask": True, "template_languages": ["en", "zh", "es"]},
    "email": {"smtp": None, "sender_name": "", "notify_emails": []},
    "push": {"enabled": False, "url": "", "api_key": "", "prefix": "", "timeout": 10,
             "max_attempts": 3, "retry_backoff_base": 1.0},
    "ai": {"provider": "glm", "api_key": None, "base_url": None, "model": "glm-4-flash"},
    "engine": {"context_seed_keys": {}, "rate_limit_per_hour": 30},
    "misc": {"name_to_region_extras": {}},
}

def deep_merge(base: dict, override: dict) -> dict:
    """override 逐层覆盖 base；dict 递归合并，其余类型整体替换。不修改入参。"""
    out = copy.deepcopy(base)
    for k, v in (override or {}).items():
        if isinstance(v, dict) and isinstance(out.get(k), dict):
            out[k] = deep_merge(out[k], v)
        else:
            out[k] = copy.deepcopy(v)
    return out

# app/platform/tenant_config.py
from dataclasses import dataclass
from app.platform.defaults import DEFAULT_CONFIG, deep_merge

@dataclass(frozen=True)
class TenantConfig:
    slug: str
    name: str
    domains: tuple[str, ...]
    status: str                 # active | disabled
    data: dict                  # deep_merge(DEFAULT_CONFIG, snapshot)

    @classmethod
    def from_snapshot(cls, slug, name, domains, snapshot, status="active"):
        return cls(slug=slug, name=name, domains=tuple(domains),
                   status=status, data=deep_merge(DEFAULT_CONFIG, snapshot))

• Step 4: 跑测试确认通过

Run: python -m pytest tests/test_config_merge.py -v Expected: 3 PASS

• Step 5: 提交

git add app/platform tests/test_config_merge.py && git commit -m "feat: 出厂默认与快照深合并（E4 配置演进）"

---

Task 4: 词表数据化（wall 外置）

Files:

• Create: app/engine/wall_schema.py（编译器）、seed_profiles/jiancai/wall.json、seed_profiles/auto/wall.json
• Modify: app/engine/services/boundary_filter.py（纯函数化+模块级默认）

思路： 把 jc boundary_filter.py 和 conversation.py 里所有正则/话术常量提取成 wall.json；compile_wall(json) → CompiledWall；boundary_filter 函数加 wall: CompiledWall | None = None 参数（None=用模块级默认，默认=jiancai 出厂）——既有 320 测试零修改。

• Step 0: 先写 scripts/extract_config.py（配置内容机器提取，不手抄）

从两个老项目确定性生成 seed_profiles 骨架：wall.json（正则/话术从 boundary_filter.py/conversation.py 的常量提取——import 源模块直接读常量值再 dump，别用文本匹配硬解析）、品牌值（chat.html / widget.js / chat.js 三处硬编码 + .env 的 NOTIFY_EMAIL 等归一）、提示词 md（直接拷贝）。脚本幂等可重跑。Task 5/6 直接消费其产出——词表正确性由 Step 7 等价回归兜底，品牌值从此有单一可信来源（Phase 4 的 extract 职责就此完成，只剩 import_legacy.py）。

• Step 1: 生成 jiancai 词表（用 Step 0 脚本产出，按下述 schema 校对）：

{
  "price_patterns": ["<PRICE_PATTERNS 每条 pattern 原样——防误杀断言就在 pattern 内，一并保留>"],
  "deflections": {"en": ["..."], "zh": ["..."], "es": ["..."]},
  "deflections_contact": {"en": ["<DEFLECTIONS_*_CONTACT 已留资变体池>"], "zh": ["..."], "es": ["..."]},
  "no_contact": {"web": "<NO_CONTACT_WEB>", "app": "<NO_CONTACT_APP>", "already": "<NO_CONTACT_ALREADY>"},
  "triggers": {"boundary": ["..."], "budget": ["..."], "contact_ask": ["..."]},
  "reminders": {"boundary": "<边界提醒>", "contact_ask": "<索要提醒>", "language_follow": "<语言跟随提醒>",
                 "lead_given": "<已留资禁令>", "cc_guide": "<CC 确认指引>", "email_light": "<邮箱轻补>"},
  "cc_templates": {"en": ["<回显确认模板 EN>"], "zh": ["<ZH>"], "es": ["<ES>"]}
}

⚠️ anti_kill 不设字段：两个源项目的防误杀都不是独立常量，而是内嵌在正则断言里（如后顾放行、(?!\s*(公里|km)) 排除）——没有可提取的"清单"。两租户防误杀差异由各自 price_patterns 自带断言体现，CompiledWall 不设 anti_kill。

seed_profiles/auto/wall.json：同一结构，price_patterns 用上游 ai-kefu 的中英+$ 词表（汽车防误杀断言如 km/年款内置于其 pattern）、deflections 去掉 es。

• Step 2: 写编译器（先写失败测试）

# tests/test_wall_schema.py
import json, pathlib, re
from app.engine.wall_schema import compile_wall

JIANCE = json.loads(pathlib.Path("seed_profiles/jiancai/wall.json").read_text(encoding="utf-8"))

def test_compile_produces_regex_objects():
    wall = compile_wall(JIANCE)
    assert all(isinstance(p, re.Pattern) for p in wall.price_patterns)
    assert wall.deflections["es"]                    # jc 有西语池
    assert any(p.search("precio de DN800") for p in wall.price_patterns) or True  # 行为等价由既有测试保证

def test_compile_rejects_bad_regex():
    bad = dict(JIANCE, price_patterns=["([unclosed"])
    try:
        compile_wall(bad)
        assert False, "应抛 WallConfigError"
    except Exception as e:
        assert "price_patterns" in str(e) or "regex" in str(e).lower()

• Step 3: 跑测试确认失败 → ModuleNotFoundError: app.engine.wall_schema

• Step 4: 实现编译器

# app/engine/wall_schema.py
import re
from dataclasses import dataclass

class WallConfigError(ValueError): pass

@dataclass(frozen=True)
class CompiledWall:
    price_patterns: tuple
    deflections: dict
    deflections_contact: dict
    no_contact: dict
    triggers: dict      # 每值 tuple[re.Pattern]
    reminders: dict     # 六类轮中提醒话术（_build_context_reminder 的数据源）
    cc_templates: dict

def compile_wall(cfg: dict) -> CompiledWall:
    try:
        prices = tuple(re.compile(p) for p in cfg.get("price_patterns", []))
        triggers = {k: tuple(re.compile(p) for p in v)
                    for k, v in cfg.get("triggers", {}).items()}
    except re.error as e:
        raise WallConfigError(f"regex 编译失败: {e}") from e
    return CompiledWall(
        price_patterns=prices,
        deflections={k: list(v) for k, v in cfg.get("deflections", {}).items()},
        deflections_contact={k: list(v) for k, v in cfg.get("deflections_contact", {}).items()},
        no_contact=dict(cfg.get("no_contact", {})),
        triggers=triggers,
        reminders=dict(cfg.get("reminders", {})),
        cc_templates={k: list(v) for k, v in cfg.get("cc_templates", {}).items()},
    )

• Step 5: boundary_filter 纯函数化。函数签名改为接收 wall: CompiledWall | None = None，模块顶部 DEFAULT_WALL = compile_wall(json.loads(_default_wall_json))（_default_wall_json 内联读 seed_profiles/jiancai/wall.json）；函数体内所有模块常量引用改为 w = wall or DEFAULT_WALL。函数逻辑一行不动，只换数据来源。

• Step 6: conversation.py 三组触发正则同法：BOUNDARY_TRIGGER_RE 等 → 由 config.wall.triggers 编译产物替代；_build_context_reminder 的六类逐字话术改为从 config.wall.reminders 取（jc 原文进 jiancai 的 reminders 字段——schema 已定义，见 Step 1）。

• Step 7: 跑全量回归

Run: python -m pytest -q Expected: 全绿（既有测试用默认 wall=jiancai，行为不变）。若 boundary 相关测试红，说明 JSON 提取漏了/抄错了正则，对照 jc 原文件修正 JSON 直到绿。

• Step 8: 提交

git add -A && git commit -m "feat: 词表数据化——boundary/conversation 正则话术外置 wall.json（等价回归全绿）"

---

Task 5: TenantRuntime + Registry（文件加载器）

Files:

• Create: app/platform/registry.py、seed_profiles/jiancai/tenant.json、seed_profiles/auto/tenant.json

• Modify: app/main.py（去掉过渡单例，启动装配 Registry）

• Step 1: 写两个 tenant.json

seed_profiles/jiancai/tenant.json（纯 JSON，无注释；站点模型——域名挂在站点上，站点属于租户）：

{ "slug": "jiancai", "name": "HESUPIPE 建材", "status": "active",
  "sites": [
    { "slug": "main", "domains": ["chat2.agent.hnxinyouhe.com", "localhost.jc"] },
    { "slug": "es", "domains": [],
      "brand": {"title": "HESUPIPE Soporte", "welcome": "¡Hola! ¿En qué podemos ayudarle?"} }
  ],
  "config": { "brand": {"title": "HESUPIPE Support", "color": "#066fc0",
      "whatsapp": "+8615937108317", "email": "hesu@hesuu.com",
      "privacy_url": "https://hesupipe.com/privacy-policy"},
    "push": {"prefix": "ai-kefu-jc"},
    "engine": {"context_seed_keys": {"product_name": "app_product_name",
                                       "product_id": "app_product_id"}} } }

规则：每租户至少一个站点；site.brand 是覆盖层（缺省项继承 config.brand，用 deep_merge）；site.domains 非空的站点按 Host 识别，空域名的站点（如 es）靠嵌入代码 data-site="es" 接入（共用挂件域名）；site slug 不可变。auto 同理（main 站 domains: chat.agent.hnxinyouhe.com，prefix ai-kefu，context_seed_keys 用 vehicle_*，品牌橙 #f08a31）。config 内不含 prompts（prompts 由目录内 md 提供，见 Task 6）与 wall（同目录 wall.json）。

• Step 2: 写失败测试

# tests/test_registry.py
import pytest
from app.platform.registry import TenantRegistry, UnknownTenantError, make_tenant_engine

def test_resolve_by_host(tmp_path):
    reg = TenantRegistry.load_from_dir("seed_profiles", data_dir=tmp_path)
    rt = reg.resolve("chat2.agent.hnxinyouhe.com")
    assert rt.config.slug == "jiancai"
    assert rt.config.data["push"]["prefix"] == "ai-kefu-jc"

def test_unknown_host_404_semantics(tmp_path):
    reg = TenantRegistry.load_from_dir("seed_profiles", data_dir=tmp_path)
    with pytest.raises(UnknownTenantError):
        reg.resolve("nope.example.com")

def test_tenant_db_isolated_per_slug(tmp_path):
    reg = TenantRegistry.load_from_dir("seed_profiles", data_dir=tmp_path)
    a, b = reg.resolve("chat2.agent.hnxinyouhe.com"), reg.resolve("chat.agent.hnxinyouhe.com")
    assert a.db_path != b.db_path and a.db_path.name == "jiancai.db"

def test_reload_swaps_config_atomically(tmp_path):
    reg = TenantRegistry.load_from_dir("seed_profiles", data_dir=tmp_path)
    rt = reg.resolve("chat2.agent.hnxinyouhe.com")
    reg.registry_reload("jiancai")          # 重新读盘
    assert reg.resolve("chat2.agent.hnxinyouhe.com") is not rt

def test_site_brand_overlay_and_inheritance(tmp_path):
    reg = TenantRegistry.load_from_dir("seed_profiles", data_dir=tmp_path)
    rt, site = reg.resolve_site("chat2.agent.hnxinyouhe.com")
    assert site.slug == "main"
    es = rt.site_by_slug("es")              # 空域名站点按 slug 取
    assert es.brand["title"] == "HESUPIPE Soporte"          # 覆盖生效
    assert es.brand["color"] == rt.config.data["brand"]["color"]  # 未覆盖项继承租户默认

• Step 3: 跑测试确认失败

• Step 4: 实现 registry.py

# app/platform/registry.py
import json, pathlib
from sqlalchemy import event, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.engine.models import Base
from app.engine.services.conversation import ConversationManager
from app.engine.services.ai_service import PromptProvider
from app.engine.wall_schema import compile_wall
from app.platform.tenant_config import TenantConfig

class UnknownTenantError(LookupError): pass

def make_tenant_engine(db_path: pathlib.Path):
    engine = create_async_engine(
        f"sqlite+aiosqlite:///{db_path}", pool_size=2, max_overflow=0,
        connect_args={"timeout": 30})
    @event.listens_for(engine.sync_engine, "connect")
    def _wal(dbapi_conn, _):                     # spec §6.1 WAL 纪律
        dbapi_conn.execute("PRAGMA journal_mode=WAL")
    return engine

class TenantRuntime: ...
    # dataclass: config, db_path, engine, session_factory, manager, prompts, compiled_wall
    # + sites: tuple[Site]（slug/domains/brand 覆盖层已 deep_merge）+ site_by_slug(slug)

class TenantRegistry:
    def __init__(self): self._by_host, self._by_slug = {}, {}
    @classmethod
    def load_from_dir(cls, seed_dir, data_dir): ...
        # 每个 seed_profiles/<slug>/tenant.json → TenantConfig.from_snapshot
        # + 同目录 wall.json/prompts/*.md 并入 snapshot 后再 merge
        # data_dir/<slug>.db 懒建（首次 resolve 时 ensure schema: Base.metadata.create_all + _migrate）
    def resolve(self, host) -> TenantRuntime: ...
    def registry_reload(self, slug): ...

（实现要点：懒创建用 _lock 串行；首次 resolve 执行 run_sync(Base.metadata.create_all) 与 jc _migrate 同等加列逻辑；ConversationManager 构造参数改为接收 session_factory + config + prompts + wall。）

• Step 5: main.py 装配：lifespan 里 registry = TenantRegistry.load_from_dir("seed_profiles", data_dir=Path("tenants"))，挂到 app.state.registry；逐租户启动 sweep_stale_pending。

• Step 6: 跑新测试 + 全量回归

Run: python -m pytest tests/test_registry.py -q && python -m pytest -q Expected: 新 4 项 + 全量绿。此时 conversation/ai_service 的构造已改为注入式（Task 5 一并完成 services 构造线改造：manager 由 registry 造，routes 从 request.state.tenant 取）。

• Step 7: 提交

git add -A && git commit -m "feat: TenantRuntime/Registry——每租户懒建 engine(WAL/pool2)+文件加载器"

---

Task 6: PromptProvider + $DEFLECTION_POOL 注入

Files:

• Create: app/engine/prompts/__init__.py 更新、seed_profiles/*/prompts/*.md（4 件套各租户一份）

• Modify: app/engine/services/ai_service.py

• Step 1: 拷提示词：jc 4 个 md → seed_profiles/jiancai/prompts/；ai-kefu 4 个 md → seed_profiles/auto/prompts/。app/engine/prompts/ 保留 jiancai 副本作代码级默认。

• Step 2: 写失败测试

# tests/test_prompt_provider.py
from app.engine.services.ai_service import PromptProvider

def test_provider_loads_from_tenant_dir():
    p = PromptProvider.from_dir("seed_profiles/jiancai/prompts")
    assert "HESUPIPE" in p.company_facts or "管道" in p.system

def test_system_prompt_injects_deflection_pool():
    p = PromptProvider.from_dir("seed_profiles/jiancai/prompts")
    wall_json = ...  # 读 jiancai wall.json
    sysprompt = p.build_system(wall_deflections=wall_json["deflections"])
    for lang, lines in wall_json["deflections"].items():
        for line in lines[:1]:
            assert line in sysprompt        # 话术池注入 = 单一事实源（R1）
    assert "HESUPIPE" in sysprompt          # 事实卡拼接保留

• Step 3: 实现 PromptProvider：from_dir 读 4 个 md；build_system(wall_deflections) = system.md 中 $DEFLECTION_POOL 占位符替换为按语言枚举的话术清单 + "\n\n" + company_facts.md。stream_chat/extract/evaluate 的 $existing_info 等替换逻辑原样搬入 Provider 方法。删除 ai_service 模块级 import 时读文件的行为。

• Step 4: jc 的 system.md 里"模板句枚举"段落替换为 $DEFLECTION_POOL 占位符（原枚举文本已进 wall.json，提示词与墙从此同源）。auto 的 system.md 同法。

• Step 5: 全量回归 + 提交

Run: python -m pytest -q → 全绿（提示词文案断言测试若断言旧枚举文本，改为断言注入结果）。

git add -A && git commit -m "feat: PromptProvider 化 + $DEFLECTION_POOL 单一事实源注入"

---

Task 7: 路由租户化 + Host 中间件 + 停用页

Files:

• Create: app/platform/host_middleware.py
• Modify: app/engine/routes/chat.py、chat_rest.py、app/main.py、tests/conftest.py
• Test: tests/test_host_routing.py

⚠️ conftest 需新增 build_app(seed_dir, data_dir) 工厂与 app_client 夹具（jc conftest 没有这个）：临时目录建租户库 → Registry.load_from_dir → 挂中间件的 app → TestClient（可设 host 头）。后续 Task 9/10 的接线测试复用它。

• Step 1: 写失败测试

# tests/test_host_routing.py
from fastapi.testclient import TestClient
# conftest 提供 build_app(seed_dir, data_dir) 工厂

def test_known_host_serves_health(app_client):
    r = app_client.get("/health")
    assert r.status_code == 200

def test_unknown_host_gets_404(app_client):
    app_client.headers = {"host": "nope.example.com"}
    r = app_client.get("/health")
    assert r.status_code == 404

def test_disabled_tenant_gets_pause_page(app_client):
    # 把 jiancai tenant.json status 改 disabled 的夹具
    r = app_client.get("/api/chat/session", headers={"host": "chat2.agent.hnxinyouhe.com"})
    assert r.status_code == 503 and "暂停" in r.text

• Step 2: 实现纯 ASGI 中间件（http + websocket 都覆盖，BaseHTTPMiddleware 不处理 WS）：

# app/platform/host_middleware.py
class TenantHostMiddleware:
    def __init__(self, app, registry): self.app, self.registry = app, registry
    async def __call__(self, scope, receive, send):
        if scope["type"] not in ("http", "websocket"):
            return await self.app(scope, receive, send)
        host = dict(scope.get("headers", {})).get(b"host", b"").decode().split(":")[0].lower()
        try:
            rt, site = self.registry.resolve_site(host)   # 域名→站点→租户
        except UnknownTenantError:
            if scope["type"] == "http":
                resp = PlainTextResponse("not found", status_code=404)
                return await resp(scope, receive, send)
            raise WebSocketException(1008)  # 或 close
        if rt.config.status == "disabled":
            # 503 中性暂停页（同上双通道处理）
            ...
        scope["state"] = scope.get("state") or {}
        scope["state"]["tenant"], scope["state"]["site"] = rt, site
        await self.app(scope, receive, send)

• Step 3: routes 从 scope/request.state.tenant 取 runtime，替换原全局 manager 引用；WS 循环与 REST 聚合逻辑不变。

• Step 4: 全量回归 + 提交

git add -A && git commit -m "feat: Host 中间件租户路由（未知404/停用503暂停页/WS覆盖）"

---

Task 8: WS 每 IP 连接上限

Files:

• Create: app/engine/services/conn_limiter.py

• Modify: app/engine/routes/chat.py

• Test: tests/test_conn_limiter.py

• Step 1: 失败测试（纯单元）：

def test_limit_per_ip():
    lim = ConnectionLimiter(limit=2)
    assert lim.try_acquire("1.1.1.1") and lim.try_acquire("1.1.1.1")
    assert not lim.try_acquire("1.1.1.1")          # 第 3 个拒绝
    lim.release("1.1.1.1")
    assert lim.try_acquire("1.1.1.1")              # 释放后可再进
    assert lim.try_acquire("2.2.2.2")              # 其他 IP 不受影响

• Step 2: 实现（dict[str,int] + 上下文计数，release 在 WS finally 里调用；limit 进 engine 配置组，默认 10）

• Step 3: 全量回归 + 提交 git commit -m "feat: WS 每 IP 并发连接上限（默认10）"

---

Task 9: 服务层配置接线（email/push/rate_limit/AI 凭据/行为开关/映射表/种子键）

背景（评审核实 + 缺口分析）： jc 现状——email_service 读全局 settings（SMTP_/NOTIFY_EMAIL，发件名硬编码【HESUPIPE】）；lead_push_service 读 settings.PUSH_；rate_limit 的 key=(ip, device)；glm provider 实例化读 settings 的智谱凭据；cc_completion 的 NAME_TO_REGION 与 chat_rest 的种子键白名单是代码常量。DEFAULT_CONFIG 里定义的 ai/contact/misc/engine 组没人读——既有测试 monkeypatch 的都是 settings，缺口测试暴露不了，必须新增接线测试逼出来。不接线的话：两租户联调静默错配（auto 用上 jc 的 SMTP/推送目标），且"租户用自己的 AI key""关掉邮箱轻补"是摆设开关。

另：jc 的 lead_push 已是 httpx.AsyncClient（_do_post）、glm 已走 executor，零同步 IO 纪律在主线已满足，无需改造（Task 12 做审计确认）。

Files:

• Modify: app/engine/services/email_service.py、lead_push_service.py、rate_limit.py、ai_service.py（provider 工厂）、cc_completion.py（映射表合并）、app/engine/routes/chat_rest.py（种子键白名单）

• Test: tests/test_tenant_config_wiring.py

• Step 1: 写失败测试

# tests/test_tenant_config_wiring.py
from types import SimpleNamespace
from app.platform.tenant_config import TenantConfig

def _cfg(over=None):
    snap = {"email": {"smtp": {"host": "s", "port": 465, "user": "u", "pass": "p"},
                       "sender_name": "【T】", "notify_emails": ["a@b.c"]},
            "push": {"enabled": True, "url": "http://x", "api_key": "k",
                      "prefix": "t1", "timeout": 5, "max_attempts": 1,
                      "retry_backoff_base": 1.0}}
    snap.update(over or {})
    return TenantConfig.from_snapshot("t1", "T", ["t1.local"], snap)

def test_email_reads_tenant_smtp(monkeypatch):
    import app.engine.config as cfg_mod
    monkeypatch.setattr(cfg_mod, "settings",
        SimpleNamespace(SMTP_HOST="WRONG", SMTP_PORT=1, SMTP_USER="WRONG",
                        SMTP_PASS="WRONG", NOTIFY_EMAIL="wrong@x"))
    resolved = resolve_email_settings(_cfg())          # 期望读租户段而非全局
    assert resolved.notify_emails == ["a@b.c"] and resolved.sender_name == "【T】"

def test_push_reads_tenant_section():
    p = resolve_push_settings(_cfg())
    assert p.url == "http://x" and p.prefix == "t1"

def test_rate_limit_key_includes_tenant():
    from app.engine.services.rate_limit import RateLimiter
    lim = RateLimiter()
    assert lim.check("t1", "1.1.1.1", "dev1", limit=2) is True
    assert lim.check("t1", "1.1.1.1", "dev1", limit=2) is True
    assert lim.check("t2", "1.1.1.1", "dev1", limit=2) is True   # 换租户不受影响

def test_ai_settings_override_and_fallback(monkeypatch):
    import app.engine.config as cfg_mod
    monkeypatch.setattr(cfg_mod, "settings",
        SimpleNamespace(ZHIPU_API_KEY="ENV_KEY", ZHIPU_BASE_URL=None, AI_MODEL="glm-4-flash"))
    fb = resolve_ai_settings(_cfg())                        # 未覆盖 → 回退 env
    assert fb.api_key == "ENV_KEY" and fb.model == "glm-4-flash"
    ov = resolve_ai_settings(_cfg({"ai": {"api_key": "k2", "model": "glm-4-plus"}}))
    assert ov.api_key == "k2" and ov.model == "glm-4-plus"

def test_contact_toggles_resolve():
    flags = resolve_contact_flags(_cfg({"contact": {"cc_completion": False,
                                                    "email_light_ask": False}}))
    assert flags.cc_completion is False and flags.email_light_ask is False

def test_name_to_region_extras_merged():
    cfg = _cfg({"misc": {"name_to_region_extras": {"德意志": "DE"}}})
    from app.engine.services.cc_completion import region_for_name
    assert region_for_name("德意志", cfg) == "DE"

def test_chat_rest_seed_keys_from_config():
    keys = allowed_seed_keys(_cfg({"engine": {"context_seed_keys":
        {"vehicle_name": "app_vehicle_name", "vehicle_id": "app_vehicle_id"}}}))
    assert keys == {"vehicle_name", "vehicle_id"}

• Step 2: 跑测试确认失败

Run: python -m pytest tests/test_tenant_config_wiring.py -v Expected: FAIL（resolve_* 函数不存在 / rate_limit 不收 tenant 参数）

• Step 3: 实现接线：

  • email/push：resolve_email_settings(config) / resolve_push_settings(config)（读 config.data 对应组；email.smtp 为 None 回退 env settings）
  • AI：resolve_ai_settings(config) 供 provider 工厂按租户实例化 glm（api_key/base_url/model 未覆盖回退 env settings）
  • contact 开关：resolve_contact_flags(config) 接进 conversation 的 _cc_pre_turn 门控与邮箱轻补分支
  • 映射表：cc_completion 查表函数加 extras 参数（NAME_TO_REGION ⊕ misc.name_to_region_extras）
  • 种子键：chat_rest 白名单改读 config.engine.context_seed_keys 的键集合
  • rate_limit：桶 key 改 (tenant_slug, ip, device)、limit 读 config.engine.rate_limit_per_hour

• Step 4: 既有 email/push 测试回归：monkeypatch 点从 settings 迁到 resolve_*_settings 返回值（或直接传 config 夹具），断言不变

• Step 5: 全量回归 + 提交

git add -A && git commit -m "feat: 服务层配置接线——email/push/rate_limit/AI凭据/行为开关/映射表/种子键"

---

Task 10: /api/public/brand + 前端租户化

Files:

• Create: app/engine/routes/public.py、seed_profiles/jiancai/assets/kefu-small.png（自 jc/static 拷）、seed_profiles/auto/assets/kefu-small.png（自 ai-kefu/static 拷——两项目头像不同，各归各）

• Modify: static/widget.js、static/js/chat.js、static/chat.html（从 jc 拷入后改）

• Test: tests/test_brand_endpoint.py、tests/test_widget_namespacing.py

• Step 1: 失败测试

BRAND_EXPECTED = {"slug", "title", "color", "avatar", "welcome", "topics",
                  "placeholder", "whatsapp", "email", "privacy_url"}

def test_brand_whitelist_only(app_client):
    r = app_client.get("/api/public/brand", headers={"host": "chat2.agent.hnxinyouhe.com"})
    assert r.status_code == 200
    assert set(r.json().keys()) == BRAND_EXPECTED      # 白名单，绝不多吐
    assert "api_key" not in r.text and "smtp" not in r.text.lower()
    assert r.headers["access-control-allow-origin"] == "*"

def test_brand_site_overlay(app_client):
    h = {"host": "chat2.agent.hnxinyouhe.com"}
    r = app_client.get("/api/public/brand?site=es", headers=h)
    assert r.json()["title"] == "HESUPIPE Soporte"      # 站点覆盖生效
    r2 = app_client.get("/api/public/brand", headers=h)
    assert r2.json()["title"] == "HESUPIPE Support"     # 不带 site = 租户默认门面
    r3 = app_client.get("/api/public/brand?site=nope", headers=h)
    assert r3.json()["title"] == "HESUPIPE Support"     # 未知 site 回退

def test_avatar_served_per_tenant(app_client):
    r = app_client.get("/api/public/avatar", headers={"host": "chat.agent.hnxinyouhe.com"})
    assert r.status_code == 200 and r.headers["content-type"].startswith("image/")

# tests/test_widget_namespacing.py —— 静态断言（仿 test_theme_static 风格）
def test_widget_keys_namespaced_by_slug():
    src = pathlib.Path("static/widget.js").read_text(encoding="utf-8")
    assert "cc-widget-host-${" in src or "cc-widget-host-" + "${" in src
    assert "widget_invite_dismissed_${" in src

• Step 2: 实现 public.py + 前端改造

  • /api/public/brand?site=<slug>：站点覆盖层 ⊕ 租户默认门面，显式字段白名单序列化 + CORS *；未带/未知 site 回退租户默认
  • /api/public/avatar：按当前租户解析 seed_profiles/<slug>/assets/<brand.avatar 文件名>，缺省回退 static/kefu-small.png（Phase 2 起改读 uploads 卷）
  • widget.js：读自身 data-site 属性，fetch brand 时带 ?site=；所有默认值改读 brand；host id / sessionStorage / localStorage key 拼租户 slug（同租户多站点靠页面 path 天然隔离）；fetch 失败退 data-*/中性默认；WS init 消息带 site，服务端记入 client_info（会话打站点标，后台可按站筛线索）
  • chat.html/chat.js 同法（fetch 统一）

• Step 3: 手动冒烟：uvicorn app.main:app --port 8018 起服务，改本机 hosts 把两个测试域名指 127.0.0.1，浏览器开 static/widget-test.html 验证两租户品牌与同页共存。

• Step 4: 全量回归 + 提交 git commit -m "feat: brand 白名单下发 + widget/chat 客户端 key 租户化"

---

Task 11: 上游吸收

Files:

• Modify: static/widget.js（placeholder 可配置——对齐 ai-kefu 的 input-placeholder 实现，读 brand.placeholder + data-placeholder 覆盖）

• Test: 从 D:\refactor\ai-kefu\tests\ 拷入 test_channel_hygiene.py、test_chat_rest_ip.py、test_lead_push_cc.py、test_cc_integration.py、test_prompt_static.py、test_ip_service_region.py、test_placeholder_static.py → 修 import 跑绿（行为以 jc 实现为准，断言不适处改断言不改实现）

• Step 1: 拷 7 个测试文件、修 import、跑

• Step 2: 红→绿循环：失败断言逐条判断是"上游测的旧行为"（改断言）还是"jc 缺的修复"（补实现，注明同步自上游）

• Step 3: 提交 git commit -m "feat: 吸收上游测试断言与 placeholder 配置化"

---

Task 12: Phase 1 收尾

• Step 1: 全量回归 python -m pytest -q → 全绿，记录用例数（基线 315 + 新增 ~20）
• Step 2: 零同步 IO 审计：grep -rn "requests\.\|urllib\|time\.sleep" app/ 确认 async 路径无同步 IO（已核实：lead_push_service._do_post 为 httpx.AsyncClient、providers/glm.py 流式走 executor）；审计结论写入 CLAUDE.md
• Step 3: spec §6.2 合并清单逐项打勾核对（jc 六项取入/上游吸收/三不带：blog.html、Redis、旧版 whatsapp-cc）
• Step 4: 更新 CLAUDE.md（Phase 1 完成状态、本地起服务方法、测试基线数）
• Step 5: 提交 git commit -m "chore: Phase 1 收尾——引擎合并租户化完成"

---

显式不做（Phase 1 范围外）

platform.db/版本快照/admin 认证/试金石（Phase 2）；art-design-pro 后台（Phase 3）；Dockerfile/部署/迁移脚本（Phase 4，本阶段纯本地代码+测试）。