ai-kefu-admin/app/engine/services/conversation.py

import logging
import re
import uuid
from collections.abc import AsyncGenerator
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.engine.models import (
    Conversation, ConversationStatus, ConversationSource,
    Message, MessageRole, CustomerProfile, Evaluation, IntentLevel,
    LeadPushLog,
)
from app.engine.services import (
    cc_completion, email_service, ip_service, lead_push_service,
)
from app.engine.services.ai_service import AIService
from app.engine.services.boundary_filter import (
    DEFAULT_WALL, asks_contact, deflection_for, no_contact_value, pivot_for,
    split_sentences, violates_boundary,
)
from app.engine.services.contact_pipeline import (
    canon, digits_only, latest_stated_contacts, merge_contacts,
    normalize_contacts, resolve_by_recency, scan_user_messages, split_validate,
)
from app.engine.wall_schema import CompiledWall

if TYPE_CHECKING:
    from app.platform.registry import TenantRuntime

logger = logging.getLogger(__name__)


# ---- Task 9: contact 行为开关 / context 种子白名单(config → 引擎消费口) ----

def resolve_contact_flags(config_data: dict | None) -> dict:
    """租户 config.contact → 联系方式行为开关。

    cc_completion: 号码区号补全(判定树/确认状态机/CC 强制墙总闸);
    email_light_ask: 邮箱轻补(缺失渠道一次性追问)。False = 该行为整体
    跳过(引擎级门控: 不建状态/不注入/不写计数键)。
    (template_languages 暂无运行期消费方, 不在此解析——Phase 2 后台批次
    再接。)
    """
    contact = ((config_data or {}).get("contact")) or {}
    return {
        "cc_completion": bool(contact.get("cc_completion", True)),
        "email_light_ask": bool(contact.get("email_light_ask", True)),
    }


def context_seed_map(config_data: dict | None) -> dict:
    """context 种子映射 = config.engine.context_seed_keys(ctx_key → raw_key)。

    Task 9 前白名单(product_name/product_id)硬编码在本模块——两租户差异
    (auto 是 vehicle_*)由此配置化;create_conversation 是唯一消费口(按
    映射值写 raw_info,白名单 = 本映射的键集合——审查 Minor-4: 单一取值
    路径,不留第二条内联读法)。
    """
    engine = ((config_data or {}).get("engine")) or {}
    return dict(engine.get("context_seed_keys") or {})

# 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明:
# - 裸 "how long"/"多久" 不收(规格参数类问题常用),英文要求 take/arrive/deliver/ship 邻接,
#   中文只收 到货/多久能到/交货 等明确交付语境
# - 【评审#1】施压话术本身也触发: 磨价轮("just roughly"/"说个大概")没有价格词,
#   却是最易破防的时刻——压力词单列一组
# - 【评审#4】v1 补齐漏词: expensive/fee/贵/价钱/总价/落地/划算/免费(短语式)等;
#   cost/fee 用 \b 边界,costa rica 不再误命中("costa"含cost但无词边界)
# - 原则: 误触发低害(多一条提醒),漏触发高危(红线失守),词表偏宽;上线后漏词案例回流补词
# Task 4 词表数据化: 三组触发正则的数据源 = DEFAULT_WALL(jiancai 出厂)
# .triggers 编译产物(seed_profiles/jiancai/wall.json),上方精修说明随数据留存
# C-1(终审): 本组模块常量降级为"默认墙视图"——供直调/旧测试按名锚定;
# 运行期触发词查找一律走租户墙(manager 的 self._wall.triggers)
BOUNDARY_TRIGGER_RE = DEFAULT_WALL.triggers["boundary"]

# 预算路径路由(评审#9): 命中 BOUNDARY 触发词且含预算词 → 走预算 playbook
BUDGET_TRIGGER_RE = DEFAULT_WALL.triggers["budget"]

# 索要我方联系方式触发(15.2, doctrine v2): 客户问"你们怎么联系/你的微信多少"
# = 最高意向信号。提醒层按 source 分支: web 指按钮, App 无按钮走反向留资
# - 排除第一人称将来时: "I'll call you later" 是日常告别语,不是索要联系方式。
#   必须用定宽 negative lookbehind——lookahead 在动词位置求值,永远看不到
#   前面的 let me / i'll,不生效。四个 lookbehind 全部定宽合法: i'll(5)、
#   i’ll(5, 弯撇 U+2019 手机常见)、i will(7)、let me(7);IGNORECASE 兼容大写
CONTACT_ASK_TRIGGER_RE = DEFAULT_WALL.triggers["contact_ask"]

# 语言跟随提醒(Task 11 真人验收发现): glm-4-flash 对拉丁字符占比高的中文
# 消息(DN800/HDPE 规格混排)会整轮漂成英文,system.md 语言规则压不住。修复位 =
# 当轮 context 提醒(prompt 最末、最新鲜)——与 contact-ask 提醒"逐字话术防
# 漂移"同款方法论。2026-10-08 Task 5 收尾: 语言规则放开(模板句跟随客户语言
# 自然表达)后,ES note 的"整句使用英文原版/绝不翻译成西语"与新规则相反,
# 已同步删除;两条 note 统一补"补全号码逐字照抄"(回显确认轮唯一例外)。
# Task 4 词表数据化: 话术数据源 = DEFAULT_WALL.reminders.language_follow
_CJK_RE = re.compile(r"[一-鿿]")
_ES_FEATURE_RE = re.compile(r"[áéíóúñãç¿¡]")
LANG_NOTE_ZH = DEFAULT_WALL.reminders["language_follow"]["zh"]
LANG_NOTE_ES = DEFAULT_WALL.reminders["language_follow"]["es"]


def _lang_note_for(msg: str, wall: CompiledWall | None = None) -> str:
    # C-1(终审): wall=None = 模块级默认墙(LANG_NOTE_* 常量即该默认视图,
    # 旧测试锚定);manager 一律传租户墙
    w = wall if wall is not None else DEFAULT_WALL
    if _CJK_RE.search(msg or ""):
        return w.reminders["language_follow"]["zh"]
    if _ES_FEATURE_RE.search(msg or ""):
        return w.reminders["language_follow"]["es"]
    return ""


def _cc_lang_of(msg: str) -> str:
    """CC 话术语言路由(P3, 2026-10-08 验收): 汉字→ZH;西语特征(与
    boundary_filter._lang_pool/_lang_note_for 同款重音符粗匹配)→ES;
    否则 EN。只管 ECHO/ASK 模板选型——墙裁决锚(e164 digits)语言无关,
    no_ip 问句短语锚另含西语同词(见 cc_wall 裁决段)。"""
    if _CJK_RE.search(msg or ""):
        return "zh"
    if _ES_FEATURE_RE.search(msg or ""):
        return "es"
    return "en"


# Task 3(spec §4.1-2/§4.4, 2026-10-08): 缺失渠道轻补逐字模板——与 system.md
# 联系方式引导策略节同源。注入的是"缺失渠道"(客户没给过的),与再索要墙的
# 渠道精确化(Task 1)配套: 墙只拦已在手渠道的索要,这些话术恰好问缺失渠道。
# CC 强制墙(2026-10-08 增补设计 §2)复用同源正文: _CC_ASK_* 是追问句正文;
# _CC_WALL_* 是墙的整段替换句(EN 加 "Got it — " 衔接头;ZH 与正文同句)。
# 2026-10-08 判定树设计 Task 3: cc 轻补注入分支已删(判定树接管),_CC_NOTE_*
# 不再运行时注入——常量保留作 no_ip 路径/Task 4 回显模板的正文基准
# (test_boundary_filter / test_system_prompt_dual_contact 仍锚定其正文)。
# Task 4 词表数据化: CC/邮箱模板数据源 = DEFAULT_WALL.cc_templates 与
# .reminders.email_light(seed_profiles/jiancai/wall.json);_CC_WALL_*/_CC_NOTE_*
# 的静态组合("Got it — " 前缀/外壳+正文)在数据文件里存整句,值逐字等价。
_CC_ASK_EN = DEFAULT_WALL.cc_templates["ask"]["en"]
_CC_ASK_ZH = DEFAULT_WALL.cc_templates["ask"]["zh"]
# P3(2026-10-08 真实验收): 西语轮模板补齐——修复前 ECHO/ASK 只有 EN/ZH,
# 西语轮 GLM 照抄英文。WALL 沿用 EN 形态(衔接头 + 追问句正文)。
_CC_ASK_ES = DEFAULT_WALL.cc_templates["ask"]["es"]
_CC_WALL_EN = DEFAULT_WALL.cc_templates["wall"]["en"]
_CC_WALL_ZH = DEFAULT_WALL.cc_templates["wall"]["zh"]
_CC_WALL_ES = DEFAULT_WALL.cc_templates["wall"]["es"]
_CC_NOTE_EN = DEFAULT_WALL.cc_templates["note"]["en"]
_CC_NOTE_ZH = DEFAULT_WALL.cc_templates["note"]["zh"]
# Task 4(spec §4 回显确认轮/§4b): 判定树给出 e164 建议的等待态轮用回显确认
# 模板——{intl} 由 Python format_intl 算好填入(reminder/墙替换/测试断言三方
# 同源引用本常量),GLM 只逐字复述不拼号。no_ip 轮(无可回显号)仍走 _CC_ASK_*
# (上方保留不拆),两条路径按 _cc.whatsapp 的 e164 在场分流。
_CC_ECHO_EN = DEFAULT_WALL.cc_templates["echo"]["en"]
_CC_ECHO_ZH = DEFAULT_WALL.cc_templates["echo"]["zh"]
_CC_ECHO_ES = DEFAULT_WALL.cc_templates["echo"]["es"]
_EMAIL_NOTE_EN = DEFAULT_WALL.reminders["email_light"]["en"]
_EMAIL_NOTE_ZH = DEFAULT_WALL.reminders["email_light"]["zh"]


def _cc_intl(wa_state: dict | None) -> str:
    """等待态的分组展示号: intl 键缺失时按 e164 防御重算(存量/迁移数据)。"""
    e164 = str((wa_state or {}).get("e164") or "")
    if not e164:
        return ""
    return str((wa_state or {}).get("intl") or cc_completion.format_intl(e164))


def _cc_script(wa_state: dict | None, lang: str,
               wall: CompiledWall | None = None) -> str:
    """CC 等待态本轮逐字话术(Task 4, spec §4;P3 扩 ES): 有 e164 → 回显
    确认句(intl 由 Python 算好填入,GLM 不拼号);no_ip → 追问句
    (_CC_ASK_* 服务)。reminder 注入/墙替换/测试断言三方同源,统一走本函数。
    C-1(终审): wall=None = 模块级默认墙(_CC_ECHO_*/_CC_ASK_* 常量即该
    默认视图);manager 一律传租户墙。"""
    w = wall if wall is not None else DEFAULT_WALL
    if str((wa_state or {}).get("e164") or ""):
        intl = _cc_intl(wa_state)
        echo = w.cc_templates["echo"]
        return {"zh": echo["zh"], "es": echo["es"]}.get(
            lang, echo["en"]).format(intl=intl)
    ask = w.cc_templates["ask"]
    return {"zh": ask["zh"], "es": ask["es"]}.get(lang, ask["en"])


def _cc_wall_template(wa_state: dict | None, lang: str,
                      wall: CompiledWall | None = None) -> str:
    """CC 触发轮墙整段替换句: 回显轮 = ECHO 模板(自带 "Got it —" 衔接头);
    no_ip 轮 = _CC_WALL_*(ASK 追问句 + 衔接头,沿用 Task 3 形态;P3 扩 ES)。
    C-1(终审): wall=None = 模块级默认墙(_CC_WALL_* 常量即该默认视图);
    manager 一律传租户墙。"""
    if str((wa_state or {}).get("e164") or ""):
        return _cc_script(wa_state, lang, wall=wall)
    w = wall if wall is not None else DEFAULT_WALL
    tmpl = w.cc_templates["wall"]
    return {"zh": tmpl["zh"], "es": tmpl["es"]}.get(lang, tmpl["en"])


def _cc_confirm_note(wa_state: dict | None, lang: str,
                     wall: CompiledWall | None = None) -> str:
    """cc 等待态确认句注入(Task 4, spec §4/§4b): cc_due 才注入,与邮箱轻补
    cc_settled 串行互斥(两个"逐字使用这句"不同轮竞争)。外壳仿轻补指引:
    ⚠️ 系统指令声明 + 逐字话术 + 客户答复后致谢不追问。回显分支带零号码
    唯一例外(4b-3 让位语义重建: 仅系统提供的补全号原文可写出)。
    Task 4 词表数据化: 指引话术数据源 = 墙 reminders.cc_guide
    ({intl}/{cc_script} 占位由本函数 .format() 回填,值逐字等价)。
    C-1(终审): wall=None = 模块级默认墙;manager 一律传租户墙。"""
    if not wa_state or not cc_completion.cc_due(wa_state):
        return ""
    w = wall if wall is not None else DEFAULT_WALL
    if str(wa_state.get("e164") or ""):
        intl = _cc_intl(wa_state)
        return w.reminders["cc_guide"]["echo"].format(
            intl=intl, cc_script=_cc_script(wa_state, lang, wall=wall))
    return w.reminders["cc_guide"]["ask"].format(
        cc_script=_cc_script(wa_state, lang, wall=wall))


def _contact_completion_note(
    raw_info: dict | None, confirmed: dict | None, suspect: dict | None,
    lang: str = "en", email_ask: bool = True,
    wall: CompiledWall | None = None,
) -> tuple[str, list[str]]:
    """邮箱轻补提醒(spec §4.2/§4.3b,2026-10-08 判定树设计收编)。
    纯函数,返回(注入文案,计数键)。

    cc 追问分支已删除——号码区号补全由轮前判定树钩子接管(_cc 状态机,
    _cc_pre_turn),确认提醒由 cc_due 驱动(Task 4 接)。本函数只管邮箱轻补,
    且串行解锁条件改为 cc_settled(_cc.whatsapp): 确认流未落定时注入邮箱
    轻补会让两个"逐字使用这句"同轮竞争(旧 §4.3b 场景①教训);落定即解锁。
    带 + 号客户无 _cc 状态 → cc_settled(None)=True,旧轻补行为不回归。
    suspect 仅为与调用处画像判定同签名,规则只看 confirmed(spec §4.2)。
    email_ask=False(Task 9: contact.email_light_ask 开关)→ 恒空,不注入
    也不产计数键。
    C-1(终审): wall=None = 模块级默认墙(_EMAIL_NOTE_* 常量即该默认视图);
    manager 一律传租户墙(流式调用处与本函数在 reminder 内的调用同参,
    写回与注入永不漂移)。
    """
    if not email_ask:
        return "", []
    info = raw_info or {}
    numbers = [str(v).strip() for k, v in (confirmed or {}).items()
               if k in ("whatsapp", "phone") and v]
    if not numbers:
        return "", []
    if "email" in (confirmed or {}):
        return "", []
    if info.get("_email_asked"):
        return "", []
    if not cc_completion.cc_settled((info.get("_cc") or {}).get("whatsapp")):
        return "", []   # whatsapp 确认流未落定: 邮箱轻补串行等待
    w = wall if wall is not None else DEFAULT_WALL
    return ((w.reminders["email_light"]["zh"] if lang == "zh"
             else w.reminders["email_light"]["en"]), ["_email_asked"])


def _cc_pre_turn(raw_info: dict, confirmed: dict, ip_iso: str | None) -> bool:
    """轮前判定树钩子(spec §6-1/§3/§4): 对 confirmed 号码建/重置/迁移 _cc。

    - 多渠道策略: 仅 whatsapp 组走确认流;phone 与 whatsapp 同 canon 时共享
      whatsapp 组状态(不建独立键),异号时 phone 组只记静默补全态
      (ip_proposed/naive/no_ip,不进确认流)——确认流的消费与门控
      (cc_due/cc_settled/轮末 transit)一律只看 _cc.whatsapp,phone 态
      仅供出站渲染(Task 6)。
    - canon 变化即重置重走判定树(旧补全永不跟随新号);同 canon 保留原状态
      (attempts 只被 canon 变化重置,spec §4)。
    - 显式 +/00 号码不建状态(无态=天然落定);prefix_normalized 静默规范化
      态(confirmed=true)照写,供出站渲染无标注输出。
    - 存量迁移(spec §4 尾): 旧 _cc_asked=True 且 whatsapp 裸号尚无 _cc 组 →
      一次性落 fallback 态(已问过不再追问,直通邮箱轻补);判定树算出的
      e164 建议保留(首封可拨)。_cc 组一旦存在即不再触发。
    就地更新 raw_info["_cc"],返回是否有变更(调用方与计数写回同事务落库)。
    """
    before = raw_info.get("_cc")
    cc = dict(before or {})
    migrate = bool(raw_info.get("_cc_asked"))
    wa = str(confirmed.get("whatsapp") or "").strip()
    ph = str(confirmed.get("phone") or "").strip()

    def _refresh(channel: str, value: str) -> None:
        existing = cc.get(channel)
        if existing is not None \
                and str(existing.get("canon") or "") == digits_only(value):
            return   # 同 canon: 保留原状态(attempts 不重置)
        kind, state = cc_completion.classify(value, ip_iso)
        if state is None:
            cc.pop(channel, None)   # 显式 +/00(或防御性垃圾输入): 无态即落定
            return
        st = dict(state)
        if (migrate and channel == "whatsapp" and existing is None
                and kind == cc_completion.NEED_CC):
            st["confirmed"] = False
            st["source"] = "fallback"   # 存量迁移: 已问过,不再追问
            st.pop("suggest_ask_full", None)
        cc[channel] = st

    if wa:
        _refresh("whatsapp", wa)
    else:
        cc.pop("whatsapp", None)   # 号码不再在手: 状态清,防残留跟随
    if ph:
        wa_state = cc.get("whatsapp")
        if wa_state is not None \
                and str(wa_state.get("canon") or "") == digits_only(ph):
            cc.pop("phone", None)   # 同 canon 组归并: phone 共享 whatsapp 状态
        else:
            _refresh("phone", ph)
    else:
        cc.pop("phone", None)

    if cc:
        raw_info["_cc"] = cc
    else:
        raw_info.pop("_cc", None)
    return raw_info.get("_cc") != before


# Task 6(spec §5 面板同步): 等待确认期间的联系方式值尾注——防"邮件带补全、
# 后台显裸号"的分裂。尾注全为非数字字符,canon() 纯数字提取不受影响
# (24h 去重/更正判定照常)。
_CC_PENDING_TAIL = "（待确认补全号）"


def _panel_contact_info(confirmed: dict, cc_map: dict | None) -> dict:
    """ev.contact_info 存储视图(Task 6 评审钉死): 裸号原文 + 等待态尾注,
    e164 走派生不落本体——e164 替换本体会让"存储值 vs 每轮裸号确认值"
    永远不同 canon,退化成每轮误判换号→更正连发/回访退化。
    只标 whatsapp 确认流覆盖的号码(canon 与 _cc.whatsapp 一致,含同号
    phone 镜像): phone 独立静默态不进确认流、attempts 永不推进,cc_due
    恒真,若照标会永久滞留且语义失真。"""
    wa = (cc_map or {}).get("whatsapp")
    wa_canon = str((wa or {}).get("canon") or "")
    pending = bool(wa_canon) and cc_completion.cc_due(wa)
    out: dict = {}
    for k, v in (confirmed or {}).items():
        if (pending and k in ("phone", "whatsapp")
                and digits_only(str(v)) == wa_canon):
            out[k] = f"{v}{_CC_PENDING_TAIL}"
        else:
            out[k] = v
    return out


def _channel_canons(confirmed: dict, cc_map: dict | None) -> dict[str, set[str]]:
    """canon 口径双保险(Task 6 评审钉死): 每渠道候选集 = 裸号 canon ∪
    _cc.e164 的 canon——同号换形态(裸号↔+前缀完整形态)不误判换号
    (更正连发)、回访不退化成全量询盘。只扩"新值侧": 若把当前 _cc 的
    e164 canon 并入旧值/历史集,换号后新号恰为前缀完整形态时会自我满足,
    真更正被吞。"""
    out: dict[str, set[str]] = {}
    for k, v in (confirmed or {}).items():
        cands = {canon(str(v))}
        state = cc_completion.cc_state_for(cc_map, k, str(v))
        e164 = str((state or {}).get("e164") or "")
        if e164:
            cands.add(canon(e164))
        out[k] = cands
    return out


# P4'(2026-10-08 复验): 画像"目标国家"报国信号的占位值集——"无"/"未提及"
# 是写给 extractor 的指令标记(extractor.md 明令禁止作字段值输出,但 GLM
# 不保证),"未知"/"本地" 同为无信息量形态;命中占位集不当作报国消费。
_COUNTRY_PLACEHOLDERS = {"无", "未提及", "未知", "本地"}


def _country_signal(cur, pre) -> str | None:
    """P4' 画像报国信号(纯函数): 等待态轮画像"目标国家"键**新出现或值
    变化**(cur=本轮合并后值, pre=轮前快照)且值有效(非空、非占位集)
    → 返回该值作为 country 信号;值不变(含轮前已有同值)→ None(去抖:
    同国重算幂等,重复陈述无新信息)。国名是否可识别(NAME_TO_REGION
    命中)由调用方判定——不命中时信号整体忽略,不消费。"""
    c = str(cur or "").strip()
    if not c or c in _COUNTRY_PLACEHOLDERS:
        return None
    if c == str(pre or "").strip():
        return None
    return c


# 方案B(13.6): 迂回话术轮换序号——连续破防时不至于复读同一句
_boundary_seq = 0


def _boundary_rotate() -> int:
    global _boundary_seq
    _boundary_seq += 1
    return _boundary_seq


class ConversationManager:
    def __init__(self, db: AsyncSession, *,
                 runtime: "TenantRuntime",
                 ai_service: AIService | None = None):
        self.db = db
        # Task 6: runtime 必填(Task 5b 质量审查关门建议)——提示词/墙/会话
        # 工厂全部租户化,无租户上下文的构造不再合法;默认 AI 服务用该租户
        # 的 PromptProvider + compiled_wall 组装($DEFLECTION_POOL 单一事实源
        # 注入)。Task 9: AI 凭据/config 全部经 runtime.config.data 接线。
        self.runtime = runtime
        # C-1(终审审查): 执行侧墙接线——过滤/替换/提醒全走租户编译墙,
        # 与 AIService 的 $DEFLECTION_POOL 同源(runtime.compiled_wall);
        # 模块级 DEFAULT_WALL 仅供无租户上下文的直调默认(boundary_filter
        # 出厂视图 + 本模块旧测试锚定),会话流内一律本墙。
        self._wall = runtime.compiled_wall
        self.ai = ai_service or AIService(
            prompts=runtime.prompts,
            wall=runtime.compiled_wall,
            config_data=runtime.config.data)
        # Task 9: contact 行为开关(cc 补全/邮箱轻补)与国名映射扩表——
        # manager 生命周期内配置不变(reload 换 runtime 即换 manager 装配线;
        # 生效时机: WS 长连接内冻结至断连,REST 路径下一请求生效——Phase 2
        # 联调改配置后按此预期,勿误判"没生效")
        self._contact_flags = resolve_contact_flags(runtime.config.data)
        self._region_extras = (((runtime.config.data.get("misc")) or {})
                               .get("name_to_region_extras") or None)

    @property
    def session_factory(self):
        """推送后台任务写结果用的工厂(写回派发租户的库);runtime 未懒建
        engine 时 None(仅测试直连场景——生产装配线总在 ensure_ready 之后)。"""
        return self.runtime.session_factory

    async def create_conversation(
        self,
        source: ConversationSource = ConversationSource.standalone,
        client_info: dict | None = None,
        page_token: str | None = None,
        context: dict | None = None,
    ) -> Conversation:
        session_id = uuid.uuid4().hex
        conv = Conversation(
            session_id=session_id,
            source=source,
            client_info=client_info,
            page_token=page_token,
        )
        self.db.add(conv)
        await self.db.commit()
        await self.db.refresh(conv)

        # Conditional whitelist seed of app context into raw_info (red line):
        # only when `context` is provided, and only keys in the tenant's
        # engine.context_seed_keys map (jiancai: product_*; auto: vehicle_* —
        # Task 9 前白名单硬编码 product_name/product_id,现读租户配置).
        # WS path passes no context -> raw_info stays {}.
        raw_info: dict = {}
        if context:
            for ctx_key, raw_key in context_seed_map(
                    self.runtime.config.data).items():
                if context.get(ctx_key):
                    raw_info[raw_key] = context[ctx_key]
        profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info)
        self.db.add(profile)
        await self.db.commit()

        return conv

    async def get_by_session(self, session_id: str) -> Conversation | None:
        result = await self.db.execute(
            select(Conversation).where(Conversation.session_id == session_id)
        )
        return result.scalar_one_or_none()

    async def get_by_page_token(self, page_token: str) -> Conversation | None:
        result = await self.db.execute(
            select(Conversation).where(
                Conversation.page_token == page_token,
                Conversation.status == ConversationStatus.active,
            )
        )
        return result.scalar_one_or_none()

    async def stream_process_message(
        self, session_id: str, content: str
    ) -> AsyncGenerator[dict, None]:
        """Stream AI reply token by token, then do extract/evaluate."""
        conv = await self.get_by_session(session_id)
        if not conv:
            yield {"type": "error", "message": "session not found"}
            return

        if conv.status != ConversationStatus.active:
            yield {"type": "error", "message": "conversation ended"}
            return

        # Save user message
        user_msg = Message(
            conversation_id=conv.id, role=MessageRole.user, content=content
        )
        self.db.add(user_msg)
        await self.db.commit()

        history = await self._load_history(conv.id)

        # Build context reminder (merged into system prompt)
        profile = await self._get_profile(conv.id)

        # P0 同轮竞态消除: 当前消息即时正则扫描——仅并入本轮内存视图
        # (提醒/话术路由/轮前 cc 判定树),不落库;持久化仍由轮末
        # extractor+scanner 负责,邮件触发时序不变。
        # 合并方向(Task 3 修正): 当前消息捕获覆盖存量——客户当轮改号时,
        # 同轮视图必须看到新号(轮前判定树 canon 变化即重置依赖此方向;
        # 与轮末 Task 12 "最新显式自述优先"同语义)。存量只在当前消息
        # 未提及的渠道上补位。
        current_scan = scan_user_messages([{"role": "user", "content": content}])
        effective_raw = dict(profile.raw_info or {})
        if current_scan:
            effective_raw["联系方式"] = merge_contacts(
                effective_raw.get("联系方式"), current_scan)

        # P4'(2026-10-08 复验) 轮前画像快照: "目标国家"轮末信号判定基线。
        # 必须在本轮任何 raw_info 写回发生前取值(下方计数写回/轮末 extractor
        # 合并都会覆盖同一 ORM 实例);值为 str/None 不可变,持引用即安全。
        pre_turn_country = (profile.raw_info or {}).get("目标国家")

        # 联系方式视图(轮前钩子/计数写回/再索要墙区共用;原在 reminder 之后
        # 计算,前移只为喂轮前钩子,输入相同取值不变)
        _contacts_now = normalize_contacts(effective_raw.get("联系方式"))
        _confirmed_now, _suspect_now = split_validate(_contacts_now)

        # Task 3 轮前判定树钩子(spec §6-1;评审 Step 3b 钉死顺序: 必须在
        # _build_context_reminder 之前)——即时扫描并入裸号后同步跑判定树
        # (纯计算毫秒级)建/重置/迁移 _cc,当轮即可被墙门控/轮末消费/出站
        # 渲染读到;落库与下方一次性计数写回同事务。
        # Task 9: contact.cc_completion=False → 判定树整体跳过(不建不迁
        # _cc;存量 _cc 冻结为只读出站渲染,不再推进)。
        _ip_iso = ip_service.resolve_region_code(
            (conv.client_info or {}).get("ip"))
        _cc_changed = (_cc_pre_turn(effective_raw, _confirmed_now, _ip_iso)
                       if self._contact_flags["cc_completion"] else False)
        _wa_cc = (effective_raw.get("_cc") or {}).get("whatsapp")

        context_reminder = self._build_context_reminder(
            effective_raw, content, history, source=conv.source.value,
            contact_flags=self._contact_flags, wall=self._wall)
        if self._wall.triggers["contact_ask"].search(content or ""):
            logger.info("Contact-ask reminder fired (source=%s)", conv.source.value)

        # Task 3(spec §4.3/§6-1): 一次性计数写回 + _cc 状态同事务落库——
        # _build_context_reminder 是 staticmethod 收内存副本,计数键/状态须由
        # 调用方在流式开始前显式落库;轮末 extractor 合并起点是
        # dict(profile.raw_info or {}),提前提交可存活。与 reminder 内部同参
        # 调用(上方纯函数),保证写回与注入永不漂移;条件扩为"计数或 _cc
        # 变更任一存在"(评审 Step 3b)。
        _, completion_marks = _contact_completion_note(
            effective_raw, _confirmed_now, _suspect_now,
            "zh" if _CJK_RE.search(content or "") else "en",
            email_ask=self._contact_flags["email_light_ask"], wall=self._wall)
        if completion_marks or _cc_changed:
            raw = dict(profile.raw_info or {})
            for key in completion_marks:
                raw[key] = True
            if _cc_changed:
                if effective_raw.get("_cc"):
                    raw["_cc"] = effective_raw["_cc"]
                else:
                    raw.pop("_cc", None)
            profile.raw_info = raw
            self.db.add(profile)
            await self.db.commit()
            # warning 级(2026-10-08 第四轮验收): 生产曾现 2/5 裸号轮墙未介入的
            # 非确定性异常(同输入容器内链路复现正常)——本行生产可见,异常会话
            # 可立刻分辨"计数/状态/注入是否发生",供观察期定位
            logger.warning("Contact completion state persisted (conv=%s): "
                           "marks=%s cc_changed=%s",
                           conv.id, completion_marks, _cc_changed)

        # Enrich short user messages with context so AI doesn't lose track
        history_for_ai = list(history)
        enriched = self._enrich_short_message(content, profile.raw_info or {}, history)
        if history_for_ai and history_for_ai[-1]["role"] == "user" and enriched != content:
            history_for_ai[-1] = {**history_for_ai[-1], "content": enriched}
            logger.info("Enriched short message: %r -> %r", content[:50], enriched[:100])

        # 方案B'(13.7): 句子粒度门控——干净句子逐句流出(恢复打字机),
        # 违规句永不流出:已流出内容全部干净,违规时接过渡话术收尾并掐断生成。
        # F6(D6联动): 客户已留联系方式时,收尾话术不得再索要——路由 contact 版
        # 渠道精确化(2026-10-08): confirmed 渠道集合传给再索要墙——
        # 索要缺失渠道(如已有 phone 问邮箱)放行,是 Task 3 邮箱轻补的前提
        # (confirmed/suspect 已在 Task 3 写回处算好,此处复用)
        has_contact = bool(_confirmed_now)
        _confirmed_channels = set(_confirmed_now)
        # CC 强制墙(2026-10-08 cc 设计): 门控改线判定树——cc_due(_cc.whatsapp)
        # 等待态即触发(旧 `completion_marks == ["_cc_asked"]` 门控随 cc 轻补
        # 分支删除而消亡,_cc_asked 停写)。流式无法撤回 → 触发轮不逐句流出,
        # 整段缓冲到流末统一裁决(缓冲期间价格/零号码/再索要三墙照常逐句
        # 检查,优先级不变)。Task 4(spec §4/§4b): 回显轮(等待态有 e164)裁决
        # 锚 = 注入 e164 的 digits 等值在场,替换句 = ECHO 模板;no_ip 轮
        # (无可回显号)沿用 ASK 模板与 "country code" 问句短语锚。
        # Task 9: cc_completion=False → CC 强制墙同样关闭(开关管整个补全
        # 行为族: 判定树/确认流/墙);存量等待态不触发缓冲整段裁决
        cc_wall = (self._contact_flags["cc_completion"]
                   and cc_completion.cc_due(_wa_cc))
        # Task 4 白名单传参(spec §4): 零号码墙逐段放行系统注入的补全号——
        # 仅回显轮(等待态有 e164)传,其余轮 None 保持原行为(全拦)
        cc_allow_number = ((_wa_cc or {}).get("e164") if cc_wall else None)

        yield {"type": "stream_start"}
        emitted = ""
        pending = ""
        violated = False
        dropped_ask = False   # Task 5b: 本轮是否有索要句被再索要墙丢弃
        dirty = ""
        cc_buffer: list[str] = []   # CC 触发轮: 已过墙句子的延迟流出缓冲
        # 15.3: 双墙裁决——价格墙检查在前: 同句双违规时价格路径话术优先
        # (职能边界话术更具体,且自带留资引导);仅纯号码/邮箱违规时
        # violation=="no_contact",用零号码墙替换句
        violation = ""
        async for token in self.ai.stream_chat(history_for_ai, context=context_reminder):
            pending += token
            complete, pending = split_sentences(pending)
            for sentence in complete:
                if violates_boundary(sentence, wall=self._wall):
                    violated = True
                    dirty = sentence
                    violation = "boundary"
                    break
                if no_contact_value(sentence, allow_number=cc_allow_number,
                                    wall=self._wall):
                    violated = True
                    dirty = sentence
                    violation = "no_contact"
                    break
                # Task 5b 再索要墙: 已留资时索要句静默丢弃(丢句不掐断,
                # 其余句子照常流出——软于两墙的掐断+替换);
                # 渠道精确化: 只丢索要已在手渠道的句子
                if has_contact and asks_contact(sentence, _confirmed_channels,
                                                wall=self._wall):
                    logger.warning("Re-ask wall dropped sentence (conv=%s): %r",
                                   conv.id, sentence[:60])
                    dropped_ask = True
                    continue
                if cc_wall:
                    cc_buffer.append(sentence)   # 墙照检,流出延后到流末
                    continue
                for i in range(0, len(sentence), 40):
                    yield {"type": "stream_chunk", "content": sentence[i:i + 40]}
                emitted += sentence
            if violated:
                break  # 提前掐断生成,省 token
        if not violated and pending:
            if violates_boundary(pending, wall=self._wall):
                violated = True
                dirty = pending
                violation = "boundary"
            elif no_contact_value(pending, allow_number=cc_allow_number,
                                  wall=self._wall):
                violated = True
                dirty = pending
                violation = "no_contact"
            elif has_contact and asks_contact(pending, _confirmed_channels,
                                              wall=self._wall):
                # Task 5b 再索要墙: 尾巴是索要句 → 静默丢弃(不流出不掐断)
                logger.warning("Re-ask wall dropped tail (conv=%s): %r",
                               conv.id, pending[:60])
                dropped_ask = True
            elif cc_wall:
                cc_buffer.append(pending)
            else:
                for i in range(0, len(pending), 40):
                    yield {"type": "stream_chunk", "content": pending[i:i + 40]}
                emitted += pending
        if violated:
            if violation == "no_contact":
                # 零号码墙替换句: 已留资→确认话术(不再索要);未留资按 source 分支
                # (App 无按钮 → 反向留资; web/widget/standalone → 指顶部按钮)
                logger.warning(
                    "No-contact wall replaced sentence (conv=%s): %r",
                    conv.id, dirty[:60])
                if cc_wall:
                    # Important-2(审查修复) + Task 4: CC 触发轮回显被拦 →
                    # closing 改用本轮 CC 模板(回显轮 ECHO/no_ip 轮 ASK)而非
                    # NO_CONTACT_ALREADY——GLM 有 3/3 回显倾向,该路径概率高,
                    # 客户该收到确认/追问而非"稍后联系"虚假承诺;
                    # 价格墙路径(violation=="boundary")不变
                    closing = _cc_wall_template(_wa_cc, _cc_lang_of(content),
                                                wall=self._wall)
                else:
                    # Task 9 收口: 视图直取墙 no_contact(单一事实源),
                    # 不经 boundary_filter 的模块级常量转引;C-1(终审):
                    # 取租户墙——替换文案各自来自自己的池
                    _w = self._wall
                    closing = (_w.no_contact["already"] if has_contact
                               else (_w.no_contact["app"]
                                     if conv.source.value == "app"
                                     else _w.no_contact["web"]))
            else:
                logger.warning(
                    "Boundary filter intercepted sentence (conv=%s): %r | tail=%r",
                    conv.id, dirty[:80], (pending or "")[:40])
                closing = (pivot_for(content, has_contact, wall=self._wall)
                           if emitted
                           else deflection_for(content, _boundary_rotate(),
                                               has_contact=has_contact,
                                               wall=self._wall))
            # Minor-5(审查修复): 已流出前缀与替换句之间补一个空格分隔
            # ("...xxxx? Got it —"——英文衔接无空格会粘连);CC/触发轮缓冲
            # 未流出,emitted 恒空,天然不受影响。空格随 closing 一起流出,
            # 客户所见与落库保持一致
            if emitted and not emitted.endswith((" ", "\n")):
                closing = " " + closing
            for i in range(0, len(closing), 40):
                yield {"type": "stream_chunk", "content": closing[i:i + 40]}
            emitted += closing
        # CC 强制墙裁决(2026-10-08 cc 设计;Task 4 改造 spec §4b-4): 触发轮
        # 缓冲全文在此判定——回显轮(等待态有 e164)裁决锚 = **注入 e164 的
        # digits 等值在场**(GLM 任意分组形态 "+966 50 123 4567"/"+966501234567"
        # 均判在场;辅锚 "is that right"/"对吗" 随模板正文,不作独立判据——
        # 号码在场才是回显的承重语义);no_ip 轮无可回显号,沿用问句短语锚
        # ("country code"/"国家代码"/"código de país",与 _CC_ASK_* 模板句
        # 同词——P3 扩西语同词,西语服从句不再被误替换)。在场 → GLM
        # 自发服从,缓冲原样流出零干预;不在场 → 整段替换为对应 ECHO/ASK
        # 模板。violated(价格/零号码墙)优先于本墙——沿用双墙裁决"更具体
        # 话术优先"惯例;_cc 状态已在流前落库,不受替换影响。缓冲轮全部
        # 实义句被再索要墙丢弃时 cc_buffer 为空 → 同样走替换(模板句即
        # 本轮唯一实义输出,天然不触发下方兜底)。
        if cc_wall and not violated:
            cc_full = "".join(cc_buffer)
            cc_lang = _cc_lang_of(content)
            wa_e164 = str((_wa_cc or {}).get("e164") or "")
            if wa_e164:
                obeyed = digits_only(wa_e164) in digits_only(cc_full)
            else:
                obeyed = ("country code" in cc_full.lower()
                          or "国家代码" in cc_full
                          or "código de país" in cc_full.lower())
            if obeyed:
                for i in range(0, len(cc_full), 40):
                    yield {"type": "stream_chunk", "content": cc_full[i:i + 40]}
                emitted = cc_full
            else:
                logger.warning(
                    "CC enforcement wall replaced reply (conv=%s): %r",
                    conv.id, cc_full[:60])
                cc_template = _cc_wall_template(_wa_cc, cc_lang,
                                                wall=self._wall)
                for i in range(0, len(cc_template), 40):
                    yield {"type": "stream_chunk",
                           "content": cc_template[i:i + 40]}
                emitted = cc_template
        # Task 5b 评审修正: 整条回复只有索要句(全被丢)→ 兜底流出"经理主动
        # 联系"句,绝不空回复。与上方墙收尾互斥——violated 路径已自带收尾,
        # 此处仅在无任何实义内容流出时接管;已流出的纯空白前缀一并作废,
        # 落库恒为完整兜底句(客户看到的比一个空格更糟的是: 什么都没有)。
        if not violated and dropped_ask and not emitted.strip():
            logger.warning("Re-ask wall dropped entire reply (conv=%s), "
                           "falling back to closing", conv.id)
            fallback = self._wall.no_contact["already"]
            for i in range(0, len(fallback), 40):
                yield {"type": "stream_chunk",
                       "content": fallback[i:i + 40]}
            emitted = fallback
        yield {"type": "stream_end"}
        ai_reply = emitted  # 落库与流出统一为已流出文本(下方持久化)

        # Save assistant message (skip if empty)
        if ai_reply.strip():
            assistant_msg = Message(
                conversation_id=conv.id, role=MessageRole.assistant, content=ai_reply
            )
            self.db.add(assistant_msg)
            await self.db.commit()

        # Round 2: Extract info (background)
        full_history = history + [{"role": "assistant", "content": ai_reply}]
        profile = await self._get_profile(conv.id)

        # Task 12b: 轮前联系方式快照(extractor 深合并 + 覆盖 + 同号同步的比对基线)
        before_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式"))

        # _ 前缀内部键(_cc 状态机/_email_asked 计数)不进 LLM 输入(Task 3 审查
        # Minor 3 防御纵深——渲染/推送侧已过滤,此处防提示词侧意外回显)
        new_info = await self.ai.extract_info(
            full_history,
            existing_info={k: v for k, v in (profile.raw_info or {}).items()
                           if not str(k).startswith("_")},
        )
        logger.info("Extractor result: %s", new_info)
        # 出口卫生(spec §4 三封堵之②/③): cc_reply 是 extractor 的单轮瞬态
        # 分类,提取后即剥离——绝不进 raw_info 持久层(消费点在轮末钩子
        # _cc_round_end);同时防御性丢弃 extractor 回传的任何 `_` 前缀键,
        # 防 LLM 回显 _cc 覆盖 Python 管理的状态机。(入口侧封堵①: 上方
        # existing_info 已滤 `_`,复核通过)
        cc_reply = None
        if isinstance(new_info, dict):
            cc_reply = new_info.pop("cc_reply", None)
            stripped = {k: v for k, v in new_info.items()
                        if not str(k).startswith("_")}
            if len(stripped) != len(new_info):
                logger.warning(
                    "Dropped underscore keys from extractor output (conv=%s): %s",
                    conv.id, [k for k in new_info if str(k).startswith("_")])
            new_info = stripped
        if new_info:
            raw = dict(profile.raw_info or {})
            for k, v in new_info.items():
                # 联系方式深合并(同key新值覆盖),其余顶层覆盖 —— 堵浅合并抹数据
                if k == "联系方式":
                    raw[k] = merge_contacts(raw.get(k), v)
                else:
                    raw[k] = v
            profile.raw_info = raw
            self.db.add(profile)

        # Round 2.5: 联系方式正则辅源(S4: 扫全量客户消息历史)
        # 🔴 合并方向: 扫描器只填空,extractor 已有值优先——否则扫描器的
        # 首个历史值会压掉 extractor 后续捕获的改号, D4 更正邮件失效
        scanned = scan_user_messages(full_history)
        if scanned:
            raw = dict(profile.raw_info or {})
            raw["联系方式"] = merge_contacts(scanned, raw.get("联系方式"))
            profile.raw_info = raw
            self.db.add(profile)

        # Task 12: 最新显式自述优先 —— 唯一允许文本压过 extractor 回传的位置
        # (extractor 受"保留已有信息"倾向会滞留旧值,实测 wrong number 场景命中)
        latest_stated = latest_stated_contacts(full_history)
        if latest_stated:
            current = normalize_contacts((profile.raw_info or {}).get("联系方式"))
            changed = False
            for k, v in latest_stated.items():
                if v and canon(str(v)) != canon(str(current.get(k, ""))):
                    current[k] = v
                    changed = True
            if changed:
                raw = dict(profile.raw_info or {})
                raw["联系方式"] = current
                profile.raw_info = raw
                self.db.add(profile)

        # Task 12c 最近提及裁决: 镜像组分歧时以客户最晚提及的号码为准
        # (杀 extractor 历史振荡 + 压制幻觉;Task 12b 同号同步场景被涵盖)
        after_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式"))
        resolved = resolve_by_recency(before_contacts, after_contacts, full_history)
        if resolved != after_contacts:
            raw = dict(profile.raw_info or {})
            raw["联系方式"] = resolved
            profile.raw_info = raw
            self.db.add(profile)

        # Task 3 轮末 cc 钩子(spec §6-2): 联系方式四层合并完成后、D1 邮件
        # 触发前——消费 cc_reply 推进状态机,并为本轮才入管道的号码补建 _cc
        # (spec §5: 首封补全号/推送派生依赖已建好的 _cc)。cc_wall_fired =
        # 本轮墙确有触发(P1+P2 首问轮判定输入,见 _cc_round_end)。
        # P4': pre_turn_country = 轮前画像快照(本轮 extractor 合并前的
        # "目标国家"值),画像报国信号的比对基线。
        await self._cc_round_end(conv, profile, cc_reply, cc_wall_fired=cc_wall,
                                 pre_turn_country=pre_turn_country)

        # Round 3: Evaluate (仅 ≥3 条客户消息才有意义, 前两轮必然 pending —— D7)
        user_msg_count = sum(1 for m in full_history if m["role"] == "user")
        evaluation_ran = user_msg_count >= 3
        if evaluation_ran:
            evaluation = await self.ai.evaluate(
                profile={k: v for k, v in (profile.raw_info or {}).items()
                         if not str(k).startswith("_")},
                recent_messages=full_history,
            )
        else:
            # 客户端 evaluated 事件语义保持 pending;邮件标签由 evaluation_ran 决定(D2)
            evaluation = {"intent_level": "pending", "reasoning": "",
                          "contact_info": {}, "info_sufficient": False}
        logger.info("Evaluation (stream): profile=%s | eval=%s",
                    profile.raw_info, evaluation)

        # 邮件触发(D1): 与会话结束彻底解耦,联系方式到手即发
        pending_pushes: list[LeadPushLog] = []
        await self._maybe_send_lead_email(
            conv, profile, evaluation, evaluated=evaluation_ran,
            pending_pushes=pending_pushes,
        )

        # 联系方式判定(供 D8 评估落库): 有意向但没联系方式 → 卡单供 admin 人工兜底
        confirmed_now, _ = split_validate(
            normalize_contacts((profile.raw_info or {}).get("联系方式")))

        # 评估落库(D8 泛化): 每轮评估后刷新意向;email_sent/correction_count
        # 永不触碰。有意向无联系方式 = 卡单(admin 可筛);已发过邮件的
        # contact_info 由触发器更正分支管理,这里不覆盖。
        # Task 6: 存储视图走面板尾注(等待态标注,e164 派生不落本体)
        if user_msg_count >= 3:
            ev = await self._get_or_init_evaluation(conv.id)
            ev.intent_level = IntentLevel(evaluation.get("intent_level", "pending"))
            ev.reasoning = evaluation.get("reasoning", "")
            if not confirmed_now:
                ev.contact_info = {}
            elif not ev.email_sent:
                ev.contact_info = _panel_contact_info(
                    confirmed_now, (profile.raw_info or {}).get("_cc"))
            self.db.add(ev)

        # 决策A(2026-08-18): 自动结束移除——会话永不因评估结果结束;
        # evaluated 事件保留但 ended 恒为 False(客户端协议不变)。
        await self.db.commit()

        # 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存
        # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞。
        # session_factory 随任务传递: 写回本租户的库(Task 5b);config_data
        # 同随任务传递(Task 9: 重试矩阵参数走租户配置)
        lead_push_service.fire_pending_pushes(
            pending_pushes, self.session_factory,
            config_data=self.runtime.config.data)

        yield {
            "type": "evaluated",
            "ended": False,
            "intent_level": evaluation.get("intent_level", "pending"),
        }

    async def _get_or_init_evaluation(self, conv_id: str) -> Evaluation:
        result = await self.db.execute(
            select(Evaluation).where(Evaluation.conversation_id == conv_id)
        )
        ev = result.scalar_one_or_none()
        if not ev:
            ev = Evaluation(
                conversation_id=conv_id,
                intent_level=IntentLevel.pending,
                contact_info={},
            )
            self.db.add(ev)
            await self.db.flush()
        return ev

    async def _recently_emailed_canons(self) -> set[str]:
        cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
        # 只取需要的列;🔴 禁止加 LIMIT——完整 24h 窗口才正确。
        # Task 6 双保险: join 画像取历史会话的 _cc——各渠道 e164 的 canon
        # 一并入集(快照存裸号原文+尾注,客户以 + 前缀完整形态回访同号时
        # 仍要命中,否则回访退化成全量询盘)。
        result = await self.db.execute(
            select(Evaluation.contact_info, CustomerProfile.raw_info)
            .outerjoin(
                CustomerProfile,
                CustomerProfile.conversation_id == Evaluation.conversation_id)
            .where(
                Evaluation.email_sent.is_(True),
                Evaluation.evaluated_at >= cutoff,
            )
        )
        seen: set[str] = set()
        for contact_info, raw_info in result:
            for v in (contact_info or {}).values():
                if v:
                    seen.add(canon(str(v)))
            for state in ((raw_info or {}).get("_cc") or {}).values():
                e164 = str((state or {}).get("e164") or "")
                if e164:
                    seen.add(canon(e164))
        return seen

    async def _cc_round_end(
        self, conv: Conversation, profile: CustomerProfile,
        cc_reply: dict | None, cc_wall_fired: bool = False,
        pre_turn_country: str | None = None,
    ) -> None:
        """轮末 cc 钩子(spec §6-2): 联系方式四层合并后、D1 邮件前调用。

        - 首问轮免计免消费(P1+P2 一体修复,2026-10-08 真实验收): 轮前
          新建的等待态 = 回显/追问**首次发出轮**——客户本轮还没机会回应
          确认句,轮末既不消耗 attempts 也不消费 cc_reply(extractor 对
          "确认句排给号消息之后"历史的任何回应分类必是误判,P2);attempts
          从等待回应轮起计(报国/不理睬轮才递增)——修复前裸号轮轮末即耗
          attempts=1,报国轮 country 分支消费到 2 直落 fallback,重算的
          e164 永无回显确认机会(P1)。判据 = 本轮墙确有触发(cc_wall_fired)
          且状态尚未标记 asked——统一覆盖轮前新建与轮末补建两条建态路径
          的首个墙轮(补建态当轮无墙,首问轮顺延到下轮)。
        - cc_reply 消费(双门控 Python 侧): 仅 whatsapp 组等待态(cc_due)消费
          ——无回应/漏填也推进(attempts 随轮递增,达上限落定 fallback);
          落定态仅开迟到 country 升级口: 客户主动报国(非空 country)且
          source 非 no_ip 耗尽 → transit 重算 e164(客户主动给的信息不
          丢弃);迟到 affirm 不升级(对已标"未经确认"旧号的确认语义模糊,
          confirmed 态更不重算——客户改主意请重发完整号,走 canon 重置)。
        - P4'(2026-10-08 复验)画像报国信号兜底: extractor 对"画像已记、
          客户重复陈述"倾向漏填 cc_reply(复验 2/2 稳定复现),报国不能赌
          分类——等待态(cc_due)且非首问轮,本轮画像"目标国家"**新现或值
          变化**(与轮前快照 pre_turn_country 比对,占位值不算)→ 视为
          country 信号,name_to_region 转 ISO 后走 transit country 分支
          (与 cc_reply.country 同路径)。cc_reply 显式分类优先(reply 已定
          不介入);国名不识别 → 忽略(不重算,该轮按无回应语义消耗)。
        - canon 变化(换号/号消失)→ RESET 删组状态。
        - 补建 pass: 对最终联系方式重跑判定树(_cc_pre_turn)——本轮才经
          extractor 入管道的裸号/换号后的新号当场建状态,不下轮补课。
        - Task 9: contact.cc_completion=False → 整钩子跳过(轮前判定树
          已不建态,此处不消费不推进;国名映射查表带租户 extras)。
        """
        if not self._contact_flags["cc_completion"]:
            return
        original_cc = (profile.raw_info or {}).get("_cc")
        ip_iso = ip_service.resolve_region_code(
            (conv.client_info or {}).get("ip"))
        raw = dict(profile.raw_info or {})
        confirmed_end, _ = split_validate(normalize_contacts(raw.get("联系方式")))
        wa_canon_end = digits_only(str(confirmed_end.get("whatsapp") or ""))

        cc_map = dict(raw.get("_cc") or {})
        state = cc_map.get("whatsapp")

        # 首问轮: cc_wall_fired ⇔ 轮前等待态(墙必触发);asked 标记由本钩子
        # 独占管理(cc_completion 不感知),transit 的 dict 拷贝天然携带。
        # canon 分支在其后——首问轮遇换号仍以 RESET 优先,标记随旧态作废。
        first_ask = (state is not None and cc_wall_fired
                     and not state.get("asked")
                     and cc_completion.cc_due(state))

        reply = None
        if isinstance(cc_reply, dict) and not first_ask:
            country = str(cc_reply.get("country") or "").strip()
            if cc_reply.get("affirm") is not None or country:
                if cc_completion.cc_due(state):
                    reply = cc_reply
                elif (state is not None and not state.get("confirmed")
                      and country and state.get("source") != "no_ip"):
                    reply = cc_reply   # 迟到 country 升级口

        # P4' 画像报国信号兜底(仅 cc_reply 未消费时介入): raw 此时已是
        # extractor 合并后的本轮终态,"目标国家"新现/值变化 = 客户本轮
        # 陈述了新国家——确定性信号,不赌 extractor 是否把它分类进 cc_reply
        if (reply is None and not first_ask
                and cc_completion.cc_due(state)):
            signal = _country_signal(raw.get("目标国家"), pre_turn_country)
            if signal and cc_completion.name_to_region(
                    signal, extras=self._region_extras):
                reply = {"affirm": False, "country": signal}

        if state is not None:
            if not wa_canon_end or wa_canon_end != str(state.get("canon") or ""):
                cc_map.pop("whatsapp", None)   # 换号/号消失 → RESET
            elif first_ask:
                cc_map["whatsapp"] = {**state, "asked": True}   # 首问轮免计
            elif reply is not None or cc_completion.cc_due(state):
                cc_map["whatsapp"] = cc_completion.transit(
                    state, reply, extras=self._region_extras)
        if cc_map:
            raw["_cc"] = cc_map
        else:
            raw.pop("_cc", None)
        _cc_pre_turn(raw, confirmed_end, ip_iso)
        if raw.get("_cc") != original_cc:
            profile.raw_info = raw
            self.db.add(profile)

    async def _maybe_send_lead_email(
        self, conv: Conversation, profile: CustomerProfile, evaluation: dict,
        evaluated: bool = True,
        pending_pushes: list[LeadPushLog] | None = None,
    ) -> None:
        """D1: 联系方式是发邮件的唯一条件。confirmed 才触发;suspect 只进提醒。

        evaluated=False 表示本轮跳过了 GLM 评估(<3条客户消息)——此时邮件
        intent_level 必须传空串(→[未评估]),不得把客户端事件用的 "pending"
        透传成 [待确认](D2 两者语义不同)。
        """
        contacts_all = normalize_contacts((profile.raw_info or {}).get("联系方式"))
        confirmed, _suspect = split_validate(contacts_all)
        if not confirmed:
            return
        # Task 6(spec §5): 出站渲染取 _cc(首封补全号/标注/主题行/推送派生);
        # 轮末钩子已保证 _cc 与最终联系方式同 canon 同步
        cc_map = (profile.raw_info or {}).get("_cc") or {}
        ev = await self._get_or_init_evaluation(conv.id)

        if not ev.email_sent:
            # S3: 任一值已见于24h内其他会话 → 轻提醒;否则全量。
            # Task 6 双保险: 比对候选集含 _cc.e164 的 canon(见 _channel_canons)
            # Task 9: 邮件/推送参数走租户配置(config_data 透传给 resolve_*)
            seen = await self._recently_emailed_canons()
            if any(cands & seen
                   for cands in _channel_canons(confirmed, cc_map).values()):
                await self._collect_push(
                    pending_pushes, ev, conv, profile, confirmed,
                    "returning", evaluation, evaluated)
                sent = await email_service.send_returning_lead_email(
                    confirmed, cc_info=cc_map,
                    config_data=self.runtime.config.data)
            else:
                await self._collect_push(
                    pending_pushes, ev, conv, profile, confirmed,
                    "lead", evaluation, evaluated)
                messages = await self._load_history(conv.id)
                sent = await email_service.send_lead_email(
                    customer_profile=profile.raw_info or {},
                    contact_info=confirmed,
                    reasoning=evaluation.get("reasoning", ""),
                    conversation_messages=messages,
                    intent_level=evaluation.get("intent_level", "") if evaluated else "",
                    cc_info=cc_map,
                    config_data=self.runtime.config.data,
                )
            if sent:
                ev.email_sent = True
                ev.contact_info = _panel_contact_info(confirmed, cc_map)
                self.db.add(ev)
        else:
            old_canons = {canon(str(v)) for v in (ev.contact_info or {}).values()}
            # Task 6 双保险: 新值候选集含 _cc.e164 的 canon——确认落定
            # (号码未变)不补发更正;真换号(两组候选都不在旧集)才走更正
            if any(not (cands & old_canons)
                   for cands in _channel_canons(confirmed, cc_map).values()):
                if ev.correction_count < 3:
                    count = ev.correction_count + 1
                    await self._collect_push(
                        pending_pushes, ev, conv, profile, confirmed,
                        "correction", evaluation, evaluated)
                    if await email_service.send_correction_email(
                            confirmed, count=count, cc_info=cc_map,
                            config_data=self.runtime.config.data):
                        ev.correction_count = count
                        ev.contact_info = _panel_contact_info(confirmed, cc_map)
                        self.db.add(ev)
                else:
                    # D4'(2026-08-18): 锁邮件不锁快照 —— 上限已到,
                    # admin 仍要看到客户最新号码
                    ev.contact_info = _panel_contact_info(confirmed, cc_map)
                    self.db.add(ev)

    async def _collect_push(
        self, pending, ev, conv, profile, confirmed,
        push_type, evaluation, evaluated,
    ) -> None:
        """询盘推送派发(设计 §8): 触发条件=进入发信分支,与 SMTP 成败解耦。

        dispatch 内部吞异常(失败只记日志),这里无需再兜。config_data =
        租户配置(Task 9: 开关/前缀走 config, 空槽回退 env)。
        """
        log = await lead_push_service.dispatch_lead_push(
            self.db, ev, conv, profile, confirmed, push_type,
            evaluation, evaluated,
            config_data=self.runtime.config.data,
        )
        if log is not None and pending is not None:
            pending.append(log)

    async def _load_history(self, conv_id: str) -> list[dict]:
        result = await self.db.execute(
            select(Message)
            .where(Message.conversation_id == conv_id)
            .order_by(Message.created_at)
        )
        messages = result.scalars().all()
        return [{"role": m.role.value, "content": m.content} for m in messages]

    async def _get_profile(self, conv_id: str) -> CustomerProfile:
        result = await self.db.execute(
            select(CustomerProfile).where(CustomerProfile.conversation_id == conv_id)
        )
        return result.scalar_one()

    @staticmethod
    def _enrich_short_message(content: str, raw_info: dict, history: list[dict] | None = None) -> str:
        """Enrich short user messages with conversation context.

        Two sources: extractor output (primary), last substantive user
        message (fallback when extractor hasn't run yet).
        """
        if len(content.strip()) >= 30:
            return content

        facts = []

        # Primary: extractor output
        if raw_info:
            facts = [f"{k}: {v}" for k, v in raw_info.items()
                     if v and not str(k).startswith("_") and k != "联系方式"]

        # Fallback: use last substantive user message as context
        if not facts and history:
            for msg in reversed(history[:-1]):
                if msg.get("role") == "user" and len(msg.get("content", "").strip()) >= 5:
                    facts.append("客户之前说: " + msg["content"][:80])
                    break

        if not facts:
            return content

        return (
            content
            + "\n\n【系统上下文 - 系统标注，不要向客户提及】已知: "
            + " | ".join(facts)
            + "。不要重复询问这些已确认的信息！"
        )

    @staticmethod
    def _build_context_reminder(raw_info: dict, latest_msg: str,
                                history: list[dict] | None = None,
                                source: str = "",
                                contact_flags: dict | None = None,
                                wall: CompiledWall | None = None) -> str:
        """Build a context reminder from extracted profile info.
        Task 4 词表数据化: 六类轮中提醒话术(boundary/contact_ask/
        language_follow/lead_given/cc_guide/email_light)的数据源 =
        墙 reminders(seed_profiles/<slug>/wall.json);
        {shown}/{no_contact_web}/{no_contact_already} 占位由本函数
        .format() 回填,值与数据化前逐字等价。
        contact_flags(Task 9): contact 行为开关——cc_completion=False 跳过
        cc 确认指引,email_light_ask=False 跳过邮箱轻补;None = 全开
        (直调本函数的旧测试语义)。
        wall(C-1 终审): 租户编译墙(触发词/提醒话术/no_contact 全走它);
        None = 模块级默认墙(直调/旧测试语义——BOUNDARY_TRIGGER_RE 等
        模块常量即该默认视图)。"""
        w = wall if wall is not None else DEFAULT_WALL
        # 职能边界提醒(评审#7: 前置到早退之前,首条消息画像为空也要生效)
        boundary_note = ""
        if w.triggers["boundary"].search(latest_msg or ""):
            if w.triggers["budget"].search(latest_msg or ""):
                # 评审#9: 预算求推荐 → 预算路径(推荐+反问+引导),不一刀切要号
                boundary_note = w.reminders["boundary"]["budget"]
            else:
                boundary_note = w.reminders["boundary"]["normal"]

        # 联系方式状态前移(供 boundary/contact 提醒共用判定)
        contacts_all = normalize_contacts(raw_info.get("联系方式"))
        confirmed, suspect = split_validate(contacts_all)

        # 索要我方联系方式提醒(15.2): 与 boundary_note 同样前移到早退之前,
        # 空画像首条消息也要生效;按 source 分支(widget/standalone 有按钮, app 无)。
        # 冒烟实证: 抽象指令("用App话术")GLM 会无视并背 prompt 里更显眼的 web 脚本——
        # 提醒必须携带该分支的逐字话术,不让模型自己挑。
        contact_note = ""
        if w.triggers["contact_ask"].search(latest_msg or ""):
            if source == "app":
                if confirmed or suspect:
                    # 已留资: 索要我方联系方式 → 逐字"经理主动联系"版,
                    # 与"禁止再索要"提醒不再打架
                    # Task 3 收尾: 第三处禁令同步改写(禁令集完整)——与前两处
                    # 同语义,只禁已提供过的渠道,缺失渠道放行轻补一次
                    contact_note = w.reminders["contact_ask"]["app_already"]
                else:
                    contact_note = w.reminders["contact_ask"]["app"]
            else:
                contact_note = w.reminders["contact_ask"]["web"].format(
                    no_contact_web=w.no_contact["web"])

        # 语言跟随提醒(Task 11): 与 boundary/contact 提醒同挂载位——
        # 首条消息画像为空也要生效(中文客户开场即漂英文是实测重灾区)
        lang_note = _lang_note_for(latest_msg, wall=wall)

        if not raw_info:
            # 空画像:仅返回边界/联系/语言提醒(可能为空串)
            return "\n".join(x for x in (boundary_note, contact_note, lang_note) if x)

        parts = ["[上下文提醒 - 这是一条系统指令，绝对不要向客户提及这段内容]"]
        parts.append("客户已明确告知以下信息，你绝对不能再问这些：")

        for key, val in raw_info.items():
            if str(key).startswith("_") or key == "联系方式":
                continue  # 内部状态与联系方式单独渲染
            if val:
                parts.append(f"{key}: {val}")

        # Task 9: contact 行为开关注入本函数(staticmethod 保持旧直调签名,
        # None = 全开);cc 确认指引与邮箱轻补按开关短路
        flags = (contact_flags if contact_flags is not None
                 else resolve_contact_flags(None))

        # Task 3(spec §4.2/§4.3b): 邮箱轻补注入(cc 分支已删,判定树接管号码
        # 区号补全——确认提醒由 cc_due 驱动,Task 4 接)。计数键(_marks)此处
        # 弃用: 本函数是 staticmethod 收内存副本,写回由 stream 调用方以同参
        # 调用纯函数取回并落库(纯函数保证两处判定不漂移)。计算前移到
        # confirmed 块之前: 未落定轮邮箱轻补不注入(cc_settled 串行)。
        completion_note, _marks = _contact_completion_note(
            raw_info, confirmed, suspect,
            lang="zh" if _CJK_RE.search(latest_msg or "") else "en",
            email_ask=flags["email_light_ask"], wall=wall)

        # Task 4(spec §4 回显确认轮): cc 等待态注入确认句指引——有 e164 用
        # 回显模板(逐字 intl),no_ip 用 ASK 追问句;与邮箱轻补 cc_settled
        # 串行互斥,两个"逐字使用这句"永不同轮。无计数键: 轮次上限由状态机
        # attempts 管理(轮末 transit 递增),提醒每轮等待态都在场。
        # P3: 语言路由扩 ES(_cc_lang_of),西语轮注入西语模板。
        cc_note = (_cc_confirm_note(
            (raw_info.get("_cc") or {}).get("whatsapp"), _cc_lang_of(latest_msg),
            wall=wall)
            if flags["cc_completion"] else "")

        # 联系方式专项(D6) 与 职能边界(D6联动)
        if confirmed:
            shown = ", ".join(
                f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items())
            # 逐字话术直接引用墙 no_contact["already"](Task 9 收口:
            # 视图直取单一事实源)——与零号码墙替换句同源,三处"经理主动联系"
            # 话术不漂移
            # Task 3 防自锁改写(spec 4.1-6): 原"绝对禁止再索要任何联系方式！"
            # 会抑制缺失渠道轻补——只禁已提供过的渠道,缺失渠道放行一次
            parts.append(w.reminders["lead_given"]["forbidden"].format(
                shown=shown))
            parts.append(w.reminders["lead_given"]["respond_script"].format(
                no_contact_already=w.no_contact["already"]))
            if boundary_note:
                parts.append(boundary_note)
                # Task 3 防自锁改写: 尾部补缺失渠道豁免(与上方改写同因)
                parts.append(w.reminders["lead_given"]["boundary_tail"])
        elif suspect:
            shown = ", ".join(
                f"{k}: {' '.join(str(v).split())[:60]}" for k, v in suspect.items())
            parts.append(w.reminders["lead_given"]["suspect"].format(
                shown=shown))
            if boundary_note:
                parts.append(boundary_note)

        if boundary_note and not confirmed and not suspect:
            parts.append(boundary_note)  # 未留资:边界提醒引导留资(职能边界话术自带)

        # Task 4 cc 确认指引注入(与下方邮箱轻补互斥——cc_due vs cc_settled)
        if cc_note:
            parts.append(cc_note)

        # Task 3 轻补注入(计算已前移到 confirmed 块之前;cc_settled 未落定时
        # note 为空,天然不注入)
        if completion_note:
            parts.append(completion_note)

        # 15.2: 索要我方联系方式提醒注入(镜像 boundary_note 的挂载位)。
        # 互留提示仅限 web 来源且客户未留资(confirmed/suspect 都算"已留");
        # App 的反向留资话术本身就是主话术,不再叠加
        if contact_note:
            note = contact_note
            if source != "app" and not confirmed and not suspect:
                note += "客户尚未留过联系方式，可顺势轻问一句（每会话最多一次）。"
            parts.append(note)

        if len(latest_msg.strip()) < 25:
            parts.append("⚠️ 客户发的是短消息！必须结合上面已知信息来理解，绝对不能当作独立问题！")

        if lang_note:
            parts.append(lang_note)

        return "\n".join(parts)