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() == (
        "enrollment-guidance",
        "insurance-consultation",
        "operation-insights",
    )
    customer = personas.get("customer")
    operation = personas.get("operation")
    assert customer.prompt_id == "customer"
    assert "calculate_insurance_quote" in customer.tool_names
    assert operation.prompt_id == "operation"
    assert "get_attribution_performance" in operation.tool_names
    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_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",
    }