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）**

```gitignore
__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**

```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: 提交**

```bash
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   （暂留，Task 3 拆）
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（过渡版，单租户直跑）**

```python
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from app.engine.routes.chat import router as ws_router
from app.engine.routes.chat_rest import router as rest_router

app = FastAPI(title="all-kefu (phase1 transitional)")
app.include_router(ws_router)
app.include_router(rest_router)
app.mount("/static", StaticFiles(directory="static"), name="static")
```

注意：jc 的 main.py 里有 lifespan 建表/迁移/sweep 逻辑——本步先不做（Task 5 由 Registry 逐租户做）。若 tests/conftest.py 依赖 jc 的 app 工厂，先按其原样改造指向新 main。

- [ ] **Step 4: 跑全量测试建立基线**

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

- [ ] **Step 5: 提交**

```bash
git add -A && git commit -m "feat: jc 引擎主线搬入 app/engine（320 测试基线全绿）"
```

---

### 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: 写失败测试**

```python
# 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**

```python
# 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
```

```python
# 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: 提交**

```bash
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 1: 提取 jiancai 词表**。逐常量从 `app/engine/services/boundary_filter.py`（PRICE_PATTERNS、DEFLECTIONS_EN/ZH/ES、NO_CONTACT_*）和 `app/engine/services/conversation.py`（BOUNDARY_TRIGGER_RE、BUDGET_TRIGGER_RE、CONTACT_ASK_TRIGGER_RE 的模式串）抄进 JSON：

```json
{
  "price_patterns": ["<PRICE_PATTERNS 每条 pattern 原样>"],
  "deflections": {"en": ["..."], "zh": ["..."], "es": ["..."]},
  "no_contact": {"replacement": "<NO_CONTACT_RESPONSE 原文>", "asked_replacement": "<...>"},
  "triggers": {"boundary": ["..."], "budget": ["..."], "contact_ask": ["..."]},
  "anti_kill": ["DN", "PN", "SDR", "PE100", "SCH", "ISO4427", "20ft", "km", "<jc 防误杀清单>"],
  "cc_templates": {"en": ["<回显确认模板 EN>"], "zh": ["<ZH>"], "es": ["<ES>"]}
}
```

`seed_profiles/auto/wall.json`：同一结构，price_patterns 用上游 ai-kefu 的中英+$ 词表、anti_kill 换汽车清单（km/年款等）、deflections 去掉 es。

- [ ] **Step 2: 写编译器（先写失败测试）**

```python
# 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: 实现编译器**

```python
# 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
    no_contact: dict
    triggers: dict      # 每值 tuple[re.Pattern]
    anti_kill: tuple
    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()},
        no_contact=dict(cfg.get("no_contact", {})),
        triggers=triggers,
        anti_kill=tuple(cfg.get("anti_kill", [])),
        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` 取（jc 原文进 jiancai 词表 JSON 的对应字段）。

- [ ] **Step 7: 跑全量回归**

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

- [ ] **Step 8: 提交**

```bash
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**

```json
// seed_profiles/jiancai/tenant.json
{ "slug": "jiancai", "name": "HESUPIPE 建材", "status": "active",
  "domains": ["chat2.agent.hnxinyouhe.com", "localhost.jc"],
  "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"}} } }
```

auto 同理（domains: chat.agent.hnxinyouhe.com，prefix ai-kefu，context_seed_keys 用 vehicle_*，品牌橙 #f08a31）。`config` 内**不含** prompts（prompts 由目录内 md 提供，见 Task 6）与 wall（同目录 wall.json）。

- [ ] **Step 2: 写失败测试**

```python
# 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
```

- [ ] **Step 3: 跑测试确认失败**

- [ ] **Step 4: 实现 registry.py**

```python
# 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

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: 提交**

```bash
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: 写失败测试**

```python
# 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` → 全绿（提示词文案断言测试若断言旧枚举文本，改为断言注入结果）。

```bash
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`
- Test: `tests/test_host_routing.py`

- [ ] **Step 1: 写失败测试**

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

def test_known_host_routes_to_tenant(app_client):
    r = app_client.get("/api/public/brand-not-exist-yet")  # 先用 /health 或 404 路由验证不 500
    assert r.status_code in (200, 404)

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）：

```python
# 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 = self.registry.resolve(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"] = rt
        await self.app(scope, receive, send)
```

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

- [ ] **Step 4: 全量回归 + 提交**

```bash
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: 失败测试**（纯单元）：

```python
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: 推送异步化（httpx）

**Files:**
- Modify: `app/engine/services/lead_push_service.py`
- Test: `tests/test_lead_push_async.py`（新建）+ 既有 push 测试回归

- [ ] **Step 1: 写失败测试**（AsyncMock httpx）：

```python
import httpx
from unittest.mock import AsyncMock, MagicMock

async def test_push_uses_httpx_async(monkeypatch):
    resp = MagicMock(status_code=200)
    resp.json = lambda: {"results": [{"external_key": "k", "status": "accepted"}]}
    client = MagicMock()
    client.post = AsyncMock(return_value=resp)
    monkeypatch.setattr("app.engine.services.lead_push_service._open_client",
                        lambda: _Ctx(client))
    ...  # dispatch 一条 pending → 断言 client.post awaited 一次、url/headers 正确
```

- [ ] **Step 2: 改造**：同步 requests/urllib 调用 → `async with httpx.AsyncClient(timeout=cfg.timeout) as client: await client.post(url, json=payload, headers=...)`；重试矩阵逻辑不变（仍只重试 5xx/传输错误）。**事件循环零同步 IO** 纪律的落点（spec §6.1/§11）。

- [ ] **Step 3: 既有 test_lead_push_* 回归**（原 mock 点从 sync 库改挂到新 _open_client/AsyncClient 工厂）

- [ ] **Step 4: 提交** `git commit -m "feat: 推送改 httpx.AsyncClient（零同步 IO）"`

---

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

**Files:**
- Create: `app/engine/routes/public.py`
- 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: 失败测试**

```python
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"] == "*"
```

```python
# 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**（显式字段序列化 + CORS 头）+ widget.js 改造：启动 fetch `/api/public/brand` → 存 `__brand`，所有默认值改读 brand；host id / storage key / localStorage 会话 key 全部拼 slug；fetch 失败退 data-*/中性默认。chat.html/chat.js 同法（服务端也可直接渲染 brand 进页面，二选一，倾向 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 → 修 import 跑绿（行为以 jc 实现为准，断言不适处改断言不改实现）

- [ ] **Step 1: 拷 6 个测试文件、修 import、跑**
- [ ] **Step 2: 红→绿循环**：失败断言逐条判断是"上游测的旧行为"（改断言）还是"jc 缺的修复"（补实现，注明同步自上游）
- [ ] **Step 3: 提交** `git commit -m "feat: 吸收上游测试断言与 placeholder 配置化"`

---

### Task 12: Phase 1 收尾

- [ ] **Step 1: 全量回归** `python -m pytest -q` → 全绿，记录用例数（目标 ≥ 320 + 新增 ~20）
- [ ] **Step 2: spec §6.2 合并清单逐项打勾核对**（jc 六项取入/上游吸收/三不带：blog.html、Redis、旧版 whatsapp-cc）
- [ ] **Step 3: 更新 CLAUDE.md**（Phase 1 完成状态、本地起服务方法、测试基线数）
- [ ] **Step 4: 提交** `git commit -m "chore: Phase 1 收尾——引擎合并租户化完成"`

---

## 显式不做（Phase 1 范围外）

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