harness_zbt/backend/tests/test_harness_core.py

import json
from datetime import UTC, datetime

import pytest

from zbt.commands.seed import build_seed_manifest
from zbt.core.errors import AppError
from zbt.domains.agent.tools import build_agent_tool_registry
from zbt.domains.catalog.repository import InMemoryCatalogRepository
from zbt.domains.catalog.service import ProductCatalogService
from zbt.domains.enrollment.repository import InMemoryEnrollmentRepository
from zbt.domains.enrollment.service import EnrollmentService
from zbt.domains.identity.models import AdminUser, H5User
from zbt.harness.policy import HarnessPolicyEngine, ToolPolicy
from zbt.harness.registries import PersonaRegistry, PromptRegistry, SkillRegistry
from zbt.harness.schemas import ProductRecommendationsBlock
from zbt.harness.tooling import ToolExecutionContext

NOW = datetime(2026, 7, 26, 8, tzinfo=UTC)


def test_prompt_skill_and_persona_registries_form_a_closed_contract() -> None:
    prompts = PromptRegistry()
    skills = SkillRegistry()
    personas = PersonaRegistry()

    assert prompts.ids() == ("customer", "operation")
    assert skills.ids() == (
        "customer-memory",
        "enrollment-guidance",
        "insurance-consultation",
        "insurance-knowledge",
        "operation-insights",
    )
    customer = personas.get("customer")
    operation = personas.get("operation")
    assert customer.prompt_id == "customer"
    assert "calculate_insurance_quote" in customer.tool_names
    assert "search_insurance_knowledge" in customer.tool_names
    assert operation.prompt_id == "operation"
    assert "get_attribution_performance" in operation.tool_names
    assert "search_insurance_knowledge" in operation.tool_names
    customer_prompt = prompts.get("customer")
    knowledge_skill = skills.get("insurance-knowledge")
    assert "不得把宽泛条款扩展到具体场景" in customer_prompt
    assert "不得给出或估算免赔额、报销比例" in customer_prompt
    assert "宽泛条款不能自动套用到具体情形" in knowledge_skill.instructions
    for skill_id in customer.skill_ids + operation.skill_ids:
        skill = skills.get(skill_id)
        assert skill.instructions
        assert skill.source.name == "SKILL.md"


def test_operation_persona_cannot_authorize_a_draft_tool() -> None:
    policy = HarnessPolicyEngine()

    with pytest.raises(AppError) as error:
        policy.authorize(
            persona="operation",
            allowed_by_persona=("calculate_insurance_quote",),
            tool_name="calculate_insurance_quote",
            policy=ToolPolicy(personas=("operation",), effect="draft"),
            permissions=("*",),
        )

    assert error.value.code == "AGENT_OPERATION_READ_ONLY"


def test_customer_tools_use_real_catalog_and_deterministic_quote_service() -> None:
    catalog_repository = InMemoryCatalogRepository()
    manifest = build_seed_manifest()
    for product in manifest.products:
        catalog_repository.save_product(product)
    for version in manifest.versions:
        catalog_repository.save_version(version)
    catalog = ProductCatalogService(catalog_repository, lambda: NOW)
    enrollment = EnrollmentService(
        InMemoryEnrollmentRepository(),
        catalog,
        lambda: NOW,
        "x" * 32,
    )
    registry = build_agent_tool_registry(catalog, enrollment)
    user = H5User(
        id="01H5USER000000000000000001",
        mobile="18800000001",
        mobile_masked="188****0001",
        display_name="体验用户",
        status="ACTIVE",
        created_at=NOW,
    )
    context = ToolExecutionContext(persona="customer", h5_user=user)
    tools = registry.build(
        names=("list_available_products", "calculate_insurance_quote"),
        context=context,
        policy_engine=HarnessPolicyEngine(),
    )
    by_name = {tool.name: tool for tool in tools}

    products = json.loads(by_name["list_available_products"].invoke({"category": "MEDICAL"}))
    quote = json.loads(
        by_name["calculate_insurance_quote"].invoke(
            {
                "product_code": "MED-SENIOR",
                "plan_code": "SENIOR_STANDARD",
                "age": 65,
                "region_code": "510100",
                "occupation_code": "GENERAL",
                "relationship": "PARENT",
            }
        )
    )

    assert products["data"]["total"] == 3
    assert quote["data"]["premium_cents"] == 19900
    assert quote["blocks"][0]["type"] == "quote"
    assert quote["actions"][0]["type"] == "open_enrollment"
    assert [execution.name for execution in context.executions] == [
        "list_available_products",
        "calculate_insurance_quote",
    ]


def test_product_recommendation_contract_accepts_enriched_plan_data() -> None:
    catalog_repository = InMemoryCatalogRepository()
    manifest = build_seed_manifest()
    for product in manifest.products:
        catalog_repository.save_product(product)
    for version in manifest.versions:
        catalog_repository.save_version(version)
    catalog = ProductCatalogService(catalog_repository, lambda: NOW)
    enrollment = EnrollmentService(
        InMemoryEnrollmentRepository(),
        catalog,
        lambda: NOW,
        "x" * 32,
    )
    registry = build_agent_tool_registry(catalog, enrollment)
    context = ToolExecutionContext(
        persona="customer",
        h5_user=H5User(
            id="01H5USER000000000000000001",
            mobile="18800000001",
            mobile_masked="188****0001",
            display_name="体验用户",
            status="ACTIVE",
            created_at=NOW,
        ),
    )
    tool = registry.build(
        names=("list_available_products",),
        context=context,
        policy_engine=HarnessPolicyEngine(),
    )[0]

    products = json.loads(tool.invoke({"category": "MEDICAL"}))
    product = next(
        item
        for item in products["data"]["items"]
        if item["product_code"] == "MED-SENIOR"
    )
    recommendation = ProductRecommendationsBlock(
        items=[
            {
                "product_id": product["product_id"],
                "product_code": product["product_code"],
                "name": product["name"],
                "category": product["category"],
                "summary": product["summary"],
                "product_version_id": product["product_version_id"],
                "plans": product["plans"],
            }
        ],
    )

    plan = recommendation.items[0].plans[0]
    assert plan.premium_cents == 19900
    assert plan.coverage_amount_cents == 100_000_000
    assert (plan.min_age, plan.max_age) == (50, 75)


def test_operation_analytics_tool_uses_allowlisted_query_and_chart_contract() -> None:
    class AnalyticsStub:
        def query(
            self,
            *,
            metric: str,
            group_by: str,
            days: int,
            admin_user_id: str | None = None,
        ) -> dict[str, object]:
            assert metric == "ORDER_COUNT"
            assert group_by == "DAY"
            assert days == 30
            assert admin_user_id is None
            return {
                "metric": metric,
                "metric_name": "订单量",
                "unit": "笔",
                "group_by": group_by,
                "rows": [
                    {"label": "2026-07-26", "value": 12},
                    {"label": "2026-07-27", "value": 18},
                ],
                "total": 30,
                "query_plan": {
                    "source": "analytics",
                    "fact": "fact_orders",
                    "aggregation": "COUNT",
                    "dimension": "dim_date",
                    "read_only": True,
                },
            }

    catalog = ProductCatalogService(InMemoryCatalogRepository(), lambda: NOW)
    enrollment = EnrollmentService(
        InMemoryEnrollmentRepository(),
        catalog,
        lambda: NOW,
        "x" * 32,
    )
    registry = build_agent_tool_registry(
        catalog,
        enrollment,
        analytics=AnalyticsStub(),
    )
    context = ToolExecutionContext(
        persona="operation",
        admin_user=AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash="unused",
            display_name="管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("orders:read", "policies:read"),
            data_scope="ALL",
        ),
    )
    tool = registry.build(
        names=("query_operation_analytics",),
        context=context,
        policy_engine=HarnessPolicyEngine(),
    )[0]

    result = json.loads(
        tool.invoke(
            {
                "metric": "ORDER_COUNT",
                "group_by": "DAY",
                "days": 30,
            }
        )
    )

    assert result["data"]["query_plan"]["read_only"] is True
    assert result["data"]["total"] == 30
    assert result["blocks"][0]["type"] == "chart"
    assert result["blocks"][0]["chart_type"] == "line"
    assert result["blocks"][0]["points"][1] == {
        "label": "2026-07-27",
        "value": 18,
    }


def test_operation_tool_permissions_filter_unavailable_tools() -> None:
    catalog_repository = InMemoryCatalogRepository()
    catalog = ProductCatalogService(catalog_repository, lambda: NOW)
    enrollment = EnrollmentService(
        InMemoryEnrollmentRepository(),
        catalog,
        lambda: NOW,
        "x" * 32,
    )
    registry = build_agent_tool_registry(catalog, enrollment)
    admin = AdminUser(
        id="01ADMIN0000000000000000001",
        username="sales_a",
        password_hash="unused",
        display_name="业务员A",
        status="ACTIVE",
        roles=("SALESPERSON",),
        permissions=("dashboard:read", "attribution:read"),
        data_scope="SELF",
    )

    tools = registry.build(
        names=PersonaRegistry().get("operation").tool_names,
        context=ToolExecutionContext(persona="operation", admin_user=admin),
        policy_engine=HarnessPolicyEngine(),
    )

    assert {tool.name for tool in tools} == {
        "get_operation_overview",
        "get_attribution_performance",
        "list_available_products",
        "search_insurance_knowledge",
    }