X-Financial/server/tests/test_steward_intent_agent.py

from app.services.steward_intent_agent import (
    STEWARD_INTENT_FUNCTION_NAME,
    StewardIntentAgent,
)
from app.schemas.steward import StewardPlanRequest


class _NoToolCallRuntimeChatService:
    def __init__(self) -> None:
        self.kwargs = {}

    def complete_with_tool_call(self, messages, **kwargs):
        self.kwargs = kwargs

        class _Result:
            tool_call = None

            @staticmethod
            def calls_as_dicts():
                return []

        return _Result()


def test_steward_intent_tool_schema_supports_pending_flow_confirmation() -> None:
    schema = StewardIntentAgent._build_intent_tool_schema(
        ["expense_type", "time_range", "location", "reason", "transport_mode"]
    )

    function_schema = schema["function"]
    assert function_schema["name"] == STEWARD_INTENT_FUNCTION_NAME
    properties = function_schema["parameters"]["properties"]
    task_schema = properties["tasks"]["items"]
    pending_schema = properties["pending_flow_confirmation"]
    candidate_schema = pending_schema["properties"]["candidate_flows"]["items"]

    assert task_schema["properties"]["requested_action"]["enum"] == [
        "preview",
        "save_draft",
        "submit",
    ]
    assert "requested_action" in task_schema["required"]
    assert "pending_flow_confirmation" in properties
    assert pending_schema["properties"]["status"]["enum"] == ["none", "pending"]
    assert candidate_schema["properties"]["flow_id"]["enum"] == [
        "travel_application",
        "travel_reimbursement",
    ]
    assert candidate_schema["properties"]["missing_fields"]["items"]["enum"] == [
        "expense_type",
        "time_range",
        "location",
        "reason",
        "transport_mode",
    ]


def test_steward_intent_agent_uses_ten_second_timeout_and_three_attempts() -> None:
    runtime_chat = _NoToolCallRuntimeChatService()
    agent = StewardIntentAgent(runtime_chat)

    agent.detect(
        StewardPlanRequest(message="2026-02-20 至 2026-02-23，上海出差，火车，保存草稿。"),
        base_date=__import__("datetime").date(2026, 6, 24),
        canonical_fields=["expense_type", "time_range", "location", "reason", "transport_mode"],
    )

    assert runtime_chat.kwargs["timeout_seconds"] == 10
    assert runtime_chat.kwargs["max_attempts"] == 3
    assert runtime_chat.kwargs["use_failure_cooldown"] is False


def test_steward_intent_tool_schema_includes_query_task_type_from_registry() -> None:
    """function call schema 的 task_type enum 应从注册表动态生成,包含查询意图。"""
    from app.services import steward_intent_bootstrap  # noqa: F401  触发意图注册

    schema = StewardIntentAgent._build_intent_tool_schema(
        ["expense_type", "time_range", "location", "reason", "transport_mode"]
    )
    task_schema = schema["function"]["parameters"]["properties"]["tasks"]["items"]
    task_type_enum = task_schema["properties"]["task_type"]["enum"]

    assert "expense_application" in task_type_enum
    assert "reimbursement" in task_type_enum
    assert "query_travel_standard" in task_type_enum


def test_steward_intent_system_prompt_mentions_query_intent_guidance() -> None:
    """system prompt 应包含查询意图的识别指引,避免被误识别为申请。"""
    from app.services import steward_intent_bootstrap  # noqa: F401  触发意图注册

    messages = StewardIntentAgent._build_messages(
        StewardPlanRequest(message="武汉出差标准是多少"),
        base_date=__import__("datetime").date(2026, 6, 24),
        canonical_fields=["location", "employee_grade"],
    )
    system_prompt = messages[0]["content"]
    assert "query_travel_standard" in system_prompt
    assert "差旅" in system_prompt
    assert "住宿标准" in system_prompt


def test_steward_intent_system_prompt_includes_conversation_history_guidance() -> None:
    """system prompt 应包含'结合对话历史理解确认类话术'的引导。"""
    from app.services import steward_intent_bootstrap  # noqa: F401

    messages = StewardIntentAgent._build_messages(
        StewardPlanRequest(message="再提交"),
        base_date=__import__("datetime").date(2026, 6, 24),
        canonical_fields=["location", "time_range"],
    )
    system_prompt = messages[0]["content"]
    assert "recent_history" in system_prompt
    assert "再提交" in system_prompt
    assert "确认类话术" in system_prompt


def test_steward_intent_context_payload_includes_recent_history() -> None:
    """context_payload 应携带 recent_history 结构化字段（role + content）。"""
    import json

    request = StewardPlanRequest(
        message="再提交",
        context_json={
            "recent_history": [
                {"role": "user", "content": "2026-02-20 至 2026-02-23，去上海出差，火车"},
                {"role": "assistant", "content": "好的，为您整理出差申请预览。"},
                {"role": "user", "content": "直接提交"},
                {"role": "assistant", "content": "检测到重复申请，已暂停提交。"},
            ],
        },
    )
    messages = StewardIntentAgent._build_messages(
        request,
        base_date=__import__("datetime").date(2026, 6, 24),
        canonical_fields=["location", "time_range"],
    )
    user_payload = json.loads(messages[1]["content"])
    assert "recent_history" in user_payload
    assert len(user_payload["recent_history"]) == 4
    assert user_payload["recent_history"][0]["role"] == "user"
    assert "上海" in user_payload["recent_history"][0]["content"]


def test_steward_intent_context_payload_omits_empty_recent_history() -> None:
    """无 recent_history 时不应注入空列表。"""
    import json

    messages = StewardIntentAgent._build_messages(
        StewardPlanRequest(message="你好"),
        base_date=__import__("datetime").date(2026, 6, 24),
        canonical_fields=["location"],
    )
    user_payload = json.loads(messages[1]["content"])
    assert user_payload.get("recent_history", []) == []