feat: 吸收上游 7 个测试文件断言与 placeholder 配置化对齐

上游 ai-kefu → phase1 吸收(Task 11 unit 12):
- 拷入 7 个测试文件,import/夹具按目标库重锚(app.engine.* / conftest
  async_db+build_jiancai_runtime / seed_profiles 提示词锚点)
- 补实现(同步自上游): chat_rest 服务端可信 IP(_server_ip/
  _merge_client_info,直连场景自报假 IP 不再原样落库);widget.js 与
  chat.js 的 data-placeholder 读取(叠加 Task 10 管道:data-* > brand >
  中性兜底,tests/test_placeholder_static.py 守卫)
- 断言重锚以 jc 主线为准: ev.contact_info 存裸号+等待尾注(e164 走
  _cc 派生)、source 值 fallback/customer_stated 口径、推送标注键
  号码补全说明、prompt 静态锚按租户(seed_profiles)双测
- 全量回归 459 passed(基线 421 + 吸收 38)

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

app/engine/routes/chat_rest.py

@@ -33,6 +33,23 @@ class SessionRequest(BaseModel):
     client_info: dict | None = None
 
 
+def _server_ip(request: Request) -> str:
+    """服务端可信 IP：XFF 第一段优先，否则直连 host；全空返回空串。
+
+    同步自上游 ai-kefu(app/api/routes/chat_rest.py)：REST 路径原先只在有
+    XFF 时覆盖自报 ip，直连(无代理)场景自报假 IP 原样落库。"""
+    xff = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+    return xff or (request.client.host if request.client else "")
+
+
+def _merge_client_info(client_info: dict | None, server_ip: str) -> dict:
+    """服务端 IP 覆盖自报 ip 键；服务端为空时保留自报（空串不得冲掉客户端值）。"""
+    merged = dict(client_info or {})
+    if server_ip:
+        merged["ip"] = server_ip
+    return merged
+
+
 @router.post("/session")
 async def chat_session(req: SessionRequest, request: Request):
     """Resume an active conversation by session_id, or create a new app one.
@@ -41,11 +58,9 @@ async def chat_session(req: SessionRequest, request: Request):
     `raw_info` (only when non-None). `client_info` is stored for analytics only
     (ip/ua/device-id/lang) and never fed to the AI prompt.
     """
-    # 服务端 XFF 优先于客户端自报 ip（客户端自报可伪造，可伪造面最小）
-    xf = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
-    client_info = dict(req.client_info or {})
-    if xf:
-        client_info["ip"] = xf
+    # 服务端可信 IP 同步自上游 ai-kefu：XFF 第一段优先，无 XFF 用直连 host
+    # 覆盖客户端自报（可伪造面最小）；全空保留自报。?site= 处理不变。
+    client_info = _merge_client_info(req.client_info, _server_ip(request))
 
     # Task 10: ?site=<slug> 记入 client_info 作会话站点标(与 WS init 的
     # site 字段同语义)——只加记录,不参与业务逻辑

static/js/chat.js

@@ -51,6 +51,9 @@ var chatBrandReady = (function () {
     var chatScript = document.currentScript || document.querySelector('script[src*="chat.js"]');
     var whatsappAttr = chatScript && chatScript.getAttribute("data-whatsapp");
     var emailAttr = chatScript && chatScript.getAttribute("data-email");
+    // 同步自上游 ai-kefu: data-placeholder 显式覆盖输入框占位文案(裸读,
+    // 有效值在 applyBrand 推导——data-* > brand > 不动 chat.html 静态值)
+    var placeholderAttr = chatScript && chatScript.getAttribute("data-placeholder");
     var waBtn = document.querySelector('.whatsapp-btn');
     var emBtn = document.querySelector('.email-btn');
     var sessionId = null;
@@ -85,7 +88,10 @@ var chatBrandReady = (function () {
         if (waBtn) { if (wa) waBtn.href = wa; else waBtn.style.display = "none"; }
         var em = contactHref(emailAttr || b.email || "", "mail");
         if (emBtn) { if (em) emBtn.href = em; else emBtn.style.display = "none"; }
-        if (b.placeholder) inputEl.placeholder = b.placeholder;
+        // 占位文案优先级(同步自上游 ai-kefu,叠加 Task 10 管道):
+        // data-placeholder > brand.placeholder > 不触碰(chat.html 静态兜底)
+        var placeholderText = placeholderAttr || b.placeholder || "";
+        if (placeholderText) inputEl.placeholder = placeholderText;
         var firstMsg = document.querySelector("#messages .message");
         var bubble = firstMsg && firstMsg.querySelector(".bubble");
         if (b.welcome) {

static/widget.js

@@ -11,6 +11,10 @@
     var position = script.getAttribute("data-position") || "bottom-right";
     var theme = script.getAttribute("data-theme") || "light";
     var titleAttr = script.getAttribute("data-title");
+    // 同步自上游 ai-kefu: data-placeholder 显式覆盖输入框占位文案。裸读不带
+    // 默认值——Task 10 管道里默认值在 boot() 推导链尾(读取行给默认会让
+    // brand.placeholder 永远无法生效),tests/test_placeholder_static.py 守卫。
+    var placeholderAttr = script.getAttribute("data-placeholder");
     var autoOpen = script.getAttribute("data-auto-open") === "true";
     var whatsappAttr = script.getAttribute("data-whatsapp");
     var emailAttr = script.getAttribute("data-email");
@@ -164,7 +168,7 @@
         topics = Array.isArray(b.topics)
             ? b.topics.filter(function (t) { return typeof t === "string" && t; })
             : [];
-        placeholder = b.placeholder || "Type a message...";
+        placeholder = placeholderAttr || b.placeholder || "Type a message...";
         privacyUrl = b.privacy_url || "";
         brandSlug = String(b.slug || "default").replace(/[^a-zA-Z0-9_-]/g, "") || "default";
         if (script.getAttribute("data-debug") === "true") {

tests/test_cc_integration.py

@@ -0,0 +1,379 @@
+"""CC 集成测试(同步自上游 ai-kefu tests/test_cc_integration.py)。
+
+惯例复用 tests/test_lead_email_trigger.py: FakeAI 确定性桩 + 邮件 monkeypatch
+捕获 sent 列表 + async_db 独立库。jc 侧适配:
+- ConversationManager 必传 runtime(Task 6)——build_jiancai_runtime() 单例;
+- 邮件桩收 cc_info/config_data 关键字(Task 6/9 出站双视图与租户配置);
+- 存储契约重锚(Task 6 评审钉死): ev.contact_info 存裸号原文+等待态尾注
+  "（待确认补全号）",e164 只活在对 _cc 的派生(邮件/推送出站渲染),上游
+  "ev.contact_info == e164" 的断言改为裸号形态 + _cc.e164 双断言;
+- source 值: 上游 "ip_fallback" → jc "fallback";报国重算后的确认 jc 保持
+  "customer_stated"(号源如实标注),上游记 "ip_confirm";
+- 国名纠正(canon 不变、e164 变)不触发更正邮件: Task 6 _channel_canons
+  双保险——同 canon 换形态/换国不误判换号(上游存 e164 于 contact_info,
+  canon 随国变才触发 fix;jc 主线接受"已发邮件号码滞后"换取不误发,见
+  报告存疑项),状态机仍重算 e164 供后续出站使用。
+
+接线注意: 每个用例的全部轮次与断言读取跑在同一个事件循环里(asyncio.run
+一次包到底,经 _scenario)——断言读的是新事务的库值(_get_profile 会因弱
+identity map 重新 SELECT),而非可能残留内存态的实例缓存,状态机推进
+(_cc 深拷贝修复)因此能被真正验证。
+"""
+import asyncio
+
+from sqlalchemy import select
+
+from app.engine.models import Evaluation
+from app.engine.services import email_service
+from app.engine.services.conversation import ConversationManager
+from tests.conftest import build_jiancai_runtime
+
+KE_IP = "41.90.64.1"   # 肯尼亚段(iptocc 实测 → KE)
+
+# Task 6 面板尾注(等待态期间 ev.contact_info 的存储形态)
+PENDING_TAIL = "（待确认补全号）"
+
+
+class FakeAI:
+    """替代 GLM 的确定性桩: 构造参数即每轮返回值。"""
+
+    def __init__(self, extract=None, evaluate=None):
+        self._extract_seq = extract if isinstance(extract, list) else [extract or {}]
+        self._evaluate = evaluate or {}
+        self.eval_calls = 0
+
+    async def chat(self, messages, context=""):
+        return "ok reply"
+
+    async def stream_chat(self, messages, context=""):
+        for token in ("ok ", "reply"):
+            yield token
+
+    async def extract_info(self, messages, existing_info):
+        # 第 N 轮客户消息 → seq[N-1](messages 为 [u,a,u,a...] 全量历史)
+        idx = min(max(len(messages) // 2 - 1, 0), len(self._extract_seq) - 1)
+        return self._extract_seq[idx]
+
+    async def evaluate(self, profile, recent_messages):
+        self.eval_calls += 1
+        result = dict(self._evaluate)
+        result.setdefault("intent_level", "pending")
+        result.setdefault("reasoning", "")
+        result.setdefault("contact_info", {})
+        result.setdefault("info_sufficient", False)
+        if result["intent_level"] not in ("interested", "not_interested", "pending"):
+            result["intent_level"] = "pending"
+        return result
+
+
+async def _send(manager, sid, text):
+    async for _ in manager.stream_process_message(sid, text):
+        pass
+
+
+def _scenario(async_db, monkeypatch, ai, steps, ip=KE_IP):
+    """单事件循环跑完整用例: 建会话 → steps(manager, conv) → 读 ev/_cc 库值。
+
+    steps 为 async callable;返回值原样透传。sent 捕获邮件侧效应(lead 项
+    附带 cc_info 快照——jc 出站 e164 走 _cc 派生,不入 contact_info 本体)。
+    ev/状态读取与轮次同循环同事务空间,断言看到的是落库真值。
+    """
+    sent = []
+
+    async def fake_lead(customer_profile, contact_info, reasoning,
+                        conversation_messages, intent_level="",
+                        cc_info=None, config_data=None):
+        sent.append(("lead", dict(contact_info), intent_level,
+                     {k: dict(v) for k, v in (cc_info or {}).items()}))
+        return True
+
+    async def fake_nudge(contact_info, cc_info=None, config_data=None):
+        sent.append(("nudge", dict(contact_info), ""))
+        return True
+
+    async def fake_fix(contact_info, count=1, cc_info=None, config_data=None):
+        sent.append(("fix", dict(contact_info), ""))
+        return True
+
+    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
+    monkeypatch.setattr(email_service, "send_returning_lead_email", fake_nudge)
+    monkeypatch.setattr(email_service, "send_correction_email", fake_fix)
+
+    async def run():
+        async with async_db() as db:
+            manager = ConversationManager(
+                db, runtime=build_jiancai_runtime(), ai_service=ai)
+            conv = await manager.create_conversation(
+                client_info={"ip": ip} if ip else None)
+            result = await steps(manager, conv)
+            ev = (await db.execute(
+                select(Evaluation).where(
+                    Evaluation.conversation_id == conv.id)
+            )).scalar_one()
+            profile = await manager._get_profile(conv.id)
+            cc_state = ((profile.raw_info or {}).get("_cc") or {}).get(
+                "whatsapp") or {}
+            return result, ev, cc_state
+
+    return asyncio.run(run()), sent
+
+
+def test_first_email_carries_e164(async_db, monkeypatch):
+    # 首封即发(时序契约): 裸号 confirmed 当轮,邮件/落库即带 IP 补全 e164,
+    # 确认流挂 _cc(confirmed=False, source=ip_proposed)。jc 契约: 邮件收
+    # 裸号+cc_info(e164 在 _cc,由 email_service 出站渲染),ev.contact_info
+    # 存裸号+等待尾注(e164 派生不落本体)。
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract={"联系方式": {"WhatsApp": "0712345678"}}), steps)
+    assert len(sent) == 1 and sent[0][0] == "lead"
+    assert sent[0][1]["whatsapp"] == "0712345678"          # 本体=裸号原文
+    assert sent[0][3]["whatsapp"]["e164"] == "+254712345678"  # 补全号随 _cc 到邮件
+    assert ev.contact_info == {"whatsapp": f"0712345678{PENDING_TAIL}"}
+    assert st["e164"] == "+254712345678" and st["confirmed"] is False
+    assert st["source"] == "ip_proposed"
+
+
+def test_reminder_cc_pending_overrides_d6():
+    # cc-pending 提醒优先于 D6(spec §3.4 拦截1): 回显确认指令在场。
+    # jc 措辞演化(Task 3/4): 回显句 = "is that right"(上游 "Is this your
+    # full number");D6 禁令改写为"只禁已提供渠道+补全号回显豁免",上游
+    # "绝对禁止再索要"整句不复存在——改锚改写后的豁免条款在场。
+    raw = {"联系方式": {"whatsapp": "0712345678"},
+           "_cc": {"whatsapp": {"e164": "+254712345678", "intl": "+254 712 345678",
+                                "confirmed": False, "attempts": 0,
+                                "source": "ip_proposed", "canon": "0712345678"}}}
+    text = ConversationManager._build_context_reminder(raw, "yes", [])
+    assert "+254 712 345678" in text
+    assert "is that right" in text.lower()      # 回显确认句(模板同源句式)
+    assert "回显补全后的完整号码" in text        # D6 改写后的 cc 豁免条款
+    assert "缺失渠道仅可按轻补指引提一次" in text  # 防自锁改写在位
+
+
+def test_affirm_confirms_no_resend(async_db, monkeypatch):
+    # 客户确认(号码未变): 不补发(首封已可拨),状态升级 confirmed/ip_confirm
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        await _send(manager, conv.session_id, "yes that's right")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract=[{"联系方式": {"WhatsApp": "0712345678"}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": True}}]), steps)
+    assert len(sent) == 1
+    assert ev.correction_count == 0
+    # 存储形态: 首封时落的裸号+尾注不再更新(e164 派生口 _cc 已 confirmed)
+    assert ev.contact_info == {"whatsapp": f"0712345678{PENDING_TAIL}"}
+    assert st["confirmed"] is True and st["source"] == "ip_confirm"
+
+
+def test_country_correction_recomputes_state_same_canon_no_email(async_db, monkeypatch):
+    # 国名纠正(canon 不变、e164 变)——上游断言落入既有更正触发器(fix 邮件);
+    # jc 主线(Task 6 _channel_canons 双保险)同 canon 不误判换号: 不发更正
+    # 邮件,但状态机照常重算 e164/customer_stated,后续出站(回显/推送)用新号。
+    # 已发邮件的旧 e164 滞后是 jc 接受的取舍(见文件头注释与报告存疑项)。
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        await _send(manager, conv.session_id, "no, I am in Uganda")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract=[{"联系方式": {"WhatsApp": "0712345678"}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": False, "country": "Uganda"}}]),
+        steps)
+    assert len(sent) == 1 and sent[0][0] == "lead"   # 同 canon: 无更正邮件
+    assert ev.correction_count == 0
+    assert st["e164"] == "+256712345678"             # 乌干达重算
+    assert st["source"] == "customer_stated" and st["attempts"] == 1
+
+
+def test_no_change_no_correction(async_db, monkeypatch):
+    # 防误发: 号码与国名均不变 → 不触发更正;连续无回应 → attempts=2 落 fallback
+    # (上游 source 值 "ip_fallback",jc 统一为 "fallback")
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        await _send(manager, conv.session_id, "so how much to ship")
+        await _send(manager, conv.session_id, "ok thanks")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract={"联系方式": {"WhatsApp": "0712345678"}}), steps)
+    assert len(sent) == 1 and sent[0][0] == "lead"
+    assert ev.correction_count == 0
+    assert st["attempts"] == 2 and st["source"] == "fallback"
+
+
+def test_unrelated_yes_not_consumed(async_db, monkeypatch):
+    # 双门控第二道: 已 confirmed 后再来的 affirm 不得再消费(无副作用)
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        await _send(manager, conv.session_id, "yes correct")
+        await _send(manager, conv.session_id, "yes yes")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract=[{"联系方式": {"WhatsApp": "0712345678"}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": True}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": True}}]), steps)
+    assert len(sent) == 1
+    assert ev.correction_count == 0
+    assert st["confirmed"] is True and st["source"] == "ip_confirm"
+
+
+def test_extractor_only_capture_still_completed(async_db, monkeypatch):
+    # 轮前扫描未命中(消息无数字串),仅 extractor 捕获 → 轮末兜底建态,
+    # 首封仍带 e164(裸号本体 + _cc 派生) + ip_proposed 标注态
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "you can whatsapp me anytime")
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        FakeAI(extract={"联系方式": {"WhatsApp": "0712345678"}}), steps)
+    assert sent and sent[0][0] == "lead"
+    assert sent[0][1]["whatsapp"] == "0712345678"
+    assert sent[0][3]["whatsapp"]["e164"] == "+254712345678"
+    assert st["e164"] == "+254712345678" and st["source"] == "ip_proposed"
+
+
+def test_echo_sentence_survives_zero_number_wall(async_db, monkeypatch):
+    # C1(用户批准): cc-pending 期间系统补全号回显句不得被零号码墙替换——
+    # 补全号数字(254…)与客户裸号(071…)永不成子串关系,无白名单必被拦
+    # (jc Task 4: 回显轮零号码墙放行注入 e164 的 digits 等值,cc_allow_number)
+    class EchoAI(FakeAI):
+        def __init__(self, *a, **k):
+            super().__init__(*a, **k)
+            self._round = 0
+
+        async def stream_chat(self, messages, context=""):
+            self._round += 1
+            if self._round == 1:
+                for token in ("ok ", "reply"):
+                    yield token
+                return
+            for token in ("Is this your full number: ",
+                          "+254 712 345678", "?"):
+                yield token
+
+    events = []
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        async for e in manager.stream_process_message(
+                conv.session_id, "check please"):
+            events.append(e)
+
+    (result, ev, st), sent = _scenario(
+        async_db, monkeypatch,
+        EchoAI(extract=[{"联系方式": {"WhatsApp": "0712345678"}},
+                        {"联系方式": {"WhatsApp": "0712345678"}}]), steps)
+    streamed = "".join(e.get("content", "") for e in events
+                       if e["type"] == "stream_chunk")
+    assert "254 712 345678" in streamed
+
+
+def test_reminder_no_ip_variant_asks_full_number():
+    # no_ip 变体: 无建议国 → 请发带区号完整号,不用补全回显句式。
+    # 渠道重锚: 上游用 phone,jc 确认流只走 whatsapp 组(phone 态仅供出站
+    # 渲染、不进确认流——phone-only no_ip 会落到邮箱轻补分支)。
+    # jc(Task 3 D6 改写)已留资值出现在禁令行 shown 中——上游
+    # "18325698745 not in text" 随改写调整: 号码仅出现在禁令上下文。
+    raw = {"联系方式": {"whatsapp": "18325698745"},
+           "_cc": {"whatsapp": {"e164": None, "intl": None, "confirmed": False,
+                                "attempts": 0, "source": "no_ip",
+                                "canon": "18325698745"}}}
+    text = ConversationManager._build_context_reminder(raw, "yes", [])
+    assert "country code" in text
+    assert "is that right" not in text.lower()   # 无补全号可回显 → 追问句
+    assert "18325698745" in text                 # 仅禁令行 shown 上下文
+
+
+def test_country_correction_re_echo_then_affirm_confirms(async_db, monkeypatch):
+    # 国名纠正成功不烧 attempts(再回显兑现)→ 客户确认采纳: r3 回显新号供
+    # 确认,affirm 后落定。jc 主线: 同 canon 不发更正邮件(全程恰一封),
+    # 确认后 source 保持 customer_stated(号源如实标注,上游记 ip_confirm)
+    class CtxAI(FakeAI):
+        def __init__(self, *a, **k):
+            super().__init__(*a, **k)
+            self.contexts = []
+
+        async def stream_chat(self, messages, context=""):
+            self.contexts.append(context)
+            for token in ("ok ", "reply"):
+                yield token
+
+    ai = CtxAI(extract=[{"联系方式": {"WhatsApp": "0712345678"}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": False, "country": "Uganda"}},
+                        {"联系方式": {"WhatsApp": "0712345678"},
+                         "cc_reply": {"affirm": True}}])
+
+    async def steps(manager, conv):
+        await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+        await _send(manager, conv.session_id, "no, I am in Uganda")
+        await _send(manager, conv.session_id, "yes that's right")
+
+    (result, ev, st), sent = _scenario(async_db, monkeypatch, ai, steps)
+    assert len(sent) == 1 and sent[0][0] == "lead"   # 同 canon: lead 之外无 fix
+    ctx3 = ai.contexts[2]
+    assert "+256 712 345678" in ctx3          # 再回显: 新补全号进提醒
+    assert "is that right" in ctx3.lower()    # 回显确认句(jc 模板句式)
+    assert "回显补全后的完整号码" in ctx3      # cc-pending 分支优先(D6 改写形态)
+    assert ev.correction_count == 0
+    assert ev.contact_info == {"whatsapp": f"0712345678{PENDING_TAIL}"}
+    assert st["confirmed"] is True and st["source"] == "customer_stated"
+    assert st["attempts"] == 1
+
+
+def test_d8_annotates_when_email_not_sent(async_db, monkeypatch):
+    # D8 面板尾注兜底(Task7评审补钉): SMTP 失败(send_lead_email 返回 False)
+    # → email_sent 落不了 True,ev.contact_info 由 D8 分支写存储形态。
+    # D8 门控要求 ≥3 条客户消息;前两轮发信失败会逐轮重试(仍失败),
+    # 第 3 轮 D8 落面板视图(2 轮无回应后 source=fallback 已落定,无尾注)。
+    # jc 存储形态: 裸号原文(e164 派生口 _cc 仍持有,assert 双钉)——上游
+    # 把标注写进 contact_info 值本身,jc 的标注只在邮件/推送出站渲染。
+
+    async def failing_lead(customer_profile, contact_info, reasoning,
+                           conversation_messages, intent_level="",
+                           cc_info=None, config_data=None):
+        return False
+
+    monkeypatch.setattr(email_service, "send_lead_email", failing_lead)
+
+    async def run():
+        async with async_db() as db:
+            manager = ConversationManager(
+                db, runtime=build_jiancai_runtime(),
+                ai_service=FakeAI(
+                    extract={"联系方式": {"WhatsApp": "0712345678"}}))
+            conv = await manager.create_conversation(
+                client_info={"ip": KE_IP})
+            await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+            await _send(manager, conv.session_id, "how much to ship")
+            await _send(manager, conv.session_id, "ok thanks")
+            profile = await manager._get_profile(conv.id)
+            st = ((profile.raw_info or {}).get("_cc") or {}).get("whatsapp") or {}
+        async with async_db() as db:   # fresh-DB read: 新事务读落库真值
+            ev = (await db.execute(
+                select(Evaluation).where(
+                    Evaluation.conversation_id == conv.id)
+            )).scalar_one()
+            return ev, st
+
+    ev, st = asyncio.run(run())
+    assert ev.email_sent is False
+    assert ev.contact_info == {"whatsapp": "0712345678"}  # 落定态: 裸号无尾注
+    assert st["source"] == "fallback"                     # 2 轮无回应落定
+    assert st["e164"] == "+254712345678"                  # 补全号仍在 _cc 供出站

tests/test_channel_hygiene.py

@@ -0,0 +1,107 @@
+"""内部状态卫生(同步自上游 ai-kefu tests/test_channel_hygiene.py)。
+
+上游两测分别锚 AIService.extract_info 内部 `_` 过滤与 _merge_extracted
+纯函数的出口剥离;jc 主线把这两处卫生逻辑收敛进 stream_process_message
+的调用点(入口侧 existing_info 过滤 / 出口侧 cc_reply 剥离 + `_` 回显
+丢弃,不再有独立函数)——断言按 jc 锚点重锚到 manager 整轮,语义等价:
+`_` 键不进提示词、cc_reply 消费即剥离不落 raw_info、extractor 回显的
+`_` 状态不得覆盖 Python 管理的状态机。
+"""
+import asyncio
+
+from app.engine.services import email_service
+from app.engine.services.conversation import ConversationManager
+from tests.conftest import build_jiancai_runtime
+
+KE_IP = "41.90.64.1"   # 肯尼亚段(iptocc 实测 → KE)
+
+
+class FakeAI:
+    """替代 GLM 的确定性桩(仅卫生测试所需的最小面)。"""
+
+    def __init__(self, extract=None):
+        self._extract_seq = extract if isinstance(extract, list) else [extract or {}]
+
+    async def stream_chat(self, messages, context=""):
+        for token in ("ok ", "reply"):
+            yield token
+
+    async def extract_info(self, messages, existing_info):
+        idx = min(max(len(messages) // 2 - 1, 0), len(self._extract_seq) - 1)
+        return self._extract_seq[idx]
+
+    async def evaluate(self, profile, recent_messages):
+        return {"intent_level": "pending", "reasoning": "",
+                "contact_info": {}, "info_sufficient": False}
+
+
+async def _send(manager, sid, text):
+    async for _ in manager.stream_process_message(sid, text):
+        pass
+
+
+def _no_email(monkeypatch):
+    """卫生测试不关心邮件侧效应——桩掉避免真实 SMTP 配置探测。"""
+
+    async def fake_lead(*a, **k):
+        return True
+
+    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
+
+
+def test_extract_info_filters_underscore_keys(async_db, monkeypatch):
+    # 入口侧(上游锚 AIService 内部过滤,jc 锚调用点): `_` 前缀内部态
+    # (_cc 状态机等)绝不进 LLM 输入,正常键照常透传。
+    _no_email(monkeypatch)
+    captured = {}
+
+    class RecAI(FakeAI):
+        async def extract_info(self, messages, existing_info):
+            captured["existing"] = existing_info
+            return {}
+
+    async def run():
+        async with async_db() as db:
+            manager = ConversationManager(
+                db, runtime=build_jiancai_runtime(), ai_service=RecAI())
+            conv = await manager.create_conversation(client_info={"ip": KE_IP})
+            profile = await manager._get_profile(conv.id)
+            profile.raw_info = {"name": "Sam",
+                                "_cc": {"whatsapp": {"e164": "+254712345678"}}}
+            db.add(profile)
+            await db.commit()
+            await _send(manager, conv.session_id, "hello are you there")
+
+    asyncio.run(run())
+    assert captured["existing"] == {"name": "Sam"}   # _cc 不进提示词,name 保留
+
+
+def test_extractor_output_strips_cc_reply_and_underscore_echo(async_db, monkeypatch):
+    # 出口侧(上游锚 _merge_extracted,jc 锚 stream_process_message 内联):
+    # cc_reply 提取后即剥离不落 raw_info;extractor 回显的任何 `_` 键
+    # (含 _cc 状态机回显)防御性丢弃——Python 管理的状态机不被覆盖;
+    # 正常键(含联系方式深合并)照常进 raw_info。
+    _no_email(monkeypatch)
+
+    async def run():
+        async with async_db() as db:
+            manager = ConversationManager(
+                db, runtime=build_jiancai_runtime(),
+                ai_service=FakeAI(extract={
+                    "联系方式": {"whatsapp": "0712345678"},
+                    "cc_reply": {"affirm": True},
+                    "_cc": {"whatsapp": {"e164": "HACKED"}},
+                    "_evil": "x",
+                    "name": "Sam2"}))
+            conv = await manager.create_conversation(client_info={"ip": KE_IP})
+            await _send(manager, conv.session_id, "my whatsapp is 0712345678")
+            profile = await manager._get_profile(conv.id)
+            return dict(profile.raw_info or {})
+
+    raw = asyncio.run(run())
+    assert "cc_reply" not in raw and "_evil" not in raw
+    assert raw["name"] == "Sam2"
+    assert raw["联系方式"]["whatsapp"] == "0712345678"
+    # _cc 在场但属 Python 建态(判定树按 KE IP 补全),extractor 回显未生效
+    st = raw["_cc"]["whatsapp"]
+    assert st["e164"] == "+254712345678" and st["source"] == "ip_proposed"

tests/test_chat_rest_ip.py

@@ -0,0 +1,56 @@
+"""REST /session 服务端 IP 注入(同步自上游 ai-kefu tests/test_chat_rest_ip.py)。
+
+spec 语义: XFF 第一段优先压过客户端自报,堵伪造。两处按目标库重锚:
+- 实现侧: _server_ip/_merge_client_info 已照上游同步进
+  app/engine/routes/chat_rest.py( jc 原先只在有 XFF 时覆盖,直连场景自报
+  假 IP 原样落库——上游修了 jc 没跟上的 bug,补实现);
+- 读回侧: 上游覆盖 chat_rest.async_session(全局 sessionmaker),jc(Task 5b)
+  无全局单例——走 TestClient portal + app.state.registry.resolve,请求与
+  读回同一事件循环(engine 连接绑定安全,conftest app_client 夹具纪律)。
+WS 路径(app/engine/routes/chat.py)保持 jc 现状(无 XFF 时保留自报),
+两侧不对称是 jc 主线决策,见该文件注释。
+"""
+from sqlalchemy import select
+
+from app.engine.models import Conversation
+from tests.conftest import JIANCAI_HOST
+
+
+def _create(client, headers=None, body_client_info=None):
+    return client.post("/api/chat/session",
+                       json={"client_info": body_client_info}, headers=headers or {})
+
+
+def _conv_client_info(client, session_id):
+    async def run():
+        registry = client.app.state.registry
+        rt = await registry.resolve(JIANCAI_HOST)
+        async with rt.session_factory() as db:
+            conv = (await db.execute(
+                select(Conversation).where(Conversation.session_id == session_id)
+            )).scalar_one()
+            return dict(conv.client_info or {})
+
+    return client.portal.call(run)
+
+
+def test_xff_first_segment_overrides_self_reported(client):
+    resp = _create(client, headers={"X-Forwarded-For": "41.90.64.1, 10.0.0.1"},
+                   body_client_info={"ip": "1.2.3.4", "ua": "test"})
+    assert resp.status_code == 200 and resp.json()["is_new"] is True
+    ci = _conv_client_info(client, resp.json()["session_id"])
+    assert ci["ip"] == "41.90.64.1"      # 服务端 XFF 第一段压过自报假 IP
+    assert ci["ua"] == "test"            # 其余自报键保留
+
+
+def test_no_xff_falls_back_to_direct_host(client):
+    resp = _create(client, body_client_info={"ip": "1.2.3.4"})
+    ci = _conv_client_info(client, resp.json()["session_id"])
+    assert ci["ip"] == "testclient"      # TestClient 直连 host 覆盖自报
+
+
+def test_merge_client_info_empty_server_ip_preserves_self_report():
+    from app.engine.routes.chat_rest import _merge_client_info
+    assert _merge_client_info({"ip": "9.9.9.9", "ua": "x"}, "") == {"ip": "9.9.9.9", "ua": "x"}
+    assert _merge_client_info(None, "") == {}
+    assert _merge_client_info(None, "41.90.64.1") == {"ip": "41.90.64.1"}

tests/test_ip_service_region.py

@@ -0,0 +1,24 @@
+"""ip_service 区码扩展(同步自上游 ai-kefu tests/test_ip_service_region.py)。
+
+仅改 import 路径(app.services → app.engine.services);断言原样——
+iptocc 数据面行为两库同源,实测 41.90.64.1→KE / 24.114.0.1→加拿大。
+"""
+from app.engine.services.ip_service import COUNTRY_NAMES, resolve_country, resolve_region_code
+
+
+def test_resolve_region_code_direct_passthrough():
+    assert resolve_region_code("8.8.8.8") in ("US", None)  # iptocc 数据面,断言不抛
+    assert resolve_region_code("127.0.0.1") is None        # 本地
+    assert resolve_region_code("") is None
+
+
+def test_resolve_region_code_matches_resolve_country_space():
+    # 同一 IP:resolve_region_code 返回 ISO,resolve_country 走 COUNTRY_NAMES 映射
+    code = resolve_region_code("41.90.64.1")  # 肯尼亚段
+    assert code == "KE"
+
+
+def test_ca_country_name_fixed():
+    assert COUNTRY_NAMES["CA"] == "加拿大"
+    assert COUNTRY_NAMES["AU"] == "澳大利亚"
+    assert resolve_country("24.114.0.1") != "CA"  # 加拿大段不再透出字面量

tests/test_lead_push_cc.py

@@ -0,0 +1,141 @@
+"""推送侧 CC 补全渲染(同步自上游 ai-kefu tests/test_lead_push_cc.py)。
+
+惯例照搬 tests/test_lead_push_service.py: async_db fixture +
+monkeypatch settings.PUSH_ENABLED → dispatch_lead_push → 查 request_payload。
+
+断言按 jc(Task 6)主线重锚,语义等价:
+- 派生键名: 上游 "区号未确认" → jc "号码补全说明"(号码补全的推送披露口);
+- 条目措辞: 上游 "渠道(标注)" → jc completion_note 的 "原文→e164+标注"
+  (标注常量与 cc_completion 同源,双钉防漂移);
+- NOTE_CUSTOMER 已在 jc 并入 NOTE_UNCONFIRMED(customer_stated 未确认与
+  IP 推测同口径标注,上游的措辞区分被主线统一);
+- confirmed 入参传裸号原文(jc 生产契约: dispatch 收 normalize_contacts
+  的裸号, e164 由 _cc 派生),上游传 e164——canon 口径下两者等价。
+"""
+import asyncio
+
+from sqlalchemy import select
+
+from app.engine.models import (
+    Conversation, CustomerProfile, Evaluation, IntentLevel,
+)
+from app.engine.services.cc_completion import NOTE_MISSING_CC, NOTE_UNCONFIRMED
+
+
+def _cc(confirmed=False, source="ip_proposed", e164="+254712345678"):
+    return {"e164": e164, "intl": "+254 712 345678", "confirmed": confirmed,
+            "attempts": 0, "source": source, "canon": "0712345678"}
+
+
+def _seed(async_db, raw_info):
+    from app.engine.models import ConversationSource
+
+    async def run():
+        async with async_db() as db:
+            conv = Conversation(
+                session_id="s-push-cc", source=ConversationSource.widget,
+                client_info={"ip": "8.8.8.8", "lang": "en"},
+            )
+            db.add(conv)
+            await db.flush()
+            profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info)
+            ev = Evaluation(
+                conversation_id=conv.id,
+                intent_level=IntentLevel.interested, contact_info={})
+            db.add_all([profile, ev])
+            await db.commit()
+    asyncio.run(run())
+
+
+def _dispatch(async_db, confirmed):
+    from app.engine.services import lead_push_service as svc
+
+    async def run():
+        async with async_db() as db:
+            conv = (await db.execute(select(Conversation))).scalar_one()
+            profile = (await db.execute(select(CustomerProfile))).scalar_one()
+            ev = (await db.execute(select(Evaluation))).scalar_one()
+            log = await svc.dispatch_lead_push(
+                db, ev, conv, profile, confirmed, "lead",
+                {"intent_level": "interested"}, True)
+            await db.commit()
+            return log
+    return asyncio.run(run())
+
+
+def test_push_payload_uses_e164_and_filters_underscore(async_db, monkeypatch):
+    """消费 _cc → 过滤 `_`键(顺序钉死 spec §3.5);未确认 IP 推测 → 标注字段。"""
+    from app.engine.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed(async_db, {"目标国家": "肯尼亚", "_cc": {"whatsapp": _cc()},
+                     "_secret": "x"})
+    log = _dispatch(async_db, {"whatsapp": "0712345678"})
+    p = log.request_payload
+    assert p["contact"]["whatsapp"] == "+254712345678"   # 契约值=e164(Task6投影)
+    assert not [k for k in p["profile"] if str(k).startswith("_")]  # _cc/_secret不外泄
+    # 双钉: 与常量同源 + 字节级措辞(改常量必须联动,措辞本身不得漂移)
+    assert p["profile"]["号码补全说明"] == f"0712345678→+254712345678{NOTE_UNCONFIRMED}"
+
+
+def test_push_payload_confirmed_no_note(async_db, monkeypatch):
+    """confirmed=true → 无标注字段;`_` 内部态同样不得外泄。"""
+    from app.engine.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed(async_db, {"目标国家": "肯尼亚",
+                     "_cc": {"whatsapp": _cc(confirmed=True)}})
+    log = _dispatch(async_db, {"whatsapp": "0712345678"})
+    p = log.request_payload
+    assert "号码补全说明" not in p["profile"]
+    assert not [k for k in p["profile"] if str(k).startswith("_")]
+
+
+def test_push_payload_no_ip_note(async_db, monkeypatch):
+    """no_ip(e164=None,值留原裸号) → "缺国家区号"标注;contact 值=原号。
+
+    phone 态显式构造(照上游混合渠道用例的形态): jc 的 cc_state_for 按
+    canon 匹配(canon≠裸号数字一律判 None,旧补全永不跟随新号),故 canon
+    必须与渠道值一致——生产 _cc.canon 恒为客户裸号纯数字。"""
+    from app.engine.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed(async_db, {"目标国家": "肯尼亚",
+                     "_cc": {"phone": {"e164": None, "intl": None,
+                                       "confirmed": False, "attempts": 0,
+                                       "source": "no_ip",
+                                       "canon": "18325698745"}}})
+    log = _dispatch(async_db, {"phone": "18325698745"})
+    p = log.request_payload
+    assert p["profile"]["号码补全说明"] == f"18325698745{NOTE_MISSING_CC}"
+    assert p["contact"]["phone"] == "18325698745"
+
+
+def test_push_payload_customer_stated_note(async_db, monkeypatch):
+    """customer_stated 未确认 → 标注在位(jc 主线: 与 IP 推测同口径
+    NOTE_UNCONFIRMED,上游 NOTE_CUSTOMER 的独立措辞已统一)。"""
+    from app.engine.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed(async_db, {"目标国家": "肯尼亚",
+                     "_cc": {"whatsapp": _cc(source="customer_stated",
+                                             e164="+256712345678")}})
+    log = _dispatch(async_db, {"whatsapp": "0712345678"})
+    p = log.request_payload
+    # 双钉: 与常量同源 + 字节级措辞(改常量必须联动,措辞本身不得漂移)
+    assert p["profile"]["号码补全说明"] == f"0712345678→+256712345678{NOTE_UNCONFIRMED}"
+    assert p["contact"]["whatsapp"] == "+256712345678"
+
+
+def test_push_payload_mixed_channels_join(async_db, monkeypatch):
+    """混合渠道: whatsapp 已确认(不标注) + phone no_ip 未确认(标注)——
+    只记未确认渠道,join 格式钉死。"""
+    from app.engine.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed(async_db, {"目标国家": "肯尼亚", "_cc": {
+        "whatsapp": _cc(confirmed=True, e164="+254712345678"),
+        "phone": {"e164": None, "intl": None, "confirmed": False,
+                  "attempts": 0, "source": "no_ip",
+                  "canon": "18325698745"}}})
+    log = _dispatch(async_db, {"whatsapp": "0712345678",
+                               "phone": "18325698745"})
+    p = log.request_payload
+    assert p["contact"]["whatsapp"] == "+254712345678"
+    assert p["contact"]["phone"] == "18325698745"
+    assert p["profile"]["号码补全说明"] == f"18325698745{NOTE_MISSING_CC}"

tests/test_placeholder_static.py

@@ -0,0 +1,70 @@
+"""placeholder 配置静态守卫(同步自上游 ai-kefu tests/test_placeholder_static.py)。
+
+上游断言读取行自带 `|| "Type a message..."` 默认值;jc(Task 10 值管道)
+优先级为 data-* > brand > 中性兜底——默认值挪进 boot()/applyBrand 的推导
+链尾(读取行给默认会让 brand.placeholder 永远无法生效),读取行保持裸读。
+守卫语义不变: 读取-推导-赋值链路在场、默认值不漂移、chat.html 静态兜底
+不误删、写入侧赋值行不被重构删掉。
+"""
+import re
+from pathlib import Path
+
+STATIC = Path(__file__).resolve().parent.parent / "static"
+DEFAULT = "Type a message..."
+
+
+def _read(rel):
+    return (STATIC / rel).read_text(encoding="utf-8")
+
+
+def _widget_chain_default(src):
+    """提取 widget.js boot 的优先级推导链尾默认值(data-* > brand > 默认)。"""
+    m = re.search(
+        r'placeholder\s*=\s*placeholderAttr\s*\|\|\s*b\.placeholder\s*\|\|\s*"([^"]*)"',
+        src)
+    assert m, "widget.js 缺 placeholder 优先级推导链(data-placeholder > brand > 默认)"
+    return m.group(1)
+
+
+def test_widget_reads_data_placeholder_with_default():
+    src = _read("widget.js")
+    assert 'getAttribute("data-placeholder")' in src, "widget.js 缺 data-placeholder 读取"
+    assert _widget_chain_default(src) == DEFAULT, "widget.js 默认值漂移"
+
+
+def test_chat_js_reads_data_placeholder_with_default():
+    src = _read("js/chat.js")
+    assert 'getAttribute("data-placeholder")' in src, "chat.js 缺 data-placeholder 读取"
+    # chat 端链尾为空串 = 无配置不触碰输入框(chat.html 静态兜底持有可见默认值)
+    assert re.search(
+        r'placeholderText\s*=\s*placeholderAttr\s*\|\|\s*b\.placeholder\s*\|\|\s*""',
+        src), "chat.js 缺 placeholder 优先级推导链(data-placeholder > brand > 不动静态值)"
+
+
+def test_both_ends_default_identical():
+    # 两端用户可见默认值: widget 推导链尾 vs chat.html 静态兜底
+    a = _widget_chain_default(_read("widget.js"))
+    b = re.search(r'<input id="input"[^>]*placeholder="([^"]*)"',
+                  _read("chat.html")).group(1)
+    assert a == b == DEFAULT, f"两端默认值漂移: {a!r} vs {b!r}"
+
+
+def test_chat_html_static_fallback_present():
+    assert 'placeholder="Type a message..."' in _read("chat.html"), \
+        "chat.html 静态兜底被误删(禁用 JS 场景退化)"
+
+
+def test_widget_template_default_present():
+    # jc 模板注入变量(esc(placeholder)),默认值字面量由推导链尾持有
+    assert DEFAULT in _read("widget.js"), "widget.js 默认值字面量被误删"
+
+
+def test_both_ends_apply_assignment():
+    """写入侧守卫(防'读属性仍在、赋值被重构删除'的静默失效,上游断言6)。
+
+    chat.js 显式赋值行;widget.js 由模板注入 placeholder 变量(等价写入侧)。
+    """
+    assert "inputEl.placeholder = placeholderText;" in _read("js/chat.js"), \
+        "js/chat.js 缺赋值行 inputEl.placeholder = placeholderText;"
+    assert 'placeholder="\' + esc(placeholder) + \'"' in _read("widget.js"), \
+        "widget.js 缺模板注入 placeholder 变量"

tests/test_prompt_static.py

@@ -0,0 +1,71 @@
+"""提示词静态断言(同步自上游 ai-kefu tests/test_prompt_static.py)。
+
+锚点迁移(Task 6): 上游单副本 app/prompts → seed_profiles/<slug>/prompts
+按租户持有。auto 租户保留上游逐字锚点;jiancai(主线)措辞已演化——例外
+句/回显句式/禁令是其等价形态。两租户同测,任何一侧缺件都红;断言语义
+不变: 例外句在场(教学多点复述)、好例在场、extractor 门控声明在场。
+"""
+from pathlib import Path
+
+import pytest
+
+PROMPTS = Path(__file__).resolve().parent.parent / "seed_profiles"
+
+# 各租户锚点: exception=(例外句, 最少出现次数), good_example 与小写比较
+ANCHORS = {
+    # 主线措辞: "**唯一例外**：系统提供的补全号码原文" + 确认/回显各节的
+    # 机制复述("按系统提醒回显补全后的完整号码请客户确认");好例句式与
+    # wall.json cc_templates.echo 同源("— is that right?")
+    "jiancai": {
+        "exception": ("唯一例外", 1),
+        "exception_extra": ("回显补全后的完整号码", 4),
+        "forbidden": "客户自己的号码也不回显",
+        "good_example": "is that right",
+        "gate": "仅当客户在回应号码确认时填写",
+    },
+    # 上游原句(L47/L67/L127/L170 四处),auto 出厂未随主线演化
+    "auto": {
+        "exception": ("唯一例外:系统提醒提供的补全号", 4),
+        "exception_extra": None,
+        "forbidden": "禁止回显任何其他号码",
+        "good_example": "is this your full number",
+        "gate": "仅当客户消息是对号码确认的回应",
+    },
+}
+
+
+def _read(tenant, name):
+    return (PROMPTS / tenant / "prompts" / name).read_text(encoding="utf-8")
+
+
+@pytest.mark.parametrize("tenant", ["jiancai", "auto"])
+def test_system_md_exception_clauses_present(tenant):
+    src = _read(tenant, "system.md")
+    a = ANCHORS[tenant]
+    needle, floor = a["exception"]
+    assert src.count(needle) >= floor, f"{tenant}: 例外句缺失/少于 {floor} 处"
+    if a["exception_extra"]:
+        needle2, floor2 = a["exception_extra"]
+        assert src.count(needle2) >= floor2, \
+            f"{tenant}: 例外机制复述少于 {floor2} 处"
+
+
+@pytest.mark.parametrize("tenant", ["jiancai", "auto"])
+def test_system_md_echo_examples_present(tenant):
+    src = _read(tenant, "system.md").lower()
+    assert ANCHORS[tenant]["good_example"] in src, \
+        f"{tenant}: 回显好例句式缺失(应与提醒模板同源)"
+
+
+@pytest.mark.parametrize("tenant", ["jiancai", "auto"])
+def test_system_md_forbidden_clause_present(tenant):
+    src = _read(tenant, "system.md")
+    assert ANCHORS[tenant]["forbidden"] in src, \
+        f"{tenant}: 禁止回显其他号码条款缺失"
+
+
+@pytest.mark.parametrize("tenant", ["jiancai", "auto"])
+def test_extractor_md_cc_reply_declared_with_gate(tenant):
+    src = _read(tenant, "extractor.md")
+    assert "cc_reply" in src and "affirm" in src
+    assert ANCHORS[tenant]["gate"] in src, f"{tenant}: 双门控第一道门声明缺失"