X-Financial/server/tests/test_digital_employee_skill_catalog.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from app.core.agent_enums import AgentName
from app.services.agent_foundation_constants import (
    DIGITAL_EMPLOYEE_FINANCE_DASHBOARD_SNAPSHOT_TASK_CODE,
    DIGITAL_EMPLOYEE_FINANCE_REPORT_TASK_CODE,
    DIGITAL_EMPLOYEE_FINANCE_POLICY_TASK_CODE,
    DIGITAL_EMPLOYEE_PROFILE_SCAN_TASK_CODE,
    DIGITAL_EMPLOYEE_REMINDER_SCAN_TASK_CODE,
    DIGITAL_EMPLOYEE_SKILL_CATEGORIES,
    DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP,
)
from app.services.agent_foundation_digital_employee_tasks import (
    DIGITAL_EMPLOYEE_ANALYSIS_ROLE_BOUNDARY,
    AgentFoundationDigitalEmployeeTaskMixin,
)


class _CatalogHarness(AgentFoundationDigitalEmployeeTaskMixin):
    def _digital_employee_task_config(self, code: str, cron: str) -> dict[str, Any]:
        return {
            "cron": cron,
            "agent": AgentName.HERMES.value,
            "task_type": code.replace("task.hermes.", "").replace(".", "_"),
            "skill_category": DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP.get(code, "整理"),
            "skill_category_options": list(DIGITAL_EMPLOYEE_SKILL_CATEGORIES),
        }

    def _read_domain_skill_markdown(
        self,
        skill_name: str,
        fallback_lines: list[str],
    ) -> str:
        skill_path = _skill_root() / skill_name / "SKILL.md"
        if skill_path.exists():
            return skill_path.read_text(encoding="utf-8")
        return "\n".join(fallback_lines)

    def _financial_risk_graph_scan_skill_markdown(self) -> str:
        return self._read_domain_skill_markdown("financial-risk-graph-scanner", [])

    def _employee_behavior_profile_scan_skill_markdown(self) -> str:
        return self._read_domain_skill_markdown("employee-behavior-profile-scanner", [])

    def _risk_rule_discovery_skill_markdown(self) -> str:
        return self._read_domain_skill_markdown("risk-rule-discovery", [])

    def _risk_clue_collector_skill_markdown(self) -> str:
        return self._read_domain_skill_markdown("risk-clue-collector", [])


def test_digital_employee_skill_catalog_has_complete_categories_and_packages() -> None:
    harness = _CatalogHarness()
    specs = harness._runtime_digital_employee_task_specs()
    codes = [str(spec["code"]) for spec in specs]
    categories = [str(spec["skill_category"]) for spec in specs]
    skill_names = [str(dict(spec["config"])["skill_name"]) for spec in specs]

    assert len(specs) == 19
    assert len(set(codes)) == len(codes)
    assert set(categories) == set(DIGITAL_EMPLOYEE_SKILL_CATEGORIES)
    assert DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP[DIGITAL_EMPLOYEE_PROFILE_SCAN_TASK_CODE] == "积累"
    assert (
        DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP[DIGITAL_EMPLOYEE_FINANCE_DASHBOARD_SNAPSHOT_TASK_CODE]
        == "整理"
    )
    assert DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP[DIGITAL_EMPLOYEE_REMINDER_SCAN_TASK_CODE] == "升级"
    assert DIGITAL_EMPLOYEE_TASK_CATEGORY_MAP[DIGITAL_EMPLOYEE_FINANCE_REPORT_TASK_CODE] == "整理"
    assert len(set(codes + [DIGITAL_EMPLOYEE_FINANCE_POLICY_TASK_CODE])) == 20

    for skill_name in ["finance-policy-knowledge-organizer", *skill_names]:
        skill_file = _skill_root() / skill_name / "SKILL.md"
        assert skill_file.exists(), skill_name
        content = skill_file.read_text(encoding="utf-8")
        assert f"name: {skill_name}" in content


def test_digital_employee_runtime_specs_build_display_ready_config() -> None:
    harness = _CatalogHarness()
    forbidden_rule_execution = "执行" + "规则"

    for spec in harness._runtime_digital_employee_task_specs():
        config = harness._build_runtime_digital_employee_config(spec)
        spec_config = dict(spec["config"])
        markdown = spec["markdown"]()

        assert config["agent"] == AgentName.HERMES.value
        assert config["skill_category"] == spec["skill_category"]
        assert config["skill_category_options"] == list(DIGITAL_EMPLOYEE_SKILL_CATEGORIES)
        assert config["skill_name"] == spec_config["skill_name"]
        assert config["output_format"] == spec_config["output_format"]
        assert config["schedule"] == spec["cron"]
        assert config["cron_expression"] == spec["cron"]
        assert config["writes_rules"] is False
        assert isinstance(config["role_boundary"], str)
        assert config["role_boundary"] == DIGITAL_EMPLOYEE_ANALYSIS_ROLE_BOUNDARY
        assert "主流程由外层智能体执行" in config["role_boundary"]
        assert forbidden_rule_execution not in config["role_boundary"]
        assert "human_review_required" in config["allowed_outputs"]
        assert f"name: {config['skill_name']}" in markdown
        assert str(spec["name"]) in markdown


def test_digital_employee_skills_do_not_cross_rule_governance_boundary() -> None:
    harness = _CatalogHarness()
    forbidden_rule_execution = "执行" + "规则"
    specs = harness._runtime_digital_employee_task_specs()
    skill_names = {str(dict(spec["config"])["skill_name"]) for spec in specs}
    output_formats = {str(dict(spec["config"])["output_format"]) for spec in specs}
    text_contract = "\n".join(
        str(value)
        for spec in specs
        for value in (
            spec["name"],
            spec["description"],
            dict(spec["config"])["skill_name"],
            dict(spec["config"])["output_format"],
            dict(spec["config"])["role_boundary"],
        )
    )

    assert "risk-clue-collector" in skill_names
    assert "finance-dashboard-snapshot-analyst" in skill_names
    assert "digital-employee-reminder-scanner" in skill_names
    assert "finance-report-orchestrator" in skill_names
    assert "rule-execution-case-organizer" in skill_names
    assert "policy-reference-gap-hinter" in skill_names
    assert "risk-rule-discovery" not in skill_names
    assert "risk-rule-template-organizer" not in skill_names
    assert "policy-gap-rule-optimizer" not in skill_names
    assert "candidate_risk_rules" not in output_formats
    assert "risk_rule_template_library" not in output_formats
    assert "policy_gap_rule_optimization_report" not in output_formats
    assert "auto_publish" not in text_contract
    assert forbidden_rule_execution not in text_contract


def _skill_root() -> Path:
    return Path(__file__).resolve().parents[1] / "src" / "app" / "skills" / "domain"