PMFlow/backend/tests/test_internal_query_context_packet.py

import pytest

from app.services.adaptive_rag_service.mixin_answer_mainline import AdaptiveRAGAnswerMainlineMixin
from app.services.adaptive_rag_service.mixin_answer_payload import AdaptiveRAGAnswerPayloadMixin
from app.services.adaptive_rag_service.mixin_deep_guard_answer import AdaptiveRAGDeepGuardAnswerMixin
from app.services.adaptive_rag_service.mixin_fast_answer_project_utils import AdaptiveRAGFastAnswerProjectUtilsMixin
from app.services.adaptive_rag_service.mixin_query_route_gateway import AdaptiveRAGQueryRouteGatewayMixin
from app.services.adaptive_rag_service.mixin_target_resolution import AdaptiveRAGTargetResolutionMixin
from app.services.adaptive_rag_service.internal_query_context import InternalContextPacket


class _PacketBuilderService(
    AdaptiveRAGAnswerMainlineMixin,
    AdaptiveRAGQueryRouteGatewayMixin,
):
    tenant_id = 9


class _VerifierService(
    AdaptiveRAGAnswerPayloadMixin,
    AdaptiveRAGQueryRouteGatewayMixin,
):
    pass


class _RuntimeSummaryService(AdaptiveRAGTargetResolutionMixin):
    pass


class _ProjectUtilsService(AdaptiveRAGFastAnswerProjectUtilsMixin):
    pass


def _build_internal_packet(**overrides) -> InternalContextPacket:
    payload = {
        "packet_version": "v1",
        "question": "系统里面有没有开票系统？",
        "scope": {
            "tenant_id": 9,
            "project_id": 6,
            "session_id": 101,
            "conversation_scope_key": "scope:chat:1",
            "reply_to_message_id": "om_reply_1",
            "root_message_id": "om_root_1",
        },
        "route_decision": {
            "route_mode": "internal",
            "effective_source_policy": "internal",
            "decision_origin": "query_semantics_contract",
            "reply_contract": {"mode": "explicit_source", "head": "内部知识库查询："},
            "query_profile": {
                "needs_private_context": True,
                "needs_freshness": False,
                "referenced_objects": ["project_object"],
                "confidence": 0.92,
            },
        },
        "operation_contract": {
            "query_type": "existence",
            "answer_contract": "yes_no_with_evidence",
            "object_mentions": ["开票系统"],
            "clarify_policy": "ask_when_target_ambiguous",
        },
        "target_resolution": {
            "target_entity_type": "project",
            "target_ref": "project:6",
            "resolved_label": "自动开票系统",
            "resolution_mode": "deterministic",
            "confidence": 0.96,
            "candidate_summary": [],
        },
        "query_plan": {
            "plan_version": "v2",
            "query_type": "existence",
            "entity_type": "project",
            "answer_contract": "yes_no_with_evidence",
            "target_ref": "project:6",
            "filters": [{"field": "project_id", "operator": "equals", "value": 6}],
            "scope": {"tenant_bound": True, "project_id": 6, "scope_hint": "project"},
            "retrieval_policy": {"facts_first": True, "allow_fallback_text": True},
            "verifier_checks": ["contract_matches_query_type", "evidence_backed"],
        },
        "executor_summary": {},
        "verifier_summary": {},
        "uncertainties": [],
    }
    payload.update(overrides)
    return InternalContextPacket.model_validate(payload)


def test_build_internal_query_context_packet_returns_packet_for_internal_route() -> None:
    service = _PacketBuilderService()

    packet = service._build_internal_query_context_packet(
        question="系统里面有没有开票系统？",
        project_id=6,
        session_id=101,
        conversation_scope_key="scope:chat:1",
        reply_to_message_id="om_reply_1",
        root_message_id="om_root_1",
        route_decision={
            "route_mode": "internal",
            "effective_source_policy": "internal",
            "decision_origin": "query_semantics_contract",
            "reply_contract": {"mode": "explicit_source", "head": "内部知识库查询："},
            "query_profile": {
                "needs_private_context": True,
                "needs_freshness": False,
                "referenced_objects": ["project_object"],
                "confidence": 0.92,
            },
        },
        operation_contract={
            "query_type": "existence",
            "answer_contract": "yes_no_with_evidence",
            "object_mentions": ["开票系统"],
        },
        target_resolution={
            "target_entity_type": "project",
            "target_ref": "project:6",
            "resolved_label": "自动开票系统",
            "resolution_mode": "deterministic",
            "confidence": 0.96,
        },
        query_plan={
            "plan_version": "v2",
            "query_type": "existence",
            "entity_type": "project",
            "answer_contract": "yes_no_with_evidence",
            "target_ref": "project:6",
            "filters": [{"field": "project_id", "operator": "equals", "value": 6}],
        },
    )

    assert isinstance(packet, InternalContextPacket)
    assert packet.scope["tenant_id"] == 9
    assert packet.scope["project_id"] == 6
    assert packet.route_decision["effective_source_policy"] == "internal"
    assert packet.operation_contract["query_type"] == "existence"
    assert packet.target_resolution["target_ref"] == "project:6"
    assert packet.query_plan["answer_contract"] == "yes_no_with_evidence"


def test_build_internal_query_context_packet_skips_external_route() -> None:
    service = _PacketBuilderService()

    packet = service._build_internal_query_context_packet(
        question="OpenAI 最新模型是什么？",
        project_id=None,
        session_id=101,
        conversation_scope_key="scope:chat:2",
        reply_to_message_id=None,
        root_message_id=None,
        route_decision={
            "route_mode": "external",
            "effective_source_policy": "external",
            "decision_origin": "query_semantics_contract",
            "reply_contract": {"mode": "explicit_source", "head": "外部实时查询："},
            "query_profile": {
                "needs_private_context": False,
                "needs_freshness": True,
                "referenced_objects": ["public_model"],
                "confidence": 0.95,
            },
        },
        operation_contract={},
        target_resolution={},
        query_plan=None,
    )

    assert packet is None


def test_internal_query_runtime_metrics_prefer_internal_context_packet() -> None:
    service = _RuntimeSummaryService()
    packet = _build_internal_packet()

    metrics = service._inject_internal_query_runtime_summaries(
        retrieval_metrics={"db_query_ms": 12},
        query_semantics={
            "operation_contract": {
                "query_type": "count",
                "answer_contract": "count_scalar",
                "object_mentions": ["错误镜像"],
            },
            "target_resolution": {
                "target_entity_type": "project",
                "target_ref": "project:999",
                "resolved_label": "错误项目",
                "resolution_mode": "deterministic",
                "confidence": 0.5,
            },
            "query_plan": {
                "plan_version": "v0",
                "query_type": "count",
                "entity_type": "project",
                "target_ref": "project:999",
                "answer_contract": "count_scalar",
                "filters": [],
            },
        },
        internal_context_packet=packet,
    )

    assert metrics["db_query_ms"] == 12
    assert metrics["operation_contract_summary"]["query_type"] == "existence"
    assert metrics["target_resolution_summary"]["target_ref"] == "project:6"
    assert metrics["query_plan_summary"]["plan_version"] == "v2"


def test_internal_answer_contract_verifier_prefers_internal_context_packet_contract() -> None:
    service = _VerifierService()
    packet = _build_internal_packet()

    verdict = service._verify_internal_answer_contract(
        query_type="count",
        answer="当前系统里项目总数为 42 个。",
        sources=[
            {
                "source": "db",
                "label": "Project database",
                "score": 0.99,
                "metadata": {"table": "projects"},
            }
        ],
        route_decision={"route_mode": "internal", "effective_source_policy": "internal"},
        internal_context_packet=packet,
    )

    assert verdict["ok"] is False
    assert verdict["reason_code"] == "no_total_count_substitution"


def test_internal_answer_contract_verifier_blocks_target_resolution_clarify() -> None:
    service = _VerifierService()
    packet = _build_internal_packet(
        target_resolution={
            "target_entity_type": "unknown",
            "target_ref": None,
            "resolved_label": None,
            "resolution_mode": "clarify",
            "confidence": 0.41,
            "candidate_summary": [
                {
                    "target_entity_type": "project",
                    "target_ref": "project:6",
                    "resolved_label": "自动开票系统",
                    "score": 0.62,
                    "source_kind": "semantic_facts",
                }
            ],
        }
    )

    verdict = service._verify_internal_answer_contract(
        query_type="existence",
        answer="有，当前匹配到 1 个相关项目：#6 自动开票系统（状态:active）",
        sources=[
            {
                "source": "db",
                "label": "Project database",
                "score": 0.99,
                "metadata": {"table": "projects", "project_id": 6},
            }
        ],
        route_decision={"route_mode": "internal", "effective_source_policy": "internal"},
        internal_context_packet=packet,
    )

    assert verdict["ok"] is False
    assert verdict["reason_code"] == "target_resolution_clarify"


def test_internal_answer_contract_verifier_blocks_missing_internal_context_packet_for_internal_route() -> None:
    service = _VerifierService()

    verdict = service._verify_internal_answer_contract(
        query_type="existence",
        answer="有，当前匹配到 1 个相关项目：#6 自动开票系统（状态:active）",
        sources=[
            {
                "source": "db",
                "label": "Project database",
                "score": 0.99,
                "metadata": {"table": "projects", "project_id": 6},
            }
        ],
        route_decision={"route_mode": "internal", "effective_source_policy": "internal"},
        internal_context_packet=None,
    )

    assert verdict["ok"] is False
    assert verdict["reason_code"] == "missing_internal_context_packet"


def test_internal_answer_contract_verifier_blocks_missing_query_plan_contract() -> None:
    service = _VerifierService()
    packet = _build_internal_packet(
        query_plan={
            "plan_version": "v2",
            "query_type": "existence",
            "entity_type": "project",
            "answer_contract": "",
            "target_ref": "project:6",
            "filters": [{"field": "project_id", "operator": "equals", "value": 6}],
        }
    )

    verdict = service._verify_internal_answer_contract(
        query_type="existence",
        answer="有，当前匹配到 1 个相关项目：#6 自动开票系统（状态:active）",
        sources=[
            {
                "source": "db",
                "label": "Project database",
                "score": 0.99,
                "metadata": {"table": "projects", "project_id": 6},
            }
        ],
        route_decision={"route_mode": "internal", "effective_source_policy": "internal"},
        internal_context_packet=packet,
    )

    assert verdict["ok"] is False
    assert verdict["reason_code"] == "missing_internal_query_plan_contract"


def test_collect_structured_object_mentions_prefers_internal_context_packet() -> None:
    packet = _build_internal_packet(
        operation_contract={
            "query_type": "existence",
            "answer_contract": "yes_no_with_evidence",
            "object_mentions": ["真实对象"],
        },
        target_resolution={
            "target_entity_type": "project",
            "target_ref": "project:6",
            "resolved_label": "真实项目",
            "resolution_mode": "deterministic",
            "confidence": 0.96,
            "candidate_summary": [],
        },
    )

    mentions = _ProjectUtilsService._collect_structured_object_mentions(
        query_semantics={
            "operation_contract": {
                "query_type": "count",
                "answer_contract": "count_scalar",
                "object_mentions": ["错误对象"],
            },
            "target_resolution": {
                "target_entity_type": "project",
                "target_ref": "project:999",
                "resolved_label": "错误项目",
                "resolution_mode": "deterministic",
            },
        },
        internal_context_packet=packet,
    )

    assert mentions[:2] == ["真实对象", "真实项目"]


@pytest.mark.asyncio
async def test_internal_model_config_fast_answer_prefers_internal_context_packet() -> None:
    class _FakeLLM:
        async def _get_or_init_bundle(self, db, *, tenant_id: int):
            _ = (db, tenant_id)
            return object(), {
                "active_engine_id": "low-1",
                "engines": [
                    {
                        "id": "low-1",
                        "name": "Qwen 低阶引擎",
                        "provider": "dashscope",
                        "model": "qwen-plus",
                    }
                ],
                "settings": {
                    "enable_local_kb_qa": False,
                },
            }

        def _resolve_active_engine_from_bundle(self, bundle: dict[str, object]):
            engines = list(bundle.get("engines") or [])
            return dict(engines[0])

        async def _resolve_engine_for_call(self, db, *, tenant_id: int, function_name: str, routing_meta=None):
            _ = (db, tenant_id, function_name, routing_meta)
            return {
                "name": "Qwen 低阶引擎",
                "provider": "dashscope",
                "model": "qwen-plus",
                "_routing_meta": {"planned_tier": "low", "executed_tier": "low"},
            }

    class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
        def __init__(self) -> None:
            self.db = object()
            self.tenant_id = 1
            self.llm = _FakeLLM()

    service = _FakeService()
    packet = _build_internal_packet(
        question="我们系统当前接的是哪个模型配置？",
        route_decision={
            "route_mode": "internal",
            "effective_source_policy": "internal",
            "decision_origin": "query_semantics_contract",
            "reply_contract": {"mode": "explicit_source", "head": "内部知识库查询："},
            "query_profile": {
                "needs_private_context": True,
                "needs_freshness": False,
                "referenced_objects": ["system_config", "model_routing"],
                "confidence": 0.92,
            },
        },
        operation_contract={
            "query_type": "status",
            "answer_contract": "entity_snapshot",
            "object_mentions": ["当前模型配置"],
        },
        target_resolution={
            "target_entity_type": "config",
            "target_ref": "config:model_routing",
            "resolved_label": "当前模型配置",
            "resolution_mode": "deterministic",
            "confidence": 0.98,
            "candidate_summary": [],
        },
        query_plan={
            "plan_version": "v2",
            "query_type": "status",
            "entity_type": "config",
            "answer_contract": "entity_snapshot",
            "target_ref": "config:model_routing",
            "filters": [],
        },
    )

    fast_result = await service._build_internal_model_config_fast_answer(
        question="我们系统当前接的是哪个模型配置？",
        query_semantics={
            "effective_source_policy": "internal",
            "query_type": "existence",
            "operation_contract": {
                "query_type": "existence",
                "answer_contract": "yes_no_with_evidence",
            },
            "query_profile": {
                "needs_private_context": True,
                "needs_freshness": False,
                "referenced_objects": [],
            },
            "target_resolution": {
                "target_entity_type": "project",
                "target_ref": "project:72",
            },
        },
        internal_context_packet=packet,
    )

    assert isinstance(fast_result, dict)
    assert "qwen-plus" in str(fast_result["answer"])
    assert fast_result["query_type"] == "status"