# 单域名多租户(路径前缀路由)+ 上线收口项 实施计划 > **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:** 按 spec(docs/superpowers/specs/2026-10-19-shared-domain-routing-design.md)实现 `/t//` 路径前缀路由(与 Host 模式共存)、静态面前缀自推导、嵌入代码双形态,并收口登录限流 XFF 末段与 limiter 周期清扫。 **Architecture:** 纯 ASGI 中间件层加二级解析(路径前缀优先于 Host,剥离后下游路由零改动);widget.js/chat.html 从自身地址推导基础路径(Host 模式空串零变化);registry 增 slug 解析公开方法;登录限流 IP 取 XFF 末段(反代追加语义,不可伪造);lifespan 注册 300s 限流器清扫任务。 **Tech Stack:** FastAPI/Starlette 纯 ASGI、原生 JS(widget/chat 静态面,无构建)、Vue3+TS admin、pytest(PYTHONUTF8=1)/vitest。 **执行环境注意(Windows):** - 仓库根 `D:\refactor\all-kefu`;后端测试 `PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_xxx.py -q`(全量 `PYTHONUTF8=1 .venv/Scripts/python -m pytest -q`,基线 **632+2skip**) - 前端:`cd admin && pnpm test`(基线 **182**);`pnpm build`;dts 行尾纪律照旧(`git diff --ignore-cr-at-eol` 为空才 checkout 还原) - 分支:`feature/shared-domain`(从 main 切出) - 本地路径模式验证技巧:路径前缀优先于 Host——`curl http://127.0.0.1:8018/t/jiancai/`(Host 是 IP 未知也没关系,路径解析兜住) - 单进程铁律、日志/内存上限铁律(本计划 Task 7 为其进程内面)不变 --- ## Batch A:路由核心(后端) ### Task 1: registry.resolve_by_slug(TDD) **Files:** - Modify: `app/platform/registry.py`(顶部正则常量 + 解析区新增方法) - Test: `tests/test_registry.py`(追加) - [ ] **Step 1: 写失败测试**——test_registry.py 追加(沿用该文件既有夹具的建租户方式;看文件内现有用例怎么建 LoadedTenant/装配 registry,照抄同款): ```python class TestResolveBySlug: """路径前缀模式解析(spec 2026-10-19 §5)。""" async def test_已知_slug_返回租户与首站点(self, ...既有夹具...): rt, site = await registry.resolve_by_slug("jiancai") assert rt.config.slug == "jiancai" assert site.slug == rt.config.sites[0].slug # 首站点为默认 async def test_未知_slug_UnknownTenantError(self, ...): with pytest.raises(UnknownTenantError): await registry.resolve_by_slug("no-such") async def test_停用租户_TenantDisabledError(self, ...): # 夹具建 active 租户后置 status != "active" 再 resolve ... async def test_空站点表按未知处理(self, ...): # 夹具建 sites=[] 的租户 → UnknownTenantError ... ``` - [ ] **Step 2: 跑测试确认失败**(方法不存在) - [ ] **Step 3: 实现**——registry.py: 顶部(import re 后): ```python # 租户 slug 形态(创建侧 tenant_routes 原 SLUG_RE 的单一事实源——路径前缀 # 解析与创建校验共用,永不漂移;spec 2026-10-19 §3.1) TENANT_SLUG_RE = re.compile(r"^[a-z0-9-]+$") ``` 解析区(`resolve_site` 之后): ```python async def resolve_by_slug(self, slug: str) -> tuple[TenantRuntime, Site]: """路径前缀模式(/t//,spec 2026-10-19 §5):slug → (runtime, 首站点)。语义与 resolve_site 同款:未知 slug / 空站点表 → UnknownTenantError;停用 → TenantDisabledError;懒建 engine、 reload 竞态取 _ensure_ready 返回值(与 resolve_site 同理)。""" rt = self._by_slug.get(slug) if rt is None: raise UnknownTenantError(f"未知租户 slug: {slug!r}") if rt.config.status != "active": raise TenantDisabledError(f"租户已停用: {slug!r}") rt = await self._ensure_ready(rt) if not rt.config.sites: # 创建向导恒建 main 站点,此分支为防御(spec §5) raise UnknownTenantError( f"租户 {slug!r} 无站点——路径模式无从选站点") return rt, rt.config.sites[0] ``` - [ ] **Step 4: 全文件回归**——`PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_registry.py -q` 全绿 - [ ] **Step 5: Commit**——`feat(registry): resolve_by_slug 路径模式解析 + TENANT_SLUG_RE 常量上提` ### Task 2: 中间件路径前缀解析(TDD) **Files:** - Modify: `app/platform/host_middleware.py`、`app/platform/routes/tenant_routes.py`(SLUG_RE 改引 registry) - Test: `tests/test_host_routing.py`(追加;该文件既有夹具/Client 用法照抄) - [ ] **Step 1: tenant_routes 的 SLUG_RE 换源**——`SLUG_RE = re.compile(r"^[a-z0-9-]+$")`(:65)改为 `from app.platform.registry import TENANT_SLUG_RE as SLUG_RE`(删本地定义;其余用法不动;grep `SLUG_RE` 确认只有本文件用) - [ ] **Step 2: 写失败测试**——test_host_routing.py 追加(照既有用例的 app/Client 夹具;Host 一律用未知域名验证"路径优先于 Host"): ```python class TestPathPrefixRouting: """路径前缀模式(spec 2026-10-19 §3)。Host 用未知值——路径命中证明优先级。""" def test_聊天页_t_slug_根路径(self, client): r = client.get("/t/jiancai/", headers={"Host": "unknown.example"}) assert r.status_code == 200 assert "chat" in r.text.lower() def test_静态与公开API全链(self, client): assert client.get("/t/jiancai/static/widget.js", headers={"Host": "unknown.example"}).status_code == 200 r = client.get("/t/jiancai/api/public/brand", headers={"Host": "unknown.example"}) assert r.status_code == 200 and r.json()["slug"] == "jiancai" def test_orig_path留痕(self, client): # 借 brand 端点无法直接看 state——用响应正常 + 行为断言替代: # state 写入由后续 WS 用例与全链用例间接钉住;此处断言剥离后路由可达即证改写成立 ... def test_未知slug_404(self, client): assert client.get("/t/no-such/x", headers={"Host": "unknown.example"}).status_code == 404 def test_停用租户_503(self, client): ... # 既有停用夹具同款 def test_无尾斜杠不剥离_走Host解析404(self, client): assert client.get("/t/jiancai", headers={"Host": "unknown.example"}).status_code == 404 assert client.get("/t/jiancai", headers={"Host": "chat2.agent.hnxinyouhe.com"}).status_code == 200 # 已知 Host 正常(Host 模式根路径) def test_编码slug拒绝(self, client): # %61uto 解码即 auto——raw 直判必须拒(spec §3.1) assert client.get("/t/%61uto/", headers={"Host": "unknown.example"}).status_code == 404 def test_大写与点号拒绝(self, client): assert client.get("/t/Jiancai/", headers={"Host": "unknown.example"}).status_code == 404 assert client.get("/t/../api/platform/auth/login", headers={"Host": "unknown.example"}).status_code == 404 def test_平台豁免优先级不受影响(self, client): assert client.get("/health", headers={"Host": "unknown.example"}).status_code == 200 # /t/x/api/platform/... 剥离后命中平台路由(by design,认证守护) assert client.post("/t/jiancai/api/platform/auth/login", headers={"Host": "unknown.example"}).status_code in (401, 422, 429) def test_WS路径模式握手(self, client): with client.websocket_connect("/t/jiancai/ws/chat", headers={"Host": "unknown.example"}) as ws: data = ws.receive_json() assert "session_id" in data def test_WS未知slug拒绝(self, client): with pytest.raises(Exception): with client.websocket_connect("/t/no-such/ws/chat", headers={"Host": "unknown.example"}): pass def test_Host模式回归(self, client): # 既有用例全绿即回归;此处补一条:已知 Host + /t/其他租户/ → 路径优先(spec 优先级) r = client.get("/t/jiancai/api/public/brand", headers={"Host": "chat2.agent.hnxinyouhe.com"}) assert r.status_code == 200 and r.json()["slug"] == "jiancai" ``` - [ ] **Step 3: 跑测试确认失败** - [ ] **Step 4: 实现中间件**——host_middleware.py: 模块顶部 import 增加 `import urllib.parse`、`from app.platform.registry import TENANT_SLUG_RE`(并入既有 registry import 块);类内新增: ```python # 路径前缀形态(spec 2026-10-19 §3):raw_path 直判——slug 段必须 # 未编码(%61uto 解码后即 auto,解码形态判不出编码;raw bytes 上 # [a-z0-9-] 天然排除 % 与大写)。仅带尾斜杠形态剥离(/t/x 不剥)。 _PREFIX_RE = re.compile(rb"^/t/([a-z0-9-]+)(/.*)?$", re.ASCII) ``` `__call__` 内,平台豁免 return 之后、`registry = ...` 之前插入: ```python # 路径前缀模式(spec 2026-10-19 §3):/t// 前缀优先于 Host 解析。 raw = scope.get("raw_path") or b"" m = self._PREFIX_RE.match(raw) if m and m.group(2) is not None: # group(2) None = /t/x 无尾斜杠 → 不剥离 slug = m.group(1).decode("ascii") registry = getattr(scope["app"].state, "registry", None) if registry is None: raise RuntimeError("app.state.registry 未挂载(见模块 docstring)") try: runtime, site = await registry.resolve_by_slug(slug) except UnknownTenantError: if scope["type"] == "websocket": await _close_ws(receive, send, 1008, "unknown tenant") else: await PlainTextResponse("Not Found", status_code=404)(scope, receive, send) return except TenantDisabledError: if scope["type"] == "websocket": await _close_ws(receive, send, 1013, "service paused") else: await HTMLResponse(PAUSE_PAGE, status_code=503)(scope, receive, send) return state = scope.setdefault("state", {}) state["orig_path"] = scope.get("path") or "" # 排障留痕(§3.3) # 剥离量按解码 path 对应段计:slug 段在 raw 上未编码 → 解码/原始 # 同长,前缀 len("/t/"+slug+"/") 在两形态一致 scope["path"] = scope["path"][len("/t/") + len(slug) + 1:] or "/" scope["raw_path"] = urllib.parse.quote( scope["path"], safe="/:@!$&'()*+,;=-._~").encode("ascii") state["tenant"] = runtime state["site"] = site await self.app(scope, receive, send) return ``` (其后原有 Host 解析逻辑不动;既有 `registry = getattr(...)` 行保留共用——路径分支提前取过一次,注意变量复用别重复声明报错,实现时合并为一次取值放豁免之后、两个分支之前。) - [ ] **Step 5: 跑 Task 2 测试 + 全文件 + test_host_routing 既有用例全绿** - [ ] **Step 6: Commit**——`feat(middleware): /t// 路径前缀解析(与 Host 模式共存)` ### Task 3: 全链集成测试(chat REST + 限流隔离) **Files:** - Test: `tests/test_host_routing.py`(TestPathPrefixRouting 内追加) - [ ] **Step 1: 追加用例**(照抄既有 chat REST 用例的 payload/头形态,test_channel_hygiene / test_chat_rest_ip 里有样板): ```python def test_chat_REST_路径模式建会话与消息(self, client): # POST /t//api/chat/session → 200 会话;同前缀 GET messages 通路 ... def test_限流隔离_同IP双租户独立计数(self, client, monkeypatch): # 同 client(同 IP)分别打 /t/a/... 与 /t/b/... 的会话接口至各自阈值, # 断言 a 触发 429 时 b 仍可用(键含 tenant——spec §11.2) ... ``` - [ ] **Step 2: 跑绿**(实现已在 Task 2 完成,此为验收钉位;若红按信号修中间件) - [ ] **Step 3: Commit**——`test: 路径模式全链与限流隔离钉位` ## Batch B:静态面前缀自推导 ### Task 4: widget.js basePath(先写静态守卫测试) **Files:** - Modify: `static/widget.js`(:26 附近加 basePath;:132/:164-166/:683-685 四处端点构造加前缀) - Create: `tests/test_widget_static.py`(照 test_placeholder_static.py 的读文件断言风格) - [ ] **Step 1: 写失败测试**: ```python """widget.js 静态守卫(spec 2026-10-19 §4):脚本地址自推导 basePath, 三类端点构造(brand/avatar/WS)统一带前缀;Host 模式前缀为空串零变化。""" def test_有_basepath_推导(): src = read("static/widget.js") assert "indexOf(\"/static/widget.js\")" in src # 从 script.src 剥尾推导 def test_端点构造全部走_basePath(): src = read("static/widget.js") # 不允许再出现裸 wsBase + "/api/ 或 wsHost + "/ws/(必须经 basePath) assert 'wsBase + "/api/' not in src assert 'wsHost + "/ws/' not in src assert 'wsBase + basePath' in src def test_默认头像路径带前缀(): src = read("static/widget.js") assert 'basePath + DEFAULT_AVATAR_PATH' in src or "basePath + '" in src ``` - [ ] **Step 2: RED 确认** → **Step 3: 实现**——widget.js :26 后加: ```js // spec 2026-10-19 §4: 脚本地址自携带租户路径(/t//static/widget.js) // → 剥掉 /static/widget.js 起的前缀得 basePath("" = Host 模式零变化)。 // script.src 永远指向本文件,indexOf 命中即安全;query(?v=)不影响 pathname。 var basePath = (function () { var p = new URL(script.src).pathname; var i = p.indexOf("/static/widget.js"); return i > 0 ? p.slice(0, i) : ""; })(); ``` :132 brand → `wsBase + basePath + "/api/public/brand"`;:163-166 avatar → `wsBase + basePath + "/" + avatarAttr...` / `wsBase + basePath + (b.avatar...)` / `wsBase + basePath + DEFAULT_AVATAR_PATH`;:683-685 WS → `wWs = new WebSocket(wsHost + basePath + "/ws/chat")`。 - [ ] **Step 4: GREEN + 全量静态测试**(`PYTHONUTF8=1 .venv/Scripts/python -m pytest tests/test_widget_static.py tests/test_placeholder_static.py tests/test_theme_static.py tests/test_invite_static.py -q`——后三个是回归,widget.js 改动不能碰主题块) - [ ] **Step 5: Commit**——`feat(widget): 脚本地址自推导 basePath(路径模式端点前缀)` ### Task 5: chat.html 引导脚本 + chat.js __BASE_PATH__ **Files:** - Modify: `static/chat.html`(head 顶部内联引导;删静态 link/script 标签改为动态装载;avatar img 去 src 改 DOMContentLoaded 设置) - Modify: `static/js/chat.js`(:5 brand fetch、:132 WS、文件内其余 `/api/`、`/static/`、`/ws/` 绝对引用统一加 `(window.__BASE_PATH__ || "")` 前缀——动手前 grep 全列) - Create: `tests/test_chat_static.py` - [ ] **Step 1: 写失败测试**: ```python """chat 静态面前缀守卫(spec 2026-10-19 §4)。""" def test_引导脚本存在且定义_BASE_PATH(): src = read("static/chat.html") assert "__BASE_PATH__" in src assert "/t/" in src # 路径形态推导正则 def test_不再有静态绝对引用标签(): src = read("static/chat.html") assert 'href="/static/' not in src assert 'src="/static/js/' not in src assert 'src="/api/public/avatar"' not in src def test_chatjs_端点全部走_BASE_PATH(): src = read("static/js/chat.js") assert '__BASE_PATH__' in src # 裸绝对端点不允许(fetch("/api/... 或 + "/ws/ 直接拼接形态) assert 'fetch("/api/' not in src ``` - [ ] **Step 2: RED** → **Step 3: 实现**——chat.html `` 内 charset/title 之后、原 link/script 位置替换为: ```html ``` (body 末尾原 ` * shared 形态 = */ export type EmbedForm = 'domain' | 'shared' export function buildEmbedCode(form: EmbedForm, opts: { domain: string; sharedDomain: string | null; tenantSlug: string; site: string | null }): string // 不可用(缺域名/共享域名)返回 '' ``` vitest:normalize 各形态、双形态代码串、缺料空串、data-site 可选。 - [ ] **Step 2: RED → 实现 model.ts → GREEN** - [ ] **Step 3: 后端端点**(小:读 settings_store 快照返回;测试加在既有 settings 测试文件——两种角色 200 / 未认证 401) - [ ] **Step 4: index.vue 嵌入卡改造**——加载数据时请求 widget-domain;形态 radio(域名形态/共享域名形态——按可用性禁用);embedCode computed 换用 buildEmbedCode;无域名租户仅共享形态;都无显示引导文案(去平台设置配共享域名)。设置页推荐键渲染无需改(表驱动) - [ ] **Step 5: `pnpm test` + `pnpm build` 全绿;dts 纪律** - [ ] **Step 6: Commit**——`feat(admin): 嵌入代码双形态 + shared_widget_domain 平台键` ### Task 9: 创建向导域名可选 **Files:** - Modify: `admin/src/views/tenants/list/index.vue`(向导步骤②「绑定域名」改为可跳过——校验放宽、步骤文案「可留空 = 使用共享域名路径模式」;创建摘要显示「未绑定域名(路径模式)」) - Test: 若向导校验逻辑在 model.ts(查 `admin/src/views/tenants/list/` 有无 model.ts/校验纯函数)则 vitest 追加「空域名列表合法」用例;纯模板改动则手动走查记录 - [ ] **Step 1: 定位向导校验**(grep 域名/必填/required)→ 放宽 + 文案 - [ ] **Step 2: 走查**——dev server 建一个零域名租户 → 详情正常 → 配置中心品牌页出共享形态嵌入代码(依赖 Task 8 的 shared_widget_domain 已配) - [ ] **Step 3: `pnpm test` + build → Commit**——`feat(admin): 创建向导域名可跳过(纯路径模式租户)` ### Task 10: 部署工件 **Files:** - Modify: `deploy/openresty-all-kefu.conf.template`(增共享域名 server 块:同 upstream、WS 升级头单套、超时 3600s;XFF 头已有 `$proxy_add_x_forwarded_for` 确认覆盖新块并注记「登录限流取末段依赖此语义,勿改成 $remote_addr 覆盖式」) - Modify: `docs/deploy-production.md`(增「共享域名模式」一节:`/t//` 地址形态、无尾斜杠 404 语义、shared_widget_domain 设置、新租户零 DNS 接入流程、嵌入代码形态示例) - [ ] **Step 1: 模板增块 + 手册增节** → **Step 2: Commit**——`docs(deploy): 共享域名模式部署工件` ### Task 11: 全局回归 + 走查收口 - [ ] **Step 1: 后端全量**——`PYTHONUTF8=1 .venv/Scripts/python -m pytest -q`(≥ 基线 632+2skip + 新增) - [ ] **Step 2: 前端全量 + 构建**——`cd admin && pnpm test && pnpm build`(≥182 + 新增;dts 纪律) - [ ] **Step 3: 双模式手动走查**——本地 uvicorn:Host 模式(hosts 指向 chat2...)聊天页/挂件原样;路径模式 `curl http://127.0.0.1:8018/t/jiancai/` + 浏览器开 `/t/jiancai/` 全链(页面样式/WS 对话/头像/品牌色);嵌入代码两形态各贴到 widget-test.html 验证 - [ ] **Step 4: 铁律检查项**——compose `mem_limit`/日志封顶在位;清扫任务在 lifespan 启停(Task 7 测试已钉) - [ ] **Step 5: 收尾**——工作区干净、临时产物清空、`git commit`(如有杂项) --- ## 依赖与并行性 - Task 1 → 2 → 3 顺序;Task 4/5 依赖 2(联调)但代码可先行(测试独立)——建议仍顺序 - Task 6/7 独立于 A/B(可穿插);Task 8 依赖 1-5 完成(嵌入代码端到端验证);Task 9 依赖 8;Task 10 独立;Task 11 收口 - 分支:`feature/shared-domain`