fix: 零号码墙回显豁免收窄——AI输出客户号码/邮箱一律拦截

增补设计 §3(需求方决策): 旧"礼貌回显"豁免与"不回显"红线矛盾,且被
GLM 稳定利用回显号码 3/3——AI 输出中的客户号码只可能来自回显,拦截
无误杀面。

- boundary_filter.no_contact_value: 移除 user_history 参数与豁免分支,
  ≥10 位数字串(含阿语数字形态,Unicode \d 原生匹配)/邮箱一律拦;
  digits_only 导入随之清理
- conversation.py 两处调用点同步去参(句子循环 + pending 尾巴)
- contact_pipeline.py 两处注释里指向已移除豁免的表述同步修正
- 测试: 旧豁免测试(echo_exempt/arabic_echo_exemption_consistent)按新
  语义改写为一律拦;新增邮箱回显拦截/规格形态(DN110/5000m/中文)放行;
  流式集成 2 项——客户完整号/邮箱被原样回显 → 整段替换 NO_CONTACT_ALREADY

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

app/services/boundary_filter.py

@@ -14,8 +14,6 @@
   保修期限是编造的公司政策承诺,一并墙级拦截,见 PRICE_PATTERNS 时间承诺组)。"""
 import re
 
-from app.services.contact_pipeline import digits_only
-
 PRICE_PATTERNS = [
     re.compile(r"\$\s?\d"),
     re.compile(r"\d[\d,.]*\s*(usd\b|美元|块钱|万块?(?!\s*(公里|km)))", re.I),
@@ -215,22 +213,16 @@ NO_CONTACT_ALREADY = ("Got it — our manager will reach out to you at your "
                       "WhatsApp shortly with the full quote!")
 
 
-def no_contact_value(sentence: str, user_history: list[dict] | None = None) -> bool:
+def no_contact_value(sentence: str) -> bool:
     """零号码墙(v2): AI 输出不得含任何 ≥10 位数字串或邮箱。
-    唯一豁免: 该串出现在近期客户消息中(客户自己号码的礼貌回显)。
-    无值比对——年份/规格编号天然安全(已验证)。"""
-    for m in LONG_NUMBER_RE.finditer(sentence or ""):
-        c = digits_only(m.group(0))
-        if user_history and any(c in digits_only(msg.get("content") or "")
-                                for msg in user_history if msg.get("role") == "user"):
-            continue  # 客户回显豁免
+    回显豁免已移除(2026-10-08 增补设计 §3): 红线规定客户自己的号码/邮箱
+    也不回显(确认收到即可),旧豁免与红线矛盾且被 GLM 稳定利用回显号码;
+    AI 输出中的客户号码只可能来自回显,拦截无误杀面。阿语-印度数字由
+    Unicode \\d 原生匹配(转 ASCII 后同为 ≥10 位)。无值比对——年份/规格
+    编号天然安全(DN110/5000m 等已验证)。"""
+    if LONG_NUMBER_RE.search(sentence or ""):
         return True
-    for m in ANY_EMAIL_RE.finditer(sentence or ""):
-        if user_history and any(m.group(0).lower() in (msg.get("content") or "").lower()
-                                for msg in user_history if msg.get("role") == "user"):
-            continue
-        return True
-    return False
+    return bool(ANY_EMAIL_RE.search(sentence or ""))
 
 
 # ---- Task 5b: 再索要墙(已留资时输出侧拦截) ----

app/services/contact_pipeline.py

@@ -12,9 +12,8 @@ import re
 
 # Unicode 十进制数字 → ASCII(阿语٠-٩/波斯语۰-۹), 中东买家号码 CRM 可直接拨。
 # 挂在归一化链最前端(normalize_contacts 的值 / digits_only / scan 与
-# latest 的捕获赋值): 存库、canon 去重比对、boundary_filter 客户回显豁免
-# 两侧全部经此转成 ASCII——阿语消息与 ASCII 提取值比对口径一致,
-# 同一号码不会被误判成"改号"触发更正邮件。
+# latest 的捕获赋值): 存库、canon 去重比对全部经此转成 ASCII——阿语消息
+# 与 ASCII 提取值比对口径一致,同一号码不会被误判成"改号"触发更正邮件。
 _DIGIT_TRANSLATE = str.maketrans(
     "٠١٢٣٤٥٦٧٨٩۰۱۲۳۴۵۶۷۸۹", "01234567890123456789")
 
@@ -82,9 +81,9 @@ def merge_contacts(old: dict | None, new: dict | None) -> dict:
 
 def digits_only(s: str) -> str:
     """提取纯数字（split_validate 位数判断与后续 canon 复用）。
-    阿语/波斯语数字先转 ASCII——canon 与 boundary_filter 回显豁免两侧
-    都走本函数,天然保证同形比较(Unicode 正则的数字类本就匹配阿语数字,
-    位数判定不变;此处只统一"比对与存储的形态")。"""
+    阿语/波斯语数字先转 ASCII——canon 比对两侧都走本函数,天然保证同形
+    比较(Unicode 正则的数字类本就匹配阿语数字,位数判定不变;此处只统一
+    "比对与存储的形态"。boundary_filter 回显豁免移除后不再依赖本函数)。"""
     return re.sub(r"\D", "", s.translate(_DIGIT_TRANSLATE))
 
 

app/services/conversation.py

@@ -320,7 +320,7 @@ class ConversationManager:
                     dirty = sentence
                     violation = "boundary"
                     break
-                if no_contact_value(sentence, history_for_ai):
+                if no_contact_value(sentence):
                     violated = True
                     dirty = sentence
                     violation = "no_contact"
@@ -346,7 +346,7 @@ class ConversationManager:
                 violated = True
                 dirty = pending
                 violation = "boundary"
-            elif no_contact_value(pending, history_for_ai):
+            elif no_contact_value(pending):
                 violated = True
                 dirty = pending
                 violation = "no_contact"

tests/test_boundary_filter.py

@@ -117,9 +117,11 @@ def test_no_contact_wall_blocks_any_number_or_email():
     assert no_contact_value("our email is sales@example.com")
 
 
-def test_no_contact_wall_customer_echo_exempt():
-    history = [{"role": "user", "content": "my number is 07012345678"}]
-    assert not no_contact_value("Got it, 07012345678!", history)
+def test_no_contact_wall_echo_no_longer_exempt():
+    # 回显豁免收窄(2026-10-08 增补设计 §3): 客户自己的号码被 AI 回显也拦——
+    # 红线早已规定"确认收到即可,不回显",旧豁免与红线矛盾且被 GLM 稳定利用
+    # (回显 3/3)。user_history 参数已随豁免一并移除。
+    assert no_contact_value("Got it, 07012345678!")
 
 
 def test_no_contact_wall_whitelist_safe():
@@ -382,14 +384,26 @@ def test_no_auto_wording_left_in_any_pool():
             assert "什么车" not in t, f"{name}: {t}"
 
 
-def test_arabic_echo_exemption_consistent():
+def test_no_contact_wall_blocks_echo_in_all_digit_scripts():
+    # 回显豁免收窄(2026-10-08): 阿语数字形态与 ASCII 形态一律拦——
+    # Unicode \d 本就匹配阿语-印度数字,转 ASCII 后同为 ≥10 位。
+    # 旧豁免(Task 6)曾按"客户消息出现过"放行两形态,现全部拦截。
+    # 阿语数字用码位构造,避免源文件编码事故。
     ar_num = "".join(chr(c) for c in
                   (0x0669, 0x0666, 0x0666, 0x0665, 0x0660, 0x0661,
                    0x0662, 0x0663, 0x0664, 0x0665, 0x0666, 0x0667))  # 966501234567
-    # 阿语回显豁免一致性锁定(Task 6 质量审查): 客户用阿语数字留号,AI 无论
-    # 回显阿语还是 ASCII 形态都豁免——两侧均经 digits_only(内含 translate)。
-    # 阿语数字用码位构造,避免源文件编码事故。
-    hist = [{"role": "user", "content": "my whatsapp is " + ar_num}]
-    assert not no_contact_value("Got it — messaging you at " + ar_num, hist)
-    assert not no_contact_value("Got it — messaging you at +966501234567", hist)
-    assert no_contact_value("Reach me at 11987654321", hist)  # 不同号(11位)仍拦
+    assert no_contact_value("Got it — messaging you at " + ar_num)
+    assert no_contact_value("Got it — messaging you at +966501234567")
+    assert no_contact_value("Reach me at 11987654321")            # 11位裸串
+
+
+def test_no_contact_wall_email_echo_blocked():
+    # 邮箱回显同规则收窄: 确认句只需"收到",不需复述地址
+    assert no_contact_value("Got it — a@b.com, manager will write you")
+
+
+def test_no_contact_wall_spec_forms_still_pass():
+    # 误杀防护(增补设计 §3): 建材规格数字形态不构成 ≥10 位连续数字
+    assert not no_contact_value("DN110 PE100 pipes, 5000m total, 20ft container")
+    assert not no_contact_value("PN16 SDR11 to ISO4427, batches of 5000")
+    assert not no_contact_value("DN800双壁波纹管，5000米一批")

tests/test_lead_email_trigger.py

@@ -1179,3 +1179,47 @@ def test_cc_wall_chinese_template(async_db, monkeypatch):
     assert tail == _CC_WALL_ZH
     assert "国家代码" in tail
     asyncio.run(manager.db.close())
+
+
+# ---- 回显豁免收窄(2026-10-08 增补设计 §3: 零号码墙输出侧一律拦) ----
+def test_no_contact_wall_blocks_exact_customer_number_echo(async_db, monkeypatch):
+    """客户自己的号码被 AI 原样回显 → 拦截整段替换(旧豁免放行此形态——
+    GLM 稳定利用它回显号码,与"不回显"红线矛盾)。带前缀号构造,避开 CC
+    触发轮,隔离验证本规则。"""
+    from app.services.boundary_filter import NO_CONTACT_ALREADY
+
+    class ExactEchoAI(FakeAI):
+        async def stream_chat(self, messages, context=""):
+            for token in ("Got it — +966501234567! ",
+                          "Our manager will contact you soon."):
+                yield token
+
+    (manager, conv), _ = _setup(async_db, monkeypatch,
+                                ai=ExactEchoAI(extract={}),
+                                source=ConversationSource.standalone)
+    asyncio.run(_send(manager, conv.session_id, "my whatsapp is +966501234567"))
+    msgs = asyncio.run(manager._load_history(conv.id))
+    tail = [m["content"] for m in msgs if m["role"] == "assistant"][-1]
+    assert "+966501234567" not in tail     # 回显零残留
+    assert tail == NO_CONTACT_ALREADY      # 已留资 → 确认话术替换
+    asyncio.run(manager.db.close())
+
+
+def test_no_contact_wall_blocks_customer_email_echo(async_db, monkeypatch):
+    """客户邮箱被 AI 回显 → 同规则拦截(确认收到即可,不复述地址)。"""
+    from app.services.boundary_filter import NO_CONTACT_ALREADY
+
+    class EmailEchoAI(FakeAI):
+        async def stream_chat(self, messages, context=""):
+            for token in ("Got it, a@b.com — ", "quote goes there soon!"):
+                yield token
+
+    (manager, conv), _ = _setup(async_db, monkeypatch,
+                                ai=EmailEchoAI(extract={}),
+                                source=ConversationSource.standalone)
+    asyncio.run(_send(manager, conv.session_id, "my email is a@b.com"))
+    msgs = asyncio.run(manager._load_history(conv.id))
+    tail = [m["content"] for m in msgs if m["role"] == "assistant"][-1]
+    assert "a@b.com" not in tail
+    assert tail == NO_CONTACT_ALREADY
+    asyncio.run(manager.db.close())