feat: 国家代码/邮箱双要状态提醒+一次性计数(串行优先级+防自锁禁令改写)

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

tests/test_lead_email_trigger.py

@@ -473,6 +473,8 @@ def test_reminder_boundary_branches_on_contact_state():
     raw = {"联系方式": {"whatsapp": "+254712345678"}}
     text = ConversationManager._build_context_reminder(raw, "多少钱", [])
     assert "职能边界" in text and "禁止再索要联系方式" in text
+    # Task 3 防自锁改写: 本分支尾部补缺失渠道豁免(子串由尾补保留,非 :651 改写行)
+    assert "缺失渠道可按轻补指引提一次" in text
     # 未留联系方式: 引导留资
     text2 = ConversationManager._build_context_reminder({}, "多少钱", [])
     assert "职能边界" in text2 and "禁止再索要联系方式" not in text2
@@ -904,3 +906,97 @@ def test_e2e_no_reasking_after_contact_provided(async_db, monkeypatch):
     # AI 回复里没有索要话术(StrictAI 不索要;此断言守住提醒确实在场,
     # 索要与否由提醒+墙两层守,墙的确定性已由前序任务覆盖)
     asyncio.run(manager.db.close())
+
+
+# ---- Task 3: 缺失渠道一次性轻补提醒(spec §4.2/§4.3/§4.3b) ----
+def test_contact_completion_note_serial_priority():
+    """spec §4.3b: 裸号+无邮箱(主场景)只注入国家代码追问;号码闭环后才可能
+    注入邮箱轻补;各计数消耗后不再注入。"""
+    from app.services.conversation import _contact_completion_note
+    wa_bare = {"whatsapp": "0501234567"}
+    wa_full = {"whatsapp": "+966501234567"}
+    wa_email = {"whatsapp": "+966501234567", "email": "a@b.com"}
+    note, marks = _contact_completion_note({"联系方式": wa_bare}, wa_bare, {})
+    assert "country code" in note and marks == ["_cc_asked"]
+    note, marks = _contact_completion_note(
+        {"联系方式": wa_bare, "_cc_asked": True}, wa_bare, {})
+    assert "email" in note and marks == ["_email_asked"]
+    note, marks = _contact_completion_note({"联系方式": wa_full}, wa_full, {})
+    assert "email" in note and marks == ["_email_asked"]
+    note, marks = _contact_completion_note({"联系方式": wa_email}, wa_email, {})
+    assert note == "" and marks == []
+    note, marks = _contact_completion_note(
+        {"联系方式": {"email": "a@b.com"}}, {"email": "a@b.com"}, {})
+    assert note == "" and marks == []
+    note, marks = _contact_completion_note(
+        {"联系方式": wa_bare, "_cc_asked": True, "_email_asked": True}, wa_bare, {})
+    assert note == "" and marks == []
+    # lang="zh": 注入中文版逐字模板句原文
+    note, marks = _contact_completion_note(
+        {"联系方式": wa_bare}, wa_bare, {}, lang="zh")
+    assert "带国家代码的完整号码" in note and marks == ["_cc_asked"]
+    note, marks = _contact_completion_note(
+        {"联系方式": wa_full}, wa_full, {}, lang="zh")
+    assert "邮箱也留一个呗" in note and marks == ["_email_asked"]
+
+
+def test_contact_completion_two_round_serial_and_persistence(async_db, monkeypatch):
+    """spec §6-6 计数持久化集成: 裸号两轮——第一轮 cc 追问注入且 _cc_asked 落库;
+    第二轮客户没配合(号码仍裸),按 §4.3b 串行规则转邮箱轻补并落 _email_asked。
+    防"写回失效退化为每轮都问"与"写了计数没拼进 reminder"两类事故。"""
+    class CtxAI(FakeAI):
+        def __init__(self, extract):
+            super().__init__(extract=extract)
+            self.contexts = []
+
+        async def stream_chat(self, messages, context=""):
+            self.contexts.append(context)
+            async for t in super().stream_chat(messages, context):
+                yield t
+
+    ai = CtxAI(extract={})   # extractor 失明,联系方式全靠同轮正则扫描进画像
+    (manager, conv), _ = _setup(async_db, monkeypatch, ai=ai,
+                                source=ConversationSource.standalone)
+    asyncio.run(_send(manager, conv.session_id, "my whatsapp is 0501234567"))
+    # 第一轮正向断言: cc 追问句原文确实拼进了 reminder(非仅写计数)
+    assert "Could you send the full number with the country code" in ai.contexts[0]
+    assert "email address would help too" not in ai.contexts[0]   # 串行: cc 优先
+    profile = asyncio.run(manager._get_profile(conv.id))
+    assert (profile.raw_info or {}).get("_cc_asked") is True      # 计数已落库
+    # 第二轮: 号码仍裸 → cc 不再问,串行规则转邮箱轻补(_cc_asked 已消耗)
+    asyncio.run(_send(manager, conv.session_id, "so how much is the pvc pipe?"))
+    assert "Could you send the full number" not in ai.contexts[1]
+    assert "email address would help too" in ai.contexts[1]
+    profile = asyncio.run(manager._get_profile(conv.id))
+    assert (profile.raw_info or {}).get("_email_asked") is True
+    asyncio.run(manager.db.close())
+
+
+def test_contact_completion_marks_filtered_from_push(async_db, monkeypatch):
+    """Task 2 过滤 × Task 3 真实计数键: 流程写入的 _cc_asked 穿过 dispatch 后,
+    推送 payload 画像无任何 _ 前缀键(不泄漏内部状态进生产 CRM)。"""
+    from app.services import lead_push_service as svc
+
+    (manager, conv), _ = _setup(async_db, monkeypatch, ai=FakeAI(extract={}))
+    asyncio.run(_send(manager, conv.session_id, "my whatsapp is 0501234567"))
+    profile = asyncio.run(manager._get_profile(conv.id))
+    assert (profile.raw_info or {}).get("_cc_asked") is True   # 前提: 计数确已落库
+
+    # 流程跑完才开推送开关——避免 fire_pending_pushes 在流程内派生后台 HTTP 任务
+    monkeypatch.setattr(svc.settings, "PUSH_ENABLED", True)
+
+    async def push():
+        ev = (await manager.db.execute(
+            select(Evaluation).where(Evaluation.conversation_id == conv.id))
+        ).scalar_one()
+        log = await svc.dispatch_lead_push(
+            manager.db, ev, conv, profile, {"whatsapp": "0501234567"},
+            "lead", {"intent_level": "interested"}, True)
+        await manager.db.commit()
+        return log
+
+    log = asyncio.run(push())
+    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("_")]   # 内部键零泄漏
+    asyncio.run(manager.db.close())