group_fqcd_jr/tests/unit/service/test_governance_disclaimer.py

"""免责声明强制注入与 `applicable_agents` 门控的单元测试。

两条需求来源：
1. 上线门禁 F5「免责声明强制注入，面向客户输出 **100%** 附固定话术」。所以**正常输出**与
   **被负面词拦截后替换的输出**两个分支都必须带上免责声明——被拦截的回复同样是面向客户的输出。
2. Task 1 审查发现的 fail-open 缺陷：`applicable_agents` 为 `[]`/`NULL` 时旧的
   `if agents and ...` 判据不执行过滤，规则**溢出到所有 Agent**，与项目"失败关闭"原则相反。

本文件不连数据库：`PlatformGovernance` 的取数会话用替身替换（与
`tests/unit/service/test_memory_extraction_service.py` 同一手法），
真实 MySQL 上的门控行为由 `tests/integration/test_compliance_seed_mysql.py` 固定。
"""

from typing import Any

import pytest

from app.core.contracts import (
    AgentDefinition,
    AgentResult,
    CoreResult,
    RecalledMemory,
    RequestContext,
    ResolvedAgentConfig,
)
from app.service.agent import governance
from app.service.agent.governance import FALLBACK_DISCLAIMER, review_output

CONTEXT = RequestContext(user_id="9001", trace_id="t", roles=("customer",))
CONFIG = ResolvedAgentConfig(config_version="v", prompt_version="p", model_endpoint="m")

#: 固定话术的插入位置：`\n\n` 分隔，附在正文末尾。
DISCLAIMER = "\n\n" + FALLBACK_DISCLAIMER


def _result(text: str) -> AgentResult:
    return AgentResult(run_id="r", result=CoreResult(text=text))


def _review(
    text_or_result: Any, config: ResolvedAgentConfig | None = None,
    memories: tuple[RecalledMemory, ...] = (), **kwargs: Any,
) -> AgentResult:
    """面向客户的治理调用（带 `agent_type`）。

    门禁 F5 的适用范围是**已确认面向客户**的 Agent（`agent_type` 由 `PlatformGovernance`
    从发布版本反查）。本文件的断言全部针对该场景，所以统一在这里显式声明 agent 类型，
    避免每处都写一遍；`test_internal_agent_gets_no_disclaimer` 是对照组。
    """
    result = text_or_result if isinstance(text_or_result, AgentResult) \
        else _result(text_or_result)
    return review_output(result, CONTEXT, config or CONFIG, memories,
                         agent_type="customer_service", **kwargs)


def test_disclaimer_is_appended_to_normal_output() -> None:
    result = _review("基金申购后 T+1 确认份额。")
    assert FALLBACK_DISCLAIMER in result.result.text
    assert result.result.text.startswith("基金申购后 T+1 确认份额。")
    assert result.result.text.endswith(DISCLAIMER)


def test_internal_agent_gets_no_disclaimer() -> None:
    """对照组：内部 Agent（风控分析）的结构化输出**不得**被追加面向客户的免责声明。

    判据是"输出形态"：风控输出是字段化摘要（预警编号/级别/建议动作），追加一句面向
    投资者的免责声明会破坏其字段契约，下游解析与 `tests/contract/test_risk_agent_contract.py`
    都会因此失败（合并时实测复现过）。`agent_type` 为空（未声明）时同样不注入。
    """
    for agent_type in ("risk", ""):
        result = review_output(_result("工单标题：测试工单"), CONTEXT, CONFIG, (),
                               agent_type=agent_type)
        assert result.result.text == "工单标题：测试工单", agent_type


def test_disclaimer_is_not_duplicated() -> None:
    """幂等：同一结果被治理两次（重放/重试）不得把免责声明叠成两条。"""
    once = _review("答案。")
    twice = _review(once)
    assert twice.result.text.count(FALLBACK_DISCLAIMER) == 1
    assert twice.result.text == once.result.text


def test_disclaimer_inside_the_answer_is_still_appended_at_the_end() -> None:
    """幂等判断不认"文本里出现过话术"，只认"末尾是不是我们自己追加的那个形状"。

    子串包含判断下，正文中间出现同样的文字就不会追加（实测过），于是"客户拿到的话术"取决于
    正文里恰好有没有这段话——门禁 F5 要的是 100% 覆盖，正文里出现过不算"已附话术"，
    必须落在**末尾**。
    """
    middle = f"前半句。{FALLBACK_DISCLAIMER}后半句。"
    result = _review(middle)
    assert result.result.text == middle + DISCLAIMER
    assert result.result.text.endswith(DISCLAIMER)
    assert result.result.text.count(FALLBACK_DISCLAIMER) == 2


def test_answer_already_carrying_the_appended_shape_is_not_duplicated() -> None:
    """幂等的另一半语义：正文已经以"我们追加过的形状"结尾时，不重复追加（话术只出现一次）。"""
    result = _review("答案。" + DISCLAIMER)
    assert result.result.text.count(FALLBACK_DISCLAIMER) == 1
    assert result.result.text.endswith(DISCLAIMER)


def test_degenerate_published_wording_is_still_appended() -> None:
    """修复轮 2 的核心动机：**退化文案下也必须追加**。

    管理员若把免责声明发布成"。"（退化数据，`agent_reply_template.content_text` 没有任何校验），
    答案恰好是"您好。"时，`endswith(disclaimer_text)` 会判定"已经以话术结尾" → **不追加**，
    客户一条话术都拿不到。只认我们自己追加过的形状（"\\n\\n" + 文案）后，这种文本必然被追加。
    代价是极端情况下可能多追加一次——方向正确：门禁 F5 要的是 100% 附话术，宁可多一次不可漏。
    """
    result = _review("您好。", disclaimer="。")
    assert result.result.text == "您好。\n\n。"
    assert result.result.text.endswith("\n\n。")


def test_degenerate_published_wording_is_appended_only_once() -> None:
    """退化文案下幂等仍成立：第二次治理不再追加（认的正是第一次追加出来的那个形状）。"""
    once = _review("您好。", disclaimer="。")
    twice = _review(once, disclaimer="。")
    assert twice.result.text == once.result.text
    assert twice.result.text == "您好。\n\n。"


def test_fallback_disclaimer_survives_the_hardcoded_hard_patterns() -> None:
    """代码常量 `FALLBACK_DISCLAIMER` 的行为守卫：库侧 6 条话术有守卫，它此前没有。

    若该常量命中硬编码 `hard_patterns`，治理层会先把回复判定为"需人工核实"并替换成安全话术，
    也就是**平台自己的免责声明被平台自己的规则拦下**（Task 1 修过一次的"自绊"）。这条断言是哨兵。

    ⚠️ 为什么**不能**按"输入正文 = 兜底文案 + `assert FALLBACK_DISCLAIMER in 输出`"来写：
    那是恒真的空转——输入本身就是该文案，且命中硬编码词时文案会在替换**之后**被追加回末尾，
    两种情况下 `in` 都成立（已实测：把常量改成含「保证收益」后那条断言仍为真）。
    真正会失败的是下面两条：①没被判定为拦截；②正文原样保留（未被安全话术替换）。
    入参刻意用"正文 + 兜底文案"而不是只有文案，避免"输入即断言对象"造成的自证。
    """
    answer = "基金申购后 T+1 确认份额。" + FALLBACK_DISCLAIMER
    result = _review(answer)
    assert result.result.transfer_required is False, (
        "兜底免责声明被平台自己的硬编码 hard_patterns 拦下了（自绊）")
    # 正文原样保留（未被替换成安全话术），并按规范形状补一次"\n\n"分隔的追加：
    # 裸文案结尾不算"已追加"，这正是修复轮 2 收紧后的语义。
    assert result.result.text == answer + DISCLAIMER, "正文被替换成了安全话术，说明话术命中了规则"
    assert result.result.text.endswith(DISCLAIMER)


def test_disclaimer_is_appended_after_negative_word_replacement() -> None:
    """裁定 2：命中负面词被替换成安全话术的分支同样要追加免责声明（F5 要求 100%）。

    顺序固定为「校验 → 负面词替换 → 脱敏 → 追加免责声明」：免责声明必须落在**替换后**的
    文本之后，否则拦截分支就没有话术。
    """
    config = CONFIG.model_copy(update={"negative_rules": (("contains", "测试禁止词"),)})
    result = _review("这只基金测试禁止词", config=config)
    assert result.result.transfer_required is True
    assert "测试禁止词" not in result.result.text
    assert result.result.text.startswith("该内容需要人工核实")
    assert result.result.text.endswith(DISCLAIMER)


def test_disclaimer_is_appended_after_hard_pattern_replacement() -> None:
    """硬编码 `hard_patterns`（不依赖库配置）走的是同一分支，同样必须带话术。"""
    result = _review("这只产品保证收益")
    assert "保证收益" not in result.result.text
    assert result.result.text.endswith(DISCLAIMER)


def test_explicit_disclaimer_argument_wins_over_code_fallback() -> None:
    """异步层读到的库内话术优先于代码兜底。"""
    result = _review("答案。", disclaimer="库内免责声明。")
    assert result.result.text == "答案。\n\n库内免责声明。"
    assert FALLBACK_DISCLAIMER not in result.result.text


def test_blank_disclaimer_argument_still_gets_the_fixed_wording() -> None:
    """显式传入空串不得成为"绕过强制注入"的后门：空串一律退回代码兜底。

    门禁 F5 要求 100% 覆盖，"传了但传空"不算注入话术。
    """
    result = _review("答案。", disclaimer="  ")
    assert FALLBACK_DISCLAIMER in result.result.text


def test_existing_callers_without_disclaimer_still_get_wording() -> None:
    """既有调用点按旧签名调用（不传 disclaimer）必须照旧拿到固定话术。"""
    result = _review("答案。")
    assert result.result.text == "答案。" + DISCLAIMER


# ---------------------------------------------------------------------------
# `applicable_agents` 门控：fail-closed
# ---------------------------------------------------------------------------


class _FakeRows:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self._rows = rows

    def mappings(self) -> "_FakeRows":
        return self

    def all(self) -> list[dict[str, Any]]:
        return self._rows


class _FakeSession:
    """`resolve()` 用到的替身：`scalar` 取发布版本/话术，`execute` 取规则。

    这里**不**按 SQL 文本分派：`resolve()` 自己那条 `scalar` 就是取 release，改动它会
    让"是否命中发布配置"的语义漂移，把门控测试变成假绿。`review()` 需要区分两次查询，
    因此另有 `_FakeReviewSession`。
    """

    def __init__(self, *, rows: list[dict[str, Any]] = (), template: str | None = None) -> None:
        self._rows = rows
        self._template = template

    async def scalar(self, _statement: Any, _params: Any = None) -> Any:
        # 返回话术文本（话术查询）或 None（没有 active release，工具分支整段跳过）。
        return self._template

    async def execute(self, _statement: Any) -> _FakeRows:
        return _FakeRows(self._rows)

    async def __aenter__(self) -> "_FakeSession":
        return self

    async def __aexit__(self, *args: object) -> None:
        return None


class _FakeReviewSession(_FakeSession):
    """`PlatformGovernance.review()` 用到的替身：`scalar` 只回话术文本。

    `agent_type` **不再**从库里反查——它由 `BaseAgent` 从 Agent 定义直接传给 `review()`
    （`definition.agent_type` 是定义的一部分，没有理由为它多付一次查询）。
    因此替身只需回答"话术是什么"。
    """

    def __init__(self, *, template: str | None = None) -> None:
        super().__init__(template=template)


def _rows(applicable_agents: Any) -> list[dict[str, Any]]:
    """一行规则：MySQL JSON 列经驱动回来是字符串，`NULL` 回来是 None（实测口径）。"""
    return [{"match_type": "contains", "word_pattern": "测试禁止词",
             "applicable_agents": applicable_agents}]


@pytest.mark.parametrize(
    ("applicable_agents", "customer_service_rules", "advisor_rules"),
    [
        # (客服加载数, 投顾加载数)：与派发说明里的真值表逐格对应，第 2、3 行是本次修复的缺陷。
        ('["customer_service"]', 1, 0),  # 正常门控：只对声明的 Agent 生效
        ([], 0, 0),                      # 空数组：不适用任何 Agent（fail-closed；原来溢出成 1/1）
        ('[]', 0, 0),                    # 同上，但按驱动真实口径写成 JSON 文本（实测回 str）
        (None, 0, 0),                    # NULL：同上
        ('["advisor"]', 0, 1),           # 指向别的 Agent：客服一条都拿不到（原来的静默失效）
    ],
)
async def test_applicable_agents_gating_fails_closed(
    monkeypatch: pytest.MonkeyPatch,
    applicable_agents: Any,
    customer_service_rules: int,
    advisor_rules: int,
) -> None:
    """三种坏取值都不得让规则溢出：空数组/NULL 是"不适用任何 Agent"，不是"适用全部"。"""
    monkeypatch.setattr(governance, "SessionFactory",
                        lambda: _FakeSession(rows=_rows(applicable_agents)))
    service = governance.PlatformGovernance()
    customer = await service.resolve(
        AgentDefinition(agent_type="customer_service", version="1"), CONTEXT)
    advisor = await service.resolve(AgentDefinition(agent_type="advisor", version="1"), CONTEXT)
    assert len(customer.negative_rules) == customer_service_rules
    assert len(advisor.negative_rules) == advisor_rules


async def test_empty_applicable_agents_rules_are_not_loaded_at_all(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """空取值的反证断言：不是"条数变少"，而是该规则对**任何** Agent 都不存在。"""
    monkeypatch.setattr(governance, "SessionFactory",
                        lambda: _FakeSession(rows=_rows([])))
    service = governance.PlatformGovernance()
    for agent_type in ("customer_service", "advisor", "fund_query_demo"):
        resolved = await service.resolve(
            AgentDefinition(agent_type=agent_type, version="1"), CONTEXT)
        assert resolved.negative_rules == ()


# ---------------------------------------------------------------------------
# 同步/异步分离（裁定 1）：读库在异步层，追加在同步层
# ---------------------------------------------------------------------------


CONFIG_WITH_RELEASE = CONFIG.model_copy(update={"release_id": 186})


async def test_review_reads_template_from_database(monkeypatch: pytest.MonkeyPatch) -> None:
    """走 `PlatformGovernance.review()`：话术取库，`agent_type` 由调用方透传。"""
    monkeypatch.setattr(governance, "SessionFactory",
                        lambda: _FakeReviewSession(template="库内固定免责声明。"))
    reviewed = await governance.PlatformGovernance().review(
        _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
    assert reviewed.result.text == "答案。\n\n库内固定免责声明。"


async def test_review_skips_disclaimer_for_internal_agent(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """内部 Agent（风控）不追加话术：它的输出是字段化摘要，加话术会破坏字段契约。

    `agent_type` 由 `BaseAgent` 从定义传入，与发布配置无关（没有发布版本也要能判定）。
    """
    monkeypatch.setattr(governance, "SessionFactory",
                        lambda: _FakeReviewSession(template="库内固定免责声明。"))
    reviewed = await governance.PlatformGovernance().review(
        _result("工单标题：测试工单"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="risk")
    assert reviewed.result.text == "工单标题：测试工单"


async def test_review_without_agent_type_declaration_does_not_inject(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """治理替身按旧签名调用（未传 `agent_type`）时不注入，而不是猜成面向客户。

    这条是给**测试替身**定的口径：它们刻意不连库、也不知道 agent 类型，若默认注入，
    每个替身测试都会被塞进一句话术，等于用测试噪声换一个假的安全感。生产路径恒有类型
    （`BaseAgent` 从定义传），所以 F5 的覆盖不受影响。
    """
    monkeypatch.setattr(governance, "SessionFactory",
                        lambda: _FakeReviewSession(template="库内固定免责声明。"))
    reviewed = await governance.PlatformGovernance().review(
        _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, ())
    assert reviewed.result.text == "答案。"


@pytest.mark.parametrize("template", [None, "", "   "])
async def test_review_falls_back_to_code_constant_when_template_missing(
    monkeypatch: pytest.MonkeyPatch, template: str | None,
) -> None:
    """查不到/查到空白话术时退回代码常量：门禁 F5 不依赖数据库可用性。"""
    monkeypatch.setattr(governance, "SessionFactory", lambda: _FakeReviewSession(template=template))
    reviewed = await governance.PlatformGovernance().review(
        _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
    assert reviewed.result.text == "答案。" + DISCLAIMER


async def test_review_survives_database_failure(monkeypatch: pytest.MonkeyPatch) -> None:
    """取数抛异常不得让回复失败，也不得让回复**没有**免责声明。

    场景取"会话可建、查询必炸"（比"建会话就炸"更贴近真实的连接池故障）。
    """

    class _ExplodingTemplateSession(_FakeSession):
        async def scalar(self, _statement: Any, _params: Any = None) -> Any:
            raise RuntimeError("database unavailable")

    monkeypatch.setattr(governance, "SessionFactory", lambda: _ExplodingTemplateSession())
    reviewed = await governance.PlatformGovernance().review(
        _result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
    assert reviewed.result.text == "答案。" + DISCLAIMER


def test_recalled_memories_do_not_affect_disclaimer() -> None:
    """带记忆的正常路径同样注入（确认追加在引用校验之后、不受 memories 影响）。"""
    memories = (RecalledMemory(memory_uuid="m1", customer_id="9001", content="偏好低风险"),)
    reviewed = _review("答案。", memories=memories)
    assert reviewed.result.text == "答案。" + DISCLAIMER