ai-kefu-admin/tests/test_prompt_provider.py

"""PromptProvider(Task 6): 租户提示词四件套 + $DEFLECTION_POOL 单一事实源注入。

- 提示词从租户目录(seed_profiles/<slug>/prompts)读,不再有代码级
  app/engine/prompts 副本;
- system.md 的迂回话术枚举段替换为 $DEFLECTION_POOL 占位符,组装时注入
  wall.json deflections 池(话术只存 wall.json 一份,改一处两边同步变);
- extractor/evaluator 的 $existing_info/$messages/$profile/$recent_messages
  替换逻辑从 AIService 原样搬入 Provider(行为等价)。

路径纪律: 一律 Path(__file__) 锚定仓库根,禁 cwd 相对。
"""
import json
import pathlib

import pytest

from app.engine.services.ai_service import DEFAULT_PROVIDER, PromptProvider

REPO = pathlib.Path(__file__).resolve().parents[1]
JC_PROMPTS = REPO / "seed_profiles" / "jiancai" / "prompts"
AUTO_PROMPTS = REPO / "seed_profiles" / "auto" / "prompts"


def _wall(slug: str) -> dict:
    return json.loads(
        (REPO / "seed_profiles" / slug / "wall.json").read_text(encoding="utf-8"))


# ---- 加载 --------------------------------------------------------------

def test_provider_loads_from_tenant_dir():
    p = PromptProvider.from_dir(JC_PROMPTS)
    assert "HESUPIPE" in p.company_facts or "管道" in p.system


def test_provider_reads_all_four_prompts():
    p = PromptProvider.from_dir(JC_PROMPTS)
    assert "$existing_info" in p.extractor and "$messages" in p.extractor
    assert "$profile" in p.evaluator and "$recent_messages" in p.evaluator


def test_default_provider_is_jiancai_seed():
    # 旧测试/默认路径的代码级默认 = jiancai 出厂(Path(__file__) 锚定,不随 cwd 漂移)
    assert DEFAULT_PROVIDER.system == PromptProvider.from_dir(JC_PROMPTS).system


# ---- $DEFLECTION_POOL 注入(单一事实源) ----------------------------------

def test_system_prompt_injects_deflection_pool():
    p = PromptProvider.from_dir(JC_PROMPTS)
    wall_json = _wall("jiancai")
    sysprompt = p.build_system(wall_deflections=wall_json["deflections"])
    for lang, lines in wall_json["deflections"].items():
        assert lines[0] in sysprompt        # 话术池注入 = 单一事实源
    assert "$DEFLECTION_POOL" not in sysprompt
    assert "HESUPIPE" in sysprompt          # 事实卡拼接保留


def test_system_md_carries_placeholder():
    # 防漂移锚: 两租户 system.md 必须保留占位符,丢了它话术池就静默失联
    for prompts_dir in (JC_PROMPTS, AUTO_PROMPTS):
        assert "$DEFLECTION_POOL" in PromptProvider.from_dir(prompts_dir).system


def test_build_system_appends_facts_after_system():
    p = PromptProvider.from_dir(JC_PROMPTS)
    out = p.build_system(wall_deflections=_wall("jiancai")["deflections"])
    # 顺序: 行为 prompt(含注入话术池) → 事实卡(与旧 _BASE_CHAT_PROMPT 拼接同构)
    assert out.index("## 语言规则") < out.index("# HESUPIPE 公司事实卡")


def test_build_system_auto_tenant_pool():
    # auto 墙只有 en/zh 池(无 es)——注入按实际池语言枚举,不猜不补
    p = PromptProvider.from_dir(AUTO_PROMPTS)
    wall = _wall("auto")
    sysprompt = p.build_system(wall_deflections=wall["deflections"])
    for lang, lines in wall["deflections"].items():
        assert lines[0] in sysprompt
    assert "$DEFLECTION_POOL" not in sysprompt
    assert "- ES:" not in sysprompt


def test_build_system_injection_is_language_grouped():
    # 注入格式: 按语言分组一行(- EN: … / - ZH: …),紧凑不膨胀
    p = PromptProvider.from_dir(JC_PROMPTS)
    out = p.build_system(wall_deflections=_wall("jiancai")["deflections"])
    assert "- EN:" in out and "- ZH:" in out and "- ES:" in out
    # 分组行序: EN 在 ZH 前(确定性输出,防未来语言序随机漂移断言抖动)
    assert out.index("- EN:") < out.index("- ZH:") < out.index("- ES:")


def test_injected_pool_lines_are_wall_clean():
    # 注入让话术池进入 system prompt(AI 会照抄)——池线必须过两墙;
    # 既有 test_spanish_pools_are_wall_clean 只盖 ES,这里补全两租户全语言
    from app.engine.services.boundary_filter import (
        no_contact_value, violates_boundary,
    )
    for slug in ("jiancai", "auto"):
        for lines in _wall(slug)["deflections"].values():
            for line in lines:
                assert not violates_boundary(line), line[:40]
                assert not no_contact_value(line), line[:40]


# ---- extractor/evaluator 组装(自 AIService 原样搬入) ----------------------

def test_build_extractor_replaces_placeholders():
    p = PromptProvider.from_dir(JC_PROMPTS)
    out = p.build_extractor(
        {"目标国家": "Egypt"},
        [{"role": "user", "content": "hi"},
         {"role": "assistant", "content": "hello there"}],
    )
    assert "$existing_info" not in out and "$messages" not in out
    assert '"目标国家": "Egypt"' in out        # json.dumps(ensure_ascii=False)
    assert "客户: hi" in out and "顾问: hello there" in out


def test_build_evaluator_replaces_placeholders_and_caps_messages():
    p = PromptProvider.from_dir(JC_PROMPTS)
    msgs = [{"role": "user", "content": f"m{i}"} for i in range(10)]
    out = p.build_evaluator({"intent_level": "pending"}, msgs)
    assert "$profile" not in out and "$recent_messages" not in out
    assert '"intent_level": "pending"' in out
    assert "客户: m9" in out                    # 取最近 6 条
    assert "客户: m0" not in out and "客户: m3" not in out


def test_build_extractor_handles_none_info():
    p = PromptProvider.from_dir(JC_PROMPTS)
    out = p.build_extractor(None, [])
    assert "{}" in out                           # existing_info or {} 语义保留


# ---- runtime 接线 --------------------------------------------------------

def test_registry_runtime_carries_prompts(tmp_path):
    from app.platform.registry import TenantRegistry
    reg = TenantRegistry.load_from_dir(REPO / "seed_profiles", data_dir=tmp_path)
    by_slug = {rt.config.slug: rt for rt in reg.runtimes()}
    assert "管道" in by_slug["jiancai"].prompts.system
    assert "汽车" in by_slug["auto"].prompts.system
    # runtime 的 prompts 即租户目录四件套(与手工 from_dir 等价)
    assert by_slug["jiancai"].prompts.system == PromptProvider.from_dir(JC_PROMPTS).system


def test_conversation_manager_runtime_is_required():
    # Task 5b 质量审查关门建议: runtime 从可选转必填——缺参必须显性炸
    from app.engine.services.conversation import ConversationManager
    with pytest.raises(TypeError):
        ConversationManager(db=None)