fix: 推送画像过滤_前缀内部键(防CRM泄漏)

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

app/services/lead_push_service.py

@@ -86,7 +86,10 @@ async def _build_payload(
 
     raw_info = profile.raw_info or {}
     client_info = conv.client_info or {}
-    profile_out = dict(raw_info)
+    # _ 前缀内部状态(一次性计数等)不进推送画像(spec §4.3——渲染侧
+    # 两处有过滤,推送端曾整体拷贝泄漏)
+    profile_out = {k: v for k, v in raw_info.items()
+                   if not str(k).startswith("_")}
     profile_out["country"] = resolve_country(client_info.get("ip", ""))
     profile_out["source"] = conv.source.value
     if client_info.get("lang"):

tests/test_lead_push_service.py

@@ -173,6 +173,38 @@ def test_dispatch_pending_payload_structure(async_db, monkeypatch):
     assert p["conversation"][0]["at"].endswith("+08:00")
 
 
+def test_build_payload_filters_internal_underscore_keys(async_db, monkeypatch):
+    """内部状态键(_cc_asked/_email_asked 等)不得泄进推送画像(spec §4.3——
+    渲染侧两处有过滤,推送端曾整体拷贝泄漏)。"""
+    from app.services import lead_push_service as svc
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+    _seed_push_ctx(async_db)
+
+    async def run():
+        async with async_db() as db:
+            from app.models.database import Conversation, CustomerProfile, Evaluation
+            conv = (await db.execute(select(Conversation))).scalar_one()
+            profile = (await db.execute(select(CustomerProfile))).scalar_one()
+            ev = (await db.execute(select(Evaluation))).scalar_one()
+            profile.raw_info = {   # 模拟 Task3 写入的 _ 前缀一次性计数键
+                "目标国家": "肯尼亚", "意向品类": "PE 给水管",
+                "_cc_asked": 1, "_email_asked": 2,
+            }
+            await db.commit()
+            log = await svc.dispatch_lead_push(
+                db, ev, conv, profile, {"phone": "+254722111117"},
+                "lead", {"intent_level": "interested"}, True)
+            await db.commit()
+            return log
+    log = asyncio.run(run())
+    assert log is not None and log.status == "pending"
+    profile_out = log.request_payload["profile"]
+    assert not [k for k in profile_out if str(k).startswith("_")]   # 内部键零泄漏
+    assert profile_out["目标国家"] == "肯尼亚"     # 正常键保留
+    assert profile_out["意向品类"] == "PE 给水管"
+    assert profile_out["source"] == "widget"       # country/source 追加逻辑不受影响
+
+
 def test_dispatch_correction_event_rules(async_db, monkeypatch):
     """7.1 保护规则: 同会话有 success 前科 → update;没有 → new。"""
     from app.services import lead_push_service as svc