group_fqcd_jr/tests/unit/service/test_intent_agent_entrypoint.py

import pytest

from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
from app.service.agent.base import BaseAgent
from app.service.agent.factory import AgentFactory
from app.service.intent_classifier import IntentClassifier
from app.service.model_gateway import ModelExecution


class DemoAgent(BaseAgent):
    async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
        return CoreResult(text="ok")


class StubModel:
    async def generate(self, _endpoints: list[object], _prompt: str) -> ModelExecution:
        return ModelExecution("intent", '{"intent":"profile","confidence":0.9}', 1)


class StubResolver:
    async def resolve(self, *, agent_type: str, task_type: str) -> list[object]:
        assert agent_type == "demo"
        assert task_type == "intent_classification"
        return [object()]


@pytest.mark.asyncio
async def test_execute_classifies_before_handle_and_attaches_result(governance) -> None:
    definition = AgentDefinition(
        agent_type="demo", version="1", allowed_roles=("customer",),
        allowed_portals=("api",), supported_intents=("profile", "general")
    )
    factory = AgentFactory(
        governance=governance,
        intent_classifier=IntentClassifier(StubModel()),
        intent_endpoint_resolver=StubResolver(),
    )
    factory.register(definition, lambda _context: DemoAgent(definition))
    context = RequestContext(
        user_id="1", trace_id="intent", roles=("customer",), permissions=("agent:run",)
    )
    request = AgentRequest(
        agent_type="demo", message="我的资料", session_id="s",
        idempotency_key="intent-request-0001"
    )
    events = [event async for event in factory.create("demo", context).execute(
        request, context, "run"
    )]
    result = events[-1].payload["result"]["result"]
    assert result["intent"]["intent"] == "profile"
    assert result["intent"]["confidence"] == 0.9


@pytest.mark.asyncio
async def test_fixed_route_agent_skips_model_intent_classification(governance) -> None:
    """固定路由 Agent 不能因模型意图端点不可用而阻断服务。"""
    definition = AgentDefinition(
        agent_type="demo", version="1", allowed_roles=("customer",),
        allowed_portals=("api",), requires_model_intent_classification=False,
    )
    factory = AgentFactory(
        governance=governance,
        intent_classifier=IntentClassifier(StubModel()),
        intent_endpoint_resolver=StubResolver(),
    )
    factory.register(definition, lambda _context: DemoAgent(definition))
    context = RequestContext(
        user_id="1", trace_id="fixed-route", roles=("customer",), permissions=("agent:run",)
    )
    request = AgentRequest(
        agent_type="demo", message="本地路由", session_id="s",
        idempotency_key="fixed-route-request-0001",
    )

    events = [
        event async for event in factory.create("demo", context).execute(request, context, "run")
    ]

    assert events[-1].payload["result"]["result"]["intent"] is None


def test_business_agent_cannot_override_intent_governance() -> None:
    with pytest.raises(TypeError, match="classify_intent"):
        class InvalidAgent(BaseAgent):
            async def classify_intent(self, request: AgentRequest):
                return None

            async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
                return CoreResult(text="bad")