X-Financial/server/tests/test_steward_slot_decision_agent.py

from __future__ import annotations

from app.schemas.steward import StewardSlotDecisionRequest
from app.services.runtime_chat import RuntimeChatCallTrace, RuntimeChatToolCall, RuntimeToolCallResult
from app.services.steward_slot_decision_agent import (
    STEWARD_SLOT_DECISION_FUNCTION_NAME,
    StewardSlotDecisionAgent,
)


class FakeSlotRuntime:
    def __init__(self, arguments=None):
        self.arguments = arguments
        self.messages = None

    def complete_with_tool_call(self, messages, **kwargs):
        self.messages = messages
        if self.arguments is None:
            return RuntimeToolCallResult(tool_call=None, calls=[])
        return RuntimeToolCallResult(
            tool_call=RuntimeChatToolCall(
                name=STEWARD_SLOT_DECISION_FUNCTION_NAME,
                arguments=self.arguments,
            ),
            calls=[
                RuntimeChatCallTrace(
                    slot="main",
                    provider="OpenAI Compatible",
                    model="fake",
                    attempt=1,
                    status="succeeded",
                )
            ],
        )


def test_steward_slot_decision_uses_function_calling_result() -> None:
    runtime = FakeSlotRuntime(
        {
            "next_action": "ask_user",
            "required_fields": ["expense_type", "time_range", "location", "reason", "transport_mode"],
            "missing_fields": ["transport_mode"],
            "question": "请问你这次打算怎么出行？",
            "options": [
                {"field_key": "transport_mode", "label": "飞机", "value": "飞机"},
                {"field_key": "transport_mode", "label": "火车", "value": "火车"},
            ],
            "rationale": "出行方式会影响交通费用测算。",
        }
    )

    result = StewardSlotDecisionAgent(runtime).decide(
        StewardSlotDecisionRequest(
            task_type="expense_application",
            user_message="明天出差北京3天，支撑国网仿生产部署",
            ontology_fields={
                "expense_type": "travel",
                "time_range": "2026-06-05 至 2026-06-07",
                "location": "北京",
                "reason": "支撑国网仿生产部署",
            },
            missing_fields=["transport_mode"],
        )
    )

    assert result.decision_source == "llm_function_call"
    assert result.next_action == "ask_user"
    assert result.missing_fields == ["transport_mode"]
    assert [item.value for item in result.options] == ["飞机", "火车"]
    assert "出行方式会影响" in result.rationale


def test_steward_slot_decision_falls_back_to_intent_missing_fields_only() -> None:
    runtime = FakeSlotRuntime(arguments=None)

    result = StewardSlotDecisionAgent(runtime).decide(
        StewardSlotDecisionRequest(
            task_type="expense_application",
            user_message="还需要补充：出行方式（例如高铁、飞机、自驾、出租车）",
            ontology_fields={
                "expense_type": "travel",
                "location": "北京",
                "reason": "支撑国网仿生产部署",
            },
            missing_fields=["transport_mode"],
        )
    )

    assert result.decision_source == "rule_fallback"
    assert result.next_action == "ask_user"
    assert result.missing_fields == ["transport_mode"]
    assert [item.value for item in result.options] == ["火车", "飞机", "轮船"]
    assert "高铁" not in result.required_fields


def test_steward_slot_decision_does_not_ask_user_for_application_profile_or_attachments() -> None:
    runtime = FakeSlotRuntime(
        {
            "next_action": "ask_user",
            "required_fields": [
                "expense_type",
                "time_range",
                "location",
                "reason",
                "amount",
                "attachments",
                "employee_no",
            ],
            "missing_fields": ["attachments", "employee_no"],
            "question": "请补充附件和员工编号。",
            "options": [],
            "rationale": "附件/凭证和员工编号为合规必需字段。",
        }
    )

    result = StewardSlotDecisionAgent(runtime).decide(
        StewardSlotDecisionRequest(
            task_type="expense_application",
            user_message="明天出差北京3天，支撑国网仿生产部署",
            ontology_fields={
                "expense_type": "travel",
                "time_range": "2026-06-05 至 2026-06-07",
                "location": "北京",
                "reason": "支撑国网仿生产部署",
            },
            missing_fields=["attachments", "employee_no"],
        )
    )

    assert result.decision_source == "llm_function_call"
    assert result.next_action == "render_preview"
    assert result.missing_fields == []
    assert "attachments" not in result.required_fields
    assert "employee_no" not in result.required_fields
    assert result.options == []
    assert "合规必需字段" not in result.rationale