X-Financial/server/tests/test_agent_asset_service.py

from __future__ import annotations

import shutil
import uuid
from io import BytesIO
from pathlib import Path

import pytest
from openpyxl import Workbook, load_workbook
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.api.deps import CurrentUserContext
from app.core.agent_enums import (
    AgentAssetContentType,
    AgentAssetDomain,
    AgentAssetStatus,
    AgentAssetType,
    AgentName,
    AgentReviewStatus,
    AgentRunSource,
    AgentRunStatus,
)
from app.core.config import SERVER_DIR
from app.db.base import Base
from app.models.agent_asset import AgentAsset
from app.models.employee import Employee
from app.schemas.agent_asset import (
    AgentAssetCreate,
    AgentAssetReviewCreate,
    AgentAssetVersionCreate,
)
from app.schemas.reimbursement import TravelReimbursementCalculatorRequest
from app.services import agent_foundation as agent_foundation_module
from app.services.agent_asset_spreadsheet import (
    COMPANY_COMMUNICATION_EXPENSE_RULE_CODE,
    COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME,
    COMPANY_PREAPPROVAL_RULE_CODE,
    COMPANY_PREAPPROVAL_RULE_FILENAME,
    COMPANY_TRAVEL_GRADE_MAPPING_RULE_CODE,
    COMPANY_TRAVEL_GRADE_MAPPING_RULE_FILENAME,
    COMPANY_TRAVEL_EXPENSE_RULE_CODE,
    COMPANY_TRAVEL_EXPENSE_RULE_FILENAME,
    COMPANY_TRAVEL_SEASON_MAPPING_RULE_CODE,
    COMPANY_TRAVEL_SEASON_MAPPING_RULE_FILENAME,
    COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_CODE,
    COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_FILENAME,
    FINANCE_RULES_LIBRARY,
)
from app.services.agent_foundation_constants import COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON
from app.services.agent_assets import AgentAssetService
from app.services.agent_runs import AgentRunService
from app.services.audit import AuditLogService
from app.services.expense_rule_runtime import ExpenseRuleRuntimeService
from app.services.finance_rule_catalog import (
    DEPRECATED_FINANCE_RULE_CODES,
)
from app.services.settings import OnlyOfficeRuntimeConfig
from app.services.travel_reimbursement_calculator import TravelReimbursementCalculatorService


@pytest.fixture(autouse=True)
def isolate_rule_file_storage(tmp_path, monkeypatch) -> None:
    temp_server_dir = tmp_path / "server"
    temp_rules_root = temp_server_dir / "rules"
    temp_finance_rules = temp_rules_root / FINANCE_RULES_LIBRARY
    temp_finance_rules.mkdir(parents=True, exist_ok=True)

    real_finance_rules = SERVER_DIR / "rules" / FINANCE_RULES_LIBRARY
    for file_name in (
        COMPANY_TRAVEL_EXPENSE_RULE_FILENAME,
        COMPANY_TRAVEL_GRADE_MAPPING_RULE_FILENAME,
        COMPANY_TRAVEL_SEASON_MAPPING_RULE_FILENAME,
        COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_FILENAME,
        COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME,
        COMPANY_PREAPPROVAL_RULE_FILENAME,
    ):
        source_path = real_finance_rules / file_name
        if source_path.exists():
            shutil.copy2(source_path, temp_finance_rules / file_name)

    monkeypatch.setattr(
        "app.services.agent_asset_spreadsheet.SERVER_DIR",
        temp_server_dir,
    )

    def init_manager(self, storage_root=None, rule_root=None) -> None:
        self.storage_root = Path(storage_root or tmp_path / "storage").resolve()
        self.asset_root = (self.storage_root / "agent_assets").resolve()
        self.rule_root = Path(rule_root or temp_rules_root).resolve()

    monkeypatch.setattr(
        "app.services.agent_asset_spreadsheet.AgentAssetSpreadsheetManager.__init__",
        init_manager,
    )


def build_session() -> Session:
    agent_foundation_module._foundation_ready_keys.clear()
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    return session_factory()


def build_workbook_bytes(rows: list[list[object]], *, sheet_name: str = "规则表") -> bytes:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = sheet_name
    for row in rows:
        sheet.append(row)
    buffer = BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()


def build_multi_sheet_workbook_bytes(sheets: dict[str, list[list[object]]]) -> bytes:
    workbook = Workbook()
    default_sheet = workbook.active
    workbook.remove(default_sheet)
    for sheet_name, rows in sheets.items():
        sheet = workbook.create_sheet(sheet_name)
        for row in rows:
            sheet.append(row)
    buffer = BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()


def test_agent_asset_service_seeds_assets_and_enforces_review_before_activation() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rules = service.list_assets(asset_type=AgentAssetType.RULE.value)
        assert len(rules) >= 3
        assert any(
            item.code == "rule.expense.travel_risk_control_standard"
            and item.status == AgentAssetStatus.ACTIVE.value
            for item in rules
        )
        assert all(
            item.code
            not in {
                "rule.expense.duplicate_expense_check",
                "rule.expense.travel_receipt_requirements",
                "rule.ap.payment_dual_review",
            }
            for item in rules
        )

        pending_rule = next(item for item in rules if item.status == AgentAssetStatus.REVIEW.value)

        with pytest.raises(PermissionError):
            service.activate_asset(pending_rule.id, actor="pytest")


def test_agent_asset_service_seeds_all_foundation_asset_types() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        assert len(service.list_assets(asset_type=AgentAssetType.RULE.value)) >= 3
        assert len(service.list_assets(asset_type=AgentAssetType.SKILL.value)) >= 2
        assert len(service.list_assets(asset_type=AgentAssetType.MCP.value)) >= 2
        assert len(service.list_assets(asset_type=AgentAssetType.TASK.value)) >= 3


def test_agent_asset_service_supports_backend_pagination() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        page = service.list_assets_page(
            asset_type=AgentAssetType.RULE.value,
            page=1,
            page_size=2,
        )

        assert len(page.items) <= 2
        assert page.total >= len(page.items)
        assert page.page == 1
        assert page.page_size == 2


def test_finance_rules_use_risk_rule_scenario_categories() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rules = service.list_assets(asset_type=AgentAssetType.RULE.value)
        travel_rule = next(item for item in rules if item.code == COMPANY_TRAVEL_EXPENSE_RULE_CODE)
        communication_rule = next(
            item for item in rules if item.code == COMPANY_COMMUNICATION_EXPENSE_RULE_CODE
        )
        preapproval_rule = next(item for item in rules if item.code == COMPANY_PREAPPROVAL_RULE_CODE)
        travel_config = travel_rule.config_json or {}
        communication_config = communication_rule.config_json or {}
        preapproval_config = preapproval_rule.config_json or {}

        assert travel_rule.scenario_json == ["差旅费"]
        assert travel_config["scenario_category"] == "差旅费"
        assert travel_config["ai_review_category"] == "差旅费"
        assert communication_rule.scenario_json == ["通信费"]
        assert communication_config["scenario_category"] == "通信费"
        assert communication_config["ai_review_category"] == "通信费"
        assert preapproval_rule.scenario_json == list(COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON)
        assert travel_config["tag"] == "基础规则"
        assert communication_config["tag"] == "基础规则"
        assert preapproval_config["tag"] == "申请规则"
        assert preapproval_config["finance_rule_code"] == "expense.preapproval.policy"
        assert preapproval_config["finance_rule_sheet"] == "费用申请审批规则"
        assert preapproval_config["expense_types"] == ["meal", "entertainment", "office", "all"]
        assert preapproval_config["rule_document"]["file_name"] == COMPANY_PREAPPROVAL_RULE_FILENAME

        grade_mapping_rule = next(
            item for item in rules if item.code == COMPANY_TRAVEL_GRADE_MAPPING_RULE_CODE
        )
        season_mapping_rule = next(
            item for item in rules if item.code == COMPANY_TRAVEL_SEASON_MAPPING_RULE_CODE
        )
        transport_estimate_rule = next(
            item for item in rules if item.code == COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_CODE
        )
        assert grade_mapping_rule.config_json["tag"] == "基础规则"
        assert grade_mapping_rule.config_json["rule_document"]["file_name"] == (
            COMPANY_TRAVEL_GRADE_MAPPING_RULE_FILENAME
        )
        assert season_mapping_rule.config_json["tag"] == "基础规则"
        assert season_mapping_rule.config_json["rule_document"]["file_name"] == (
            COMPANY_TRAVEL_SEASON_MAPPING_RULE_FILENAME
        )
        assert transport_estimate_rule.config_json["tag"] == "基础规则"
        assert transport_estimate_rule.config_json["rule_document"]["file_name"] == (
            COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_FILENAME
        )


def test_non_standard_finance_rule_spreadsheets_are_not_seeded() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        service.list_assets(asset_type=AgentAssetType.RULE.value)

        for code in DEPRECATED_FINANCE_RULE_CODES:
            asset = db.scalar(select(AgentAsset).where(AgentAsset.code == code))
            assert asset is None or asset.config_json["tag"] == "废弃规则"


def test_demo_budget_risk_rules_are_excluded_from_risk_rule_center() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        service.list_assets(asset_type=AgentAssetType.RULE.value)

        budget_rule = db.scalar(
            select(AgentAsset).where(
                AgentAsset.code == "risk.budget.available_balance_insufficient"
            )
        )
        communication_rule = db.scalar(
            select(AgentAsset).where(
                AgentAsset.code == "risk.standard.communication_amount_over_policy"
            )
        )

        assert budget_rule is None

        assert communication_rule is not None
        assert communication_rule.scenario_json == ["通信费"]
        assert communication_rule.config_json["finance_rule_code"] == "expense.communication.policy"
        assert communication_rule.config_json["finance_rule_sheet"] == "通信费报销规则"
        assert communication_rule.config_json["expense_types"] == ["communication"]
        assert communication_rule.config_json["budget_required"] is True


def test_existing_budget_risk_assets_are_hidden_from_rule_lists() -> None:
    with build_session() as db:
        db.add(
            AgentAsset(
                asset_type=AgentAssetType.RULE.value,
                code="risk.budget.legacy.visible",
                name="历史预算风险",
                description="旧数据中已经存在的预算风险规则。",
                domain=AgentAssetDomain.EXPENSE.value,
                scenario_json=["全部"],
                owner="pytest",
                status=AgentAssetStatus.ACTIVE.value,
                config_json={
                    "detail_mode": "json_risk",
                    "finance_rule_code": "budget.execution.policy",
                    "rule_document": {"file_name": "risk.budget.available_balance_insufficient.json"},
                },
            )
        )
        db.commit()

        service = AgentAssetService(db)
        listed_codes = {
            item.code for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
        }
        page = service.list_assets_page(
            asset_type=AgentAssetType.RULE.value,
            status=None,
            domain=None,
            keyword=None,
            page=1,
            page_size=100,
        )

        assert "risk.budget.legacy.visible" not in listed_codes
        assert "risk.budget.legacy.visible" not in {item.code for item in page.items}


def test_agent_asset_service_can_activate_rule_after_review() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        created = service.create_asset(
            AgentAssetCreate(
                asset_type=AgentAssetType.RULE,
                code=f"rule.test.{uuid.uuid4().hex[:8]}",
                name="测试规则",
                description="用于测试审核和上线流程。",
                domain=AgentAssetDomain.EXPENSE,
                scenario_json=["expense", "risk_check"],
                owner="pytest",
                reviewer="reviewer",
                status=AgentAssetStatus.DRAFT,
                config_json={"enabled": False},
            ),
            actor="pytest",
        )
        service.create_version(
            created.id,
            AgentAssetVersionCreate(
                version="v1.0.0",
                content="# 测试规则\n\n- 仅用于测试。",
                content_type=AgentAssetContentType.MARKDOWN,
                change_note="初始化版本",
                created_by="pytest",
            ),
            actor="pytest",
        )
        service.create_review(
            created.id,
            AgentAssetReviewCreate(
                version="v1.0.0",
                reviewer="reviewer",
                review_status=AgentReviewStatus.APPROVED,
                review_note="可以上线",
            ),
            actor="reviewer",
        )

        activated = service.activate_asset(created.id, actor="reviewer")

        assert activated.status == AgentAssetStatus.ACTIVE.value
        assert activated.current_version == "v1.0.0"
        assert activated.working_version == "v1.0.0"
        assert activated.published_version == "v1.0.0"
        assert activated.latest_review is not None
        assert activated.latest_review.review_status == AgentReviewStatus.APPROVED.value


def test_rule_working_version_does_not_replace_published_version_until_activation() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )

        created = service.create_version(
            rule.id,
            AgentAssetVersionCreate(
                version="v1.1.1",
                content="# 差旅报销风险管控制度\n\n- 工作稿",
                content_type=AgentAssetContentType.MARKDOWN,
                change_note="新增工作稿",
                created_by="finance_user",
            ),
            actor="finance_user",
        )
        detail = service.get_asset(rule.id)

        assert created.is_working is True
        assert created.is_published is False
        assert created.lifecycle_state == "draft"
        assert detail is not None
        assert detail.status == AgentAssetStatus.ACTIVE.value
        assert detail.current_version == "v1.1.1"
        assert detail.working_version == "v1.1.1"
        assert detail.published_version == "v1.1.0"
        assert detail.latest_review is None


def test_pending_review_can_name_new_working_version_before_submission() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )

        review = service.create_review(
            rule.id,
            AgentAssetReviewCreate(
                version="v1.2.0",
                reviewer="manager_user",
                review_status=AgentReviewStatus.PENDING,
                review_note="请审核",
            ),
            actor="finance_user",
        )
        detail = service.get_asset(rule.id)

        assert review.version == "v1.2.0"
        assert detail is not None
        assert detail.current_version == "v1.2.0"
        assert detail.working_version == "v1.2.0"
        assert detail.published_version == "v1.1.0"
        assert detail.latest_review is not None
        assert detail.latest_review.reviewer == "manager_user"


def test_expense_rule_runtime_uses_published_version_instead_of_working_version() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )

        service.create_version(
            rule.id,
            AgentAssetVersionCreate(
                version="v1.1.1",
                content=(
                    "# 工作稿\n\n"
                    '```expense-rule\n{"kind":"travel_policy","version":1}\n```'
                ),
                content_type=AgentAssetContentType.MARKDOWN,
                change_note="未上线草稿",
                created_by="finance_user",
            ),
            actor="finance_user",
        )

        catalog = ExpenseRuleRuntimeService(db).load_catalog()

        assert catalog.travel_policy is not None
        assert catalog.travel_policy.rule_version == "v1.1.0"


def test_restore_version_creates_new_working_copy_without_rewriting_published_version() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )

        restored = service.restore_version_as_working_copy(
            rule.id,
            "v1.0.0",
            actor="manager_user",
        )

        assert restored.working_version == "v1.1.1"
        assert restored.current_version == "v1.1.1"
        assert restored.published_version == "v1.1.0"
        assert restored.current_version_change_note == "基于历史版本 v1.0.0 恢复生成工作稿"


def test_spreadsheet_upload_records_sheet_and_cell_changes_without_versions() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.company_travel_expense_reimbursement"
        )

        service.upload_rule_spreadsheet(
            rule.id,
            filename="公司差旅费报销规则.xlsx",
                content=build_workbook_bytes([["城市", "住宿"], ["北京", 500]]),
            actor="finance_user",
        )
        service.upload_rule_spreadsheet(
            rule.id,
            filename="公司差旅费报销规则.xlsx",
            content=build_workbook_bytes([["城市", "住宿"], ["北京", 550], ["武汉", 450]]),
            actor="finance_user",
        )

        records = service.list_spreadsheet_change_records(rule.id)
        latest = records[0]

        assert latest.changed_sheet_count == 1
        assert latest.changed_cell_count == 3
        assert any(
            item.cell == "B2" and item.change_type == "modified"
            for item in latest.cell_changes
        )
        assert any(
            item.cell == "A3" and item.change_type == "added"
            for item in latest.cell_changes
        )
        assert not hasattr(latest, "version")


def test_spreadsheet_content_reads_current_rule_file_without_version_snapshot() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.company_travel_expense_reimbursement"
        )

        service.upload_rule_spreadsheet(
            rule.id,
            filename="公司差旅费报销规则.xlsx",
            content=build_workbook_bytes([["城市", "住宿"], ["北京", 500]]),
            actor="finance_user",
        )
        detail = service.get_asset(rule.id)
        assert detail is not None

        current_asset = service.repository.get(rule.id)
        assert current_asset is not None
        live_storage_key = str((current_asset.config_json or {})["rule_document"]["storage_key"])
        assert live_storage_key.startswith(f"rules/{FINANCE_RULES_LIBRARY}/")
        assert "agent_assets" not in live_storage_key
        live_path = service.spreadsheet_manager.resolve_storage_path(live_storage_key)
        assert not service.spreadsheet_manager.asset_root.exists()

        current_path, _, _ = service.get_rule_spreadsheet_content(rule.id)

        assert current_path == live_path
        assert ".versions" not in current_path.parts
        workbook = load_workbook(current_path, data_only=False)
        assert workbook.active["B2"].value == 500


def test_spreadsheet_change_records_return_recent_edit_details() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.company_travel_expense_reimbursement"
        )
        service.audit_service.log_action(
            actor="manager_user",
            action="edit_rule_spreadsheet",
            resource_type=rule.asset_type,
            resource_id=rule.id,
            after_json={
                "summary": "在线编辑：共 1 处改动。",
                "changed_sheet_count": 1,
                "changed_cell_count": 1,
                "sheet_changes": [],
                "cell_changes": [
                    {
                        "sheet_name": "规则表",
                        "cell": "B2",
                        "change_type": "modified",
                        "before_value": 500,
                        "after_value": 550,
                    }
                ],
            },
        )

        records = service.list_spreadsheet_change_records(rule.id)

        assert len(records) == 1
        assert records[0].actor == "manager_user"
        assert records[0].changed_cell_count == 1
        assert records[0].cell_changes[0].cell == "B2"


def test_spreadsheet_change_records_include_all_modified_sheets() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.company_travel_expense_reimbursement"
        )

        service.upload_rule_spreadsheet(
            rule.id,
            filename="公司差旅费报销规则.xlsx",
            content=build_multi_sheet_workbook_bytes(
                {
                    "差旅标准": [["城市", "住宿"], ["北京", 500]],
                    "填表说明": [["字段", "说明"], ["住宿", "按城市标准"]],
                }
            ),
            actor="finance_user",
        )
        detail = service.get_asset(rule.id)
        assert detail is not None

        service.upload_rule_spreadsheet(
            rule.id,
            filename="公司差旅费报销规则.xlsx",
            content=build_multi_sheet_workbook_bytes(
                {
                    "差旅标准": [["城市", "住宿"], ["北京", 550]],
                    "填表说明": [["字段", "说明"], ["住宿", "按城市等级标准"]],
                }
            ),
            actor="finance_user",
        )

        records = service.list_spreadsheet_change_records(rule.id)
        latest = records[0]
        changed_sheets = {item.sheet_name for item in latest.sheet_changes}
        changed_cell_sheets = {item.sheet_name for item in latest.cell_changes}

        assert not hasattr(latest, "version")
        assert latest.changed_sheet_count == 2
        assert {"差旅标准", "填表说明"}.issubset(changed_sheets)
        assert {"差旅标准", "填表说明"}.issubset(changed_cell_sheets)
        assert "差旅标准" in latest.summary
        assert "填表说明" in latest.summary


def test_editable_spreadsheet_onlyoffice_config_enables_forcesave(monkeypatch) -> None:
    with build_session() as db:
        monkeypatch.setattr(
            "app.services.agent_asset_onlyoffice.resolve_onlyoffice_settings",
            lambda: OnlyOfficeRuntimeConfig(
                enabled=True,
                public_url="http://onlyoffice.example.com",
                backend_url="http://backend.example.com",
                jwt_secret="secret",
            ),
        )

        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.company_travel_expense_reimbursement"
        )

        config = service.build_rule_spreadsheet_onlyoffice_config(
            rule.id,
            CurrentUserContext(
                username="finance_user",
                name="财务人员",
                role_codes=["finance"],
                is_admin=False,
            ),
        )

        customization = config.config["editorConfig"]["customization"]
        assert config.config["editorConfig"]["mode"] == "edit"
        assert customization["forcesave"] is True
        assert "version=" not in config.config["document"]["url"]
        assert "version=" not in config.config["editorConfig"]["callbackUrl"]


def test_version_timeline_contains_created_review_and_publish_events() -> None:
    with build_session() as db:
        service = AgentAssetService(db)
        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )
        service.create_version(
            rule.id,
            AgentAssetVersionCreate(
                version="v1.1.1",
                content="# 工作稿",
                content_type=AgentAssetContentType.MARKDOWN,
                change_note="补充口径",
                created_by="finance_user",
            ),
            actor="finance_user",
        )
        service.create_review(
            rule.id,
            AgentAssetReviewCreate(
                version="v1.1.1",
                reviewer="finance_user",
                review_status=AgentReviewStatus.PENDING,
                review_note="请审核",
            ),
            actor="finance_user",
        )
        service.create_review(
            rule.id,
            AgentAssetReviewCreate(
                version="v1.1.1",
                reviewer="manager_user",
                review_status=AgentReviewStatus.APPROVED,
                review_note="可以上线",
            ),
            actor="manager_user",
        )
        service.activate_asset(rule.id, actor="manager_user")

        timeline = service.list_version_timeline(rule.id)
        event_types = [item.event_type for item in timeline if item.version == "v1.1.1"]

        assert "created" in event_types
        assert "submitted" in event_types
        assert "approved" in event_types
        assert "published" in event_types


def test_agent_asset_service_returns_recent_versions_for_rule_detail() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.attachment_submission_requirements"
        )
        detail = service.get_asset(rule.id)

        assert detail is not None
        assert detail.current_version == "v1.0.0"
        assert detail.current_version_content_type == AgentAssetContentType.MARKDOWN.value
        assert isinstance(detail.current_version_content, str)
        assert len(detail.recent_versions) >= 2
        assert any(item.is_current for item in detail.recent_versions)
        assert {item.version for item in detail.recent_versions} >= {"v0.9.0", "v1.0.0"}
        assert detail.config_json["rule_template_key"] == "attachment_requirement_v1"
        assert "附件或单据不完整" in str(detail.current_version_content)


def test_agent_asset_service_returns_travel_policy_rule_detail() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        rule = next(
            item
            for item in service.list_assets(asset_type=AgentAssetType.RULE.value)
            if item.code == "rule.expense.travel_risk_control_standard"
        )
        detail = service.get_asset(rule.id)

        assert detail is not None
        assert detail.status == AgentAssetStatus.ACTIVE.value
        assert detail.current_version == "v1.1.0"
        assert detail.latest_review is not None
        assert detail.latest_review.review_status == AgentReviewStatus.APPROVED.value
        assert "行程闭环" in str(detail.current_version_content)
        assert "住宿标准、飞机舱位和火车席别" in str(detail.current_version_content)


def test_expense_rule_runtime_reads_amount_standards_from_travel_spreadsheet() -> None:
    with build_session() as db:
        AgentAssetService(db).list_assets(asset_type=AgentAssetType.RULE.value)
        travel_spreadsheet_rule = db.scalar(
            select(AgentAsset).where(AgentAsset.code == COMPANY_TRAVEL_EXPENSE_RULE_CODE)
        )
        assert travel_spreadsheet_rule is not None
        travel_spreadsheet_rule.status = AgentAssetStatus.REVIEW.value
        db.commit()

        catalog = ExpenseRuleRuntimeService(db).load_catalog()

        assert catalog.travel_policy is not None
        assert catalog.travel_policy.standard_rule_code == COMPANY_TRAVEL_EXPENSE_RULE_CODE
        assert catalog.travel_policy.standard_rule_name == "差旅住宿报销标准"
        assert catalog.travel_policy.hotel_city_limits["北京"]["P0"] == 450
        assert catalog.travel_policy.hotel_city_limits["北京"]["P4"] == 450
        assert catalog.travel_policy.hotel_city_limits["北京"]["P8"] == 500
        assert catalog.travel_policy.allowance_limits["meal"]["直辖市/特区"] == 65
        assert catalog.travel_policy.allowance_limits["meal"]["其他地区"] == 55
        assert catalog.travel_policy.allowance_limits["total"]["其他地区"] == 90
        assert catalog.travel_policy.transport_limits["P7"]["flight"] == 1
        assert catalog.travel_policy.transport_limits["P8"]["train"] == 2


def test_travel_reimbursement_calculator_uses_finance_spreadsheet_amounts() -> None:
    with build_session() as db:
        db.add(
            Employee(
                employee_no="E9001",
                name="测试员工",
                email="traveler@example.com",
                position="产品经理",
                grade="P4",
            )
        )
        db.commit()

        result = TravelReimbursementCalculatorService(db).calculate(
            TravelReimbursementCalculatorRequest(days=3, location="北京市朝阳区"),
            CurrentUserContext(
                username="traveler@example.com",
                name="测试员工",
                role_codes=[],
                is_admin=False,
            ),
        )

        assert result.rule_name == "差旅住宿报销标准"
        assert result.grade == "P4"
        assert result.grade_band == "P4"
        assert result.matched_city == "北京"
        assert result.hotel_rate == 450
        assert result.hotel_amount == 1350
        assert result.allowance_region == "直辖市/特区"
        assert result.total_allowance_rate == 100
        assert result.allowance_amount == 300
        assert result.transport_estimated_amount == 1040
        assert result.transport_estimate_source == "basic_rule_transport_estimate"
        assert result.total_amount == 2690
        assert (
            "交通 1040.00 + 住宿 450.00 × 3 天 + 补贴 100.00 × 3 天 = 2690.00"
            == result.formula_text
        )
        assert "申请预算占用参考总金额为 2690.00 元" in result.summary_text


def test_travel_reimbursement_calculator_uses_other_region_for_known_unlisted_location() -> None:
    with build_session() as db:
        db.add(
            Employee(
                employee_no="E9002",
                name="其他地区员工",
                email="other-region@example.com",
                position="产品经理",
                grade="P4",
            )
        )
        db.commit()

        result = TravelReimbursementCalculatorService(db).calculate(
            TravelReimbursementCalculatorRequest(days=2, location="吉林延边"),
            CurrentUserContext(
                username="other-region@example.com",
                name="其他地区员工",
                role_codes=[],
                is_admin=False,
            ),
        )

        assert result.matched_city == "延边（其他地区）"
        assert result.city_tier == "tier_3"
        assert result.hotel_rate == 380
        assert result.hotel_amount == 760
        assert result.allowance_region == "其他地区"
        assert result.total_allowance_rate == 90
        assert result.allowance_amount == 180
        assert result.transport_estimated_amount == 720
        assert result.total_amount == 1660


def test_travel_reimbursement_calculator_rejects_unrecognized_location() -> None:
    with build_session() as db:
        db.add(
            Employee(
                employee_no="E9003",
                name="无效地点员工",
                email="invalid-location@example.com",
                position="产品经理",
                grade="P4",
            )
        )
        db.commit()

        with pytest.raises(ValueError, match="未识别为有效出差地区"):
            TravelReimbursementCalculatorService(db).calculate(
                TravelReimbursementCalculatorRequest(days=2, location="背景"),
                CurrentUserContext(
                    username="invalid-location@example.com",
                    name="无效地点员工",
                    role_codes=[],
                    is_admin=False,
                ),
            )


def test_travel_reimbursement_calculator_normalizes_location_mixed_with_business_content() -> None:
    with build_session() as db:
        db.add(
            Employee(
                employee_no="E9004",
                name="混合地点员工",
                email="mixed-location@example.com",
                position="产品经理",
                grade="P4",
            )
        )
        db.commit()

        result = TravelReimbursementCalculatorService(db).calculate(
            TravelReimbursementCalculatorRequest(days=4, location="上海辅助国网仿生产服务器"),
            CurrentUserContext(
                username="mixed-location@example.com",
                name="混合地点员工",
                role_codes=[],
                is_admin=False,
            ),
        )

        assert result.location == "上海市"
        assert result.matched_city == "上海"
        assert result.hotel_amount > 0


def test_agent_run_service_lists_seeded_trace_data() -> None:
    with build_session() as db:
        service = AgentRunService(db)

        runs = service.list_runs()

        assert len(runs) >= 3
        assert any(item.tool_calls for item in runs)
        assert any(item.semantic_parse is not None for item in runs)


def test_agent_run_service_creates_run_and_persists_error_message() -> None:
    with build_session() as db:
        service = AgentRunService(db)

        created = service.create_run(
            agent=AgentName.ORCHESTRATOR.value,
            source=AgentRunSource.SYSTEM_EVENT.value,
            status=AgentRunStatus.FAILED.value,
            error_message="simulated failure",
            result_summary="failed to route request",
        )
        fetched = service.get_run(created.run_id)

        assert fetched is not None
        assert fetched.run_id.startswith("run_")
        assert fetched.status == AgentRunStatus.FAILED.value
        assert fetched.error_message == "simulated failure"
        assert fetched.result_summary == "failed to route request"


def test_agent_asset_creation_writes_audit_log() -> None:
    with build_session() as db:
        service = AgentAssetService(db)

        created = service.create_asset(
            AgentAssetCreate(
                asset_type=AgentAssetType.SKILL,
                code=f"skill.test.{uuid.uuid4().hex[:8]}",
                name="测试技能",
                description="用于测试审计日志写入。",
                domain=AgentAssetDomain.KNOWLEDGE,
                scenario_json=["knowledge", "query"],
                owner="pytest",
                reviewer="reviewer",
                status=AgentAssetStatus.DRAFT,
                config_json={"enabled": True},
            ),
            actor="pytest",
        )
        logs = AuditLogService(db).list_logs(resource_id=created.id)

        assert any(item.action == "create_agent_asset" for item in logs)