fix: 词表墙的四项审查修复（es 池降级/占位符白名单校验/键 allowlist/提取自检+防呆测试）

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

app/engine/services/boundary_filter.py

@@ -101,7 +101,11 @@ def pivot_for(user_msg: str, has_contact: bool = False,
               wall: CompiledWall | None = None) -> str:
     w = wall or DEFAULT_WALL
     lang = _lang_pool(user_msg)
-    pool = (w.pivots_contact if has_contact else w.pivots)[lang]
+    pools = w.pivots_contact if has_contact else w.pivots
+    # 缺池语言优雅降级(审查 I-1): 租户墙可无 es 池(auto 中英+$ 词表),西语
+    # 消息按重音符路由 es 后回落 EN 池——与 _lang_pool 注释的罗曼语族降级
+    # 取舍同源(宁可降级不可 KeyError 炸会话流)。
+    pool = pools.get(lang) or pools["en"]
     return pool[0]
 
 
@@ -110,7 +114,9 @@ def deflection_for(user_msg: str, rotate: int = 0,
                    wall: CompiledWall | None = None) -> str:
     w = wall or DEFAULT_WALL
     lang = _lang_pool(user_msg)
-    pool = (w.deflections_contact if has_contact else w.deflections)[lang]
+    pools = w.deflections_contact if has_contact else w.deflections
+    # 缺池语言优雅降级(审查 I-1): 同 pivot_for,回落 EN 池
+    pool = pools.get(lang) or pools["en"]
     return pool[rotate % len(pool)]
 
 

app/engine/wall_schema.py

@@ -3,12 +3,15 @@
 seed_profiles/<slug>/wall.json(由 scripts/extract_config.py 从老项目源码
 AST 提取生成)是租户墙词表/话术的唯一数据源。compile_wall 把 JSON 编译为
 运行时对象: pattern 字符串 → re.Pattern,模板话术(含 {shown}/{intl} 等
-占位符)原样承载、由 conversation.py 运行期 .format() 回填。
+占位符)原样承载、由 conversation.py 运行期 .format() 回填;占位符仅允许
+TEMPLATE_PLACEHOLDERS 白名单键,编译期干跑校验(字面花括号/未知占位符
+显性失败,不留到运行期炸会话流)。
 
 regex 条目两种形态(pattern 文本与源码逐字一致):
 - 纯字符串: 无标志位,re.compile(s)
 - {"pattern": "...", "flags": "i"}: flags 字符映射 re.IGNORECASE
-  (提取脚本只产出 "i";comment 键为设计说明注释,编译时忽略)
+  (提取脚本只产出 "i";comment 键为设计说明注释,编译时忽略;dict 条目
+  出白名单 {pattern, flags, comment} 之外的键 → 显性拒绝,防键拼错静默放宽)
 
 结构/regex 无效 → WallConfigError,信息含字段路径(坏 regex 含 "regex"
 字样与 re.error 原因)——配置错误在编译期显性失败,不流入运行期。
@@ -69,6 +72,20 @@ _KNOWN_FIELDS = frozenset(
                              "reask_patterns", "triggers", "reminders",
                              "cc_templates"])
 
+# regex 条目 dict 的合法键(审查 I-3): 提取脚本只产出 {pattern, flags,
+# comment};键拼错(如 "flagss")不再静默放宽标志位,而是编译期显性拒绝。
+_LEGAL_PATTERN_KEYS = frozenset({"pattern", "flags", "comment"})
+
+# 模板占位符白名单(审查 I-2): conversation.py 运行期 .format() 回填的全部
+# 键名(grep 实证清单)。scripts/extract_config.py 反向引用此常量(单一来源),
+# 保证提取产出的占位符与引擎回填键名不漂移。字面花括号(如把中文"格式"
+# 写成 {格式})或未知占位符在编译期即拒绝——否则运行期 KeyError 直接炸
+# 客户会话流。
+TEMPLATE_PLACEHOLDERS = frozenset({
+    "shown", "intl", "cc_script",
+    "no_contact_web", "no_contact_app", "no_contact_already",
+})
+
 
 def _err(msg: str) -> WallConfigError:
     return WallConfigError(msg)
@@ -79,6 +96,11 @@ def _compile_pattern(entry, field: str) -> re.Pattern:
     if isinstance(entry, str):
         pattern, flags = entry, 0
     elif isinstance(entry, dict):
+        unknown = set(entry) - _LEGAL_PATTERN_KEYS
+        if unknown:
+            # 审查 I-3: 键拼错(如 "flagss")曾静默忽略→墙悄悄放宽,显性拒绝
+            raise _err(f"{field}: 条目含未知键 {', '.join(sorted(unknown))}"
+                       f"(合法: {', '.join(sorted(_LEGAL_PATTERN_KEYS))}): {entry!r}")
         pattern = entry.get("pattern")
         flag_chars = str(entry.get("flags", ""))
         if not isinstance(pattern, str) or not pattern:
@@ -116,8 +138,25 @@ def _compile_pool(cfg: dict, field: str) -> dict[str, tuple[str, ...]]:
     return out
 
 
-def _str_mapping(sub, label: str, required: tuple[str, ...]) -> dict[str, str]:
-    """一层 str 映射校验(no_contact 各键/reminders 各类变体/cc_templates 各组)。"""
+def _validate_template(template: str, label: str) -> None:
+    """编译期干跑 .format()(审查 I-2): 全白名单占位符代入空串,字面花括号/
+    未知占位符/位置参数问题(原属运行期 KeyError/IndexError,炸客户会话流)
+    前移为编译期 WallConfigError。纯文本模板(无占位符)天然通过。"""
+    try:
+        template.format(**{name: "" for name in TEMPLATE_PLACEHOLDERS})
+    except (KeyError, IndexError, ValueError, AttributeError,
+            TypeError) as exc:
+        raise _err(f"{label}: 模板占位符非法(仅支持 "
+                   f"{', '.join(sorted(TEMPLATE_PLACEHOLDERS))}): "
+                   f"{template!r}: {exc}") from exc
+
+
+def _str_mapping(sub, label: str, required: tuple[str, ...],
+                 *, template: bool = False) -> dict[str, str]:
+    """一层 str 映射校验(no_contact 各键/reminders 各类变体/cc_templates 各组)。
+
+    template=True 时额外干跑 .format() 校验占位符(conversation.py 运行期
+    回填的字段才需要;no_contact 为纯文本,不校验)。"""
     if not isinstance(sub, dict):
         raise _err(f"{label}: 必须为映射(缺类别或类型错误)")
     out: dict[str, str] = {}
@@ -125,6 +164,8 @@ def _str_mapping(sub, label: str, required: tuple[str, ...]) -> dict[str, str]:
         val = sub.get(key)
         if not isinstance(val, str) or not val:
             raise _err(f"{label}.{key}: 缺非空字符串")
+        if template:
+            _validate_template(val, f"{label}.{key}")
         out[key] = val
     for key in sub:
         if key not in required:
@@ -158,7 +199,7 @@ def compile_wall(cfg: dict) -> CompiledWall:
     if not isinstance(cfg["reminders"], dict):
         raise _err("reminders: 必须为映射")
     reminders = {cat: _str_mapping(cfg["reminders"].get(cat), f"reminders.{cat}",
-                                   variants)
+                                   variants, template=True)
                  for cat, variants in _REMINDER_VARIANTS.items()}
     extra = set(cfg["reminders"]) - set(_REMINDER_VARIANTS)
     if extra:
@@ -167,7 +208,8 @@ def compile_wall(cfg: dict) -> CompiledWall:
     if not isinstance(cfg["cc_templates"], dict):
         raise _err("cc_templates: 必须为映射")
     cc_templates = {group: _str_mapping(cfg["cc_templates"].get(group),
-                                        f"cc_templates.{group}", variants)
+                                        f"cc_templates.{group}", variants,
+                                        template=True)
                     for group, variants in _CC_TEMPLATE_VARIANTS.items()}
     extra = set(cfg["cc_templates"]) - set(_CC_TEMPLATE_VARIANTS)
     if extra:

scripts/extract_config.py

@@ -20,11 +20,18 @@ import html as html_mod
 import json
 import re
 import shutil
+import sys
 from bisect import bisect_right
 from io import StringIO
 from pathlib import Path
 from tokenize import COMMENT, generate_tokens
 
+# 本仓库引擎编译器(仅 stdlib 依赖,无源项目 import 链,import 安全——
+# "绝不 import 源模块"约束针对的是老项目 app.*,非本文件所在仓库)。
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from app.engine.wall_schema import (TEMPLATE_PLACEHOLDERS, WallConfigError,
+                                    compile_wall)
+
 # ---- 占位符映射(函数体内联话术的插值表达式 → wall 模板占位符) ----
 # 源码里的 f"{shown}" / 拼接 NO_CONTACT_WEB 等,在 wall.json 里以
 # {shown} / {no_contact_web} 占位,运行期由 conversation.py .format() 回填。
@@ -37,6 +44,16 @@ PLACEHOLDER_NAMES = {
 }
 PLACEHOLDER_CALLS = {"_cc_script": "cc_script"}
 
+# 单一来源反查(审查 I-2): 引擎侧白名单 wall_schema.TEMPLATE_PLACEHOLDERS
+# 是唯一权威,本脚本产出的占位符必须 ⊆ 白名单——清单漂移(此处新增映射
+# 而引擎未同步)在提取期立即显性失败,而非运行期 .format() KeyError。
+if not (set(PLACEHOLDER_NAMES.values()) | set(PLACEHOLDER_CALLS.values())
+        <= TEMPLATE_PLACEHOLDERS):
+    raise SystemExit(
+        "提取失败: 占位符超出 wall_schema.TEMPLATE_PLACEHOLDERS 白名单: "
+        + ", ".join(sorted((set(PLACEHOLDER_NAMES.values())
+                            | set(PLACEHOLDER_CALLS.values())) - TEMPLATE_PLACEHOLDERS)))
+
 
 class NotExtractable(Exception):
     """表达式超出受限求值语法(非目标常量,跳过)。"""
@@ -220,6 +237,9 @@ def attach_price_comments(src: str, assign: ast.Assign,
     """PRICE_PATTERNS 列表内的 # 注释块挂到紧随其后的条目(comment 键)。
 
     防误杀断言的设计说明随 pattern 一并数据化,编辑 wall.json 的人能看见。
+    边界注释有意丢弃(审查 M-2 备案): 首条目之前/末条目之后的注释没有
+    可挂靠的条目(bisect 落在 0/len(entries) 之外)——首条目前的通常是
+    分组标题注释,本就不属于任何单条 pattern 的设计说明,丢弃不损信息。
     """
     comments: list[tuple[int, str]] = []
     for tok in generate_tokens(StringIO(src).readline):
@@ -272,9 +292,12 @@ def build_wall(project_root: Path, slug: str) -> dict:
         return table[name]
 
     # price_patterns: 统一先建 dict 以便挂行内注释,无 flags 且无 comment 再瘦回字符串
-    price_assign = next(n for n in bf_tree.body
-                        if isinstance(n, ast.Assign)
-                        and getattr(n.targets[0], "id", "") == "PRICE_PATTERNS")
+    price_assign = next((n for n in bf_tree.body
+                         if isinstance(n, ast.Assign)
+                         and getattr(n.targets[0], "id", "") == "PRICE_PATTERNS"),
+                        None)
+    if price_assign is None:   # 防呆(审查 M-1): 源码重构挪走赋值节点时显性失败
+        raise SystemExit(f"提取失败: {slug} 源码缺 PRICE_PATTERNS 赋值节点")
     price_entries: list[dict] = []
     for r in need(bf, "PRICE_PATTERNS"):
         e: dict = {"pattern": r["pattern"]}
@@ -475,6 +498,16 @@ def write_json(path: Path, data: dict) -> None:
                     encoding="utf-8", newline="\n")
 
 
+def _self_check(wall: dict, slug: str) -> None:
+    """提取自检(审查 I-4): 落盘前跑 compile_wall,把"提取静默漏项"
+    (锚点漂移/结构缺失/坏 regex/非法占位符)收敛到提取期失败——带 slug
+    上下文显性报错,而非留到引擎加载或客户会话运行期才炸。"""
+    try:
+        compile_wall(wall)
+    except WallConfigError as exc:
+        raise SystemExit(f"提取自检失败({slug}): {exc}") from exc
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--jc", default=r"D:\refactor\ai-kefu-jc",
@@ -492,6 +525,7 @@ def main() -> None:
 
     # jiancai(主线,完整提取;多语言墙 = 引擎出厂默认)
     jc_wall = build_wall(jc_root, "jiancai")
+    _self_check(jc_wall, "jiancai")
     jc_dir = out_root / "seed_profiles" / "jiancai"
     write_json(jc_dir / "wall.json", jc_wall)
     write_json(jc_dir / "brand_values.json", build_brand_values(jc_root))
@@ -510,6 +544,7 @@ def main() -> None:
         "fallback_from": "jiancai",
         "fields": FALLBACK_FIELDS,
     }
+    _self_check(auto_wall, "auto")
     auto_dir = out_root / "seed_profiles" / "auto"
     write_json(auto_dir / "wall.json", auto_wall)
     write_json(auto_dir / "brand_values.json", build_brand_values(auto_root))

tests/test_wall_schema.py

@@ -121,3 +121,96 @@ def test_conversation_triggers_match_seed():
                             ("contact_ask", conv.CONTACT_ASK_TRIGGER_RE)):
         assert module_re.pattern == wall.triggers[name].pattern, name
         assert module_re.flags == wall.triggers[name].flags, name
+
+
+# ---- 审查修复契约(I-1/I-2/I-3 + M-4 防呆分支) ----
+
+def test_missing_es_pool_falls_back_to_en():
+    """I-1: auto 墙无 es 池,西语消息路由 es 后回落 EN 池,绝不 KeyError。"""
+    from app.engine.services.boundary_filter import deflection_for, pivot_for
+    auto = compile_wall(_load("auto"))
+    assert "es" not in auto.deflections          # 前提: auto 确实无 es 池
+    msg = "¿Cuánto cuesta?"   # 重音符 → _lang_pool 路由 es → 无池 → 降级 EN
+    assert deflection_for(msg, wall=auto) in auto.deflections["en"]
+    assert (deflection_for(msg, has_contact=True, wall=auto)
+            in auto.deflections_contact["en"])
+    assert pivot_for(msg, wall=auto) == auto.pivots["en"][0]
+    assert pivot_for(msg, has_contact=True, wall=auto) == auto.pivots_contact["en"][0]
+    # 有 es 池的墙(jiancai)不受降级逻辑影响: 仍命中 ES 池
+    jc = compile_wall(_load("jiancai"))
+    assert deflection_for(msg, wall=jc) in jc.deflections["es"]
+    assert pivot_for(msg, wall=jc) == jc.pivots["es"][0]
+
+
+def test_template_placeholder_whitelist_exported():
+    """I-2: 白名单 = conversation.py 运行期 .format() 回填键全集(grep 实证)。"""
+    from app.engine.wall_schema import TEMPLATE_PLACEHOLDERS
+    assert TEMPLATE_PLACEHOLDERS == {
+        "shown", "intl", "cc_script",
+        "no_contact_web", "no_contact_app", "no_contact_already",
+    }
+
+
+def test_template_with_literal_braces_rejected():
+    """I-2: 话术里的字面花括号(把中文"格式"写成 {格式})编译期拒绝。"""
+    cfg = _load("jiancai")
+    cfg["reminders"]["boundary"]["normal"] = "请用 {格式} 重新描述预算"
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "reminders.boundary.normal" in str(ei.value)
+
+
+def test_template_with_unknown_placeholder_rejected():
+    """I-2: 白名单外占位符({name})编译期拒绝——原属运行期 KeyError。"""
+    cfg = _load("jiancai")
+    cfg["cc_templates"]["ask"]["en"] = "Hi {name}, could you share your WhatsApp?"
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "cc_templates.ask.en" in str(ei.value)
+
+
+def test_pattern_dict_unknown_key_rejected():
+    """I-3: 键拼错({"flagss"})曾静默忽略→墙悄悄放宽,必须显性拒绝。"""
+    cfg = _load("jiancai")
+    cfg["price_patterns"] = [{"pattern": r"\$?\s*\d+", "flagss": "i"}]
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "flagss" in str(ei.value)
+    assert "pattern" in str(ei.value)   # 合法键清单一并给出
+
+
+def test_reminder_unknown_variant_key_rejected():
+    """M-4: reminders 类别下未知变体键(拼错)显性拒绝。"""
+    cfg = _load("jiancai")
+    cfg["reminders"]["boundary"]["bogus"] = "话术"
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "reminders.boundary.bogus" in str(ei.value)
+
+
+def test_pool_missing_required_language_rejected():
+    """M-4: 池缺必填语言(eng/zh 之外的核心双语)编译期拒绝。"""
+    cfg = _load("jiancai")
+    del cfg["deflections"]["zh"]
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "deflections" in str(ei.value)
+    assert "zh" in str(ei.value)
+
+
+def test_pool_non_list_items_rejected():
+    """M-4: 池语言值非列表(标量串)编译期拒绝。"""
+    cfg = _load("jiancai")
+    cfg["deflections"]["zh"] = "not-a-list"
+    with pytest.raises(WallConfigError) as ei:
+        compile_wall(cfg)
+    assert "deflections.zh" in str(ei.value)
+
+
+def test_patterns_non_list_and_empty_rejected():
+    """M-4: patterns 字段非列表/空列表均编译期拒绝。"""
+    for bad in ("nope", []):
+        cfg = _load("jiancai")
+        cfg["price_patterns"] = bad
+        with pytest.raises(WallConfigError):
+            compile_wall(cfg)