harness_zbt/backend/tests/test_harness_core.py

import json
from datetime import UTC, datetime

import pytest

from app.commands.seed import build_seed_manifest
from app.core.errors import AppError
from app.domains.agent.tools import build_agent_tool_registry
from app.domains.catalog.repository import InMemoryCatalogRepository
from app.domains.catalog.service import ProductCatalogService
from app.domains.enrollment.repository import InMemoryEnrollmentRepository
from app.domains.enrollment.service import EnrollmentService
from app.domains.identity.models import AdminUser, H5User
from app.harness.policy import HarnessPolicyEngine, ToolPolicy
from app.harness.registries import PersonaRegistry, PromptRegistry, SkillRegistry
from app.harness.schemas import ProductRecommendationsBlock
from app.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",
    }