agent-lab/tests/test_event_kernel.py

from __future__ import annotations

import gc
import json
import warnings
from typing import Any

import pytest

from agent_lab.application.events import (
    ConfirmationPolicy,
    EventDefinition,
    EventExecutionContext,
    EventKernel,
    EventRegistry,
    EventRequest,
    ResolvedEventArguments,
    EventSource,
    EventStatus,
    ResultPolicy,
    RiskLevel,
)
from agent_lab.application.tools import (
    ToolDefinition,
    ToolExecutionContext,
    ToolRegistry,
)
from agent_lab.domain.events import ToolCallEvent
from agent_lab.domain.messages import ChatMessage


def _definition(
    name: str = "example.lookup",
    **overrides: Any,
) -> EventDefinition:
    values: dict[str, Any] = {
        "name": name,
        "description": "Look up an example value.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
        "handler": lambda request: {
            "event": request.name,
            "query": request.arguments["query"],
        },
    }
    values.update(overrides)
    return EventDefinition(**values)


def test_registry_registers_flat_definitions_and_filters_enabled_catalog():
    registry = EventRegistry(
        [_definition("example.lookup"), _definition("device.inspect")]
    )

    assert [item["name"] for item in registry.catalog()] == [
        "example.lookup",
        "device.inspect",
    ]
    assert registry.catalog(["device.inspect"]) == [
        {
            "name": "device.inspect",
            "description": "Look up an example value.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        }
    ]
    assert registry.tool_schema("example.lookup")["function"]["name"] == (
        "example.lookup"
    )
    assert registry.tool_schema("missing") is None


def test_registry_rejects_duplicate_definition_names():
    with pytest.raises(ValueError, match="duplicate event definition: example.lookup"):
        EventRegistry([_definition(), _definition()])


def test_registry_rejects_invalid_draft_2020_12_schema():
    definition = _definition(parameters={"type": 42})

    with pytest.raises(ValueError, match="invalid event schema for example.lookup"):
        EventRegistry([definition])


def test_tool_registry_public_api_remains_compatible():
    registry = ToolRegistry(
        [
            ToolDefinition(
                name="compat.lookup",
                description="Look up compatibility data.",
                parameters={
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
                handler=lambda event: {
                    "tool": event.name,
                    "query": event.arguments["query"],
                },
                argument_resolver=lambda event, context: {
                    "query": context.history[-1].content
                },
            )
        ]
    )
    event = ToolCallEvent(
        id="call-1",
        name="compat.lookup",
        arguments={"query": "provider"},
        raw_arguments='{"query":"provider"}',
    )

    assert registry.available_tools() == [
        {
            "name": "compat.lookup",
            "description": "Look up compatibility data.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        }
    ]
    assert "- compat.lookup: Look up compatibility data." in (
        registry.chat_event_system_message(["compat.lookup"])
    )
    assert registry.tool_schema("compat.lookup")["function"]["name"] == (
        "compat.lookup"
    )
    assert registry.handle(
        event,
        ToolExecutionContext(history=[ChatMessage(role="user", content="history")]),
    ) == {"tool": "compat.lookup", "query": "history"}
    assert registry.execute(event) == {"tool": "compat.lookup", "query": "provider"}


@pytest.mark.asyncio
async def test_kernel_executes_complete_deterministic_arguments_without_fallback():
    fallback_calls: list[str] = []

    async def fallback(*args: Any) -> dict[str, Any]:
        fallback_calls.append("called")
        return {"query": "fallback"}

    registry = EventRegistry(
        [_definition(resolver=lambda request, context: {"query": "deterministic"})]
    )

    result = await EventKernel(registry, argument_fallback=fallback).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.SUCCESS
    assert result.arguments == {"query": "deterministic"}
    assert result.payload == {"event": "example.lookup", "query": "deterministic"}
    assert result.used_fallback is False
    assert fallback_calls == []


@pytest.mark.asyncio
async def test_kernel_calls_fallback_once_when_required_arguments_are_incomplete():
    fallback_calls: list[dict[str, Any]] = []

    async def fallback(
        definition: EventDefinition,
        request: EventRequest,
        context: EventExecutionContext,
    ) -> ResolvedEventArguments:
        fallback_calls.append(dict(request.arguments))
        return ResolvedEventArguments(
            event_name=definition.name,
            arguments={"query": "resolved once"},
            raw_arguments='{"query":"resolved once"}',
        )

    registry = EventRegistry([_definition(resolver=lambda request, context: {})])

    result = await EventKernel(registry, argument_fallback=fallback).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.SUCCESS
    assert result.arguments == {"query": "resolved once"}
    assert result.used_fallback is True
    assert fallback_calls == [{}]


@pytest.mark.asyncio
async def test_kernel_does_not_fallback_when_definition_disallows_it():
    fallback_calls = 0

    async def fallback(*args: Any) -> ResolvedEventArguments:
        nonlocal fallback_calls
        fallback_calls += 1
        return ResolvedEventArguments(
            event_name="example.lookup",
            arguments={"query": "not allowed"},
            raw_arguments='{"query":"not allowed"}',
        )

    registry = EventRegistry(
        [
            _definition(
                resolver=lambda request, context: {},
                fallback_allowed=False,
            )
        ]
    )

    result = await EventKernel(registry, argument_fallback=fallback).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.error == "missing required arguments: query"
    assert result.used_fallback is False
    assert fallback_calls == 0


@pytest.mark.asyncio
async def test_provider_resolved_arguments_are_not_rewritten_or_fallen_back():
    resolver_calls = 0
    fallback_calls = 0

    def resolver(*args: Any) -> dict[str, Any]:
        nonlocal resolver_calls
        resolver_calls += 1
        return {"query": "rewritten"}

    async def fallback(*args: Any) -> ResolvedEventArguments:
        nonlocal fallback_calls
        fallback_calls += 1
        return ResolvedEventArguments(
            event_name="example.lookup",
            arguments={"query": "fallback"},
            raw_arguments='{"query":"fallback"}',
        )

    registry = EventRegistry([_definition(resolver=resolver)])

    result = await EventKernel(registry, argument_fallback=fallback).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": "provider value"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.SUCCESS
    assert result.arguments == {"query": "provider value"}
    assert resolver_calls == 0
    assert fallback_calls == 0


@pytest.mark.asyncio
async def test_provider_resolved_missing_arguments_return_invalid_without_fallback():
    fallback_calls = 0

    async def fallback(*args: Any) -> ResolvedEventArguments:
        nonlocal fallback_calls
        fallback_calls += 1
        return ResolvedEventArguments(
            event_name="example.lookup",
            arguments={"query": "fallback"},
            raw_arguments='{"query":"fallback"}',
        )

    result = await EventKernel(
        EventRegistry([_definition()]), argument_fallback=fallback
    ).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.error == "missing required arguments: query"
    assert fallback_calls == 0


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("event_request", "enabled_names", "expected_status", "expected_error"),
    [
        (
            EventRequest(id="event-1", name="missing"),
            ["missing"],
            EventStatus.UNKNOWN,
            "unknown event",
        ),
        (
            EventRequest(id="event-1", name="example.lookup"),
            [],
            EventStatus.DISABLED,
            "event disabled",
        ),
        (
            EventRequest(
                id="event-1",
                name="example.lookup",
                arguments={"query": 42},
                source=EventSource.PROVIDER_RESOLVED,
            ),
            ["example.lookup"],
            EventStatus.INVALID_ARGUMENTS,
            "invalid argument type for query: expected string",
        ),
    ],
)
async def test_kernel_normalizes_lookup_and_validation_failures(
    event_request: EventRequest,
    enabled_names: list[str],
    expected_status: EventStatus,
    expected_error: str,
):
    result = await EventKernel(EventRegistry([_definition()])).execute(
        event_request,
        enabled_names=enabled_names,
    )

    assert result.status is expected_status
    assert result.error == expected_error


@pytest.mark.asyncio
async def test_kernel_normalizes_handler_exceptions():
    def fail(request: EventRequest) -> dict[str, Any]:
        raise RuntimeError("boom")

    result = await EventKernel(
        EventRegistry([_definition(handler=fail)])
    ).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": "value"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.HANDLER_ERROR
    assert result.error == "event handler failed: boom"


@pytest.mark.asyncio
async def test_kernel_validates_complete_draft_2020_12_schema():
    definition = _definition(
        parameters={
            "type": "object",
            "properties": {
                "mode": {"enum": ["quick", "deep"]},
                "target": {"type": ["string", "null"]},
                "filters": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {"score": {"type": "number", "minimum": 0}},
                        "required": ["score"],
                        "additionalProperties": False,
                    },
                },
            },
            "required": ["mode", "target", "filters"],
            "additionalProperties": False,
        },
        handler=lambda request: {"event": request.name},
    )
    kernel = EventKernel(EventRegistry([definition]))

    valid = await kernel.execute(
        EventRequest(
            id="valid",
            name=definition.name,
            arguments={
                "mode": "deep",
                "target": None,
                "filters": [{"score": 0.5}],
            },
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=[definition.name],
    )
    invalid = await kernel.execute(
        EventRequest(
            id="invalid",
            name=definition.name,
            arguments={
                "mode": "other",
                "target": 7,
                "filters": [{"score": -1, "extra": True}],
                "unexpected": True,
            },
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=[definition.name],
    )

    assert valid.status is EventStatus.SUCCESS
    assert invalid.status is EventStatus.INVALID_ARGUMENTS
    assert invalid.error.startswith("invalid event arguments:")


@pytest.mark.asyncio
async def test_kernel_normalizes_validator_runtime_exceptions():
    definition = _definition(parameters={"$ref": "urn:agent-lab:missing-schema"})

    result = await EventKernel(EventRegistry([definition])).execute(
        EventRequest(
            id="event-1",
            name=definition.name,
            arguments={},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=[definition.name],
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.error.startswith("event argument validation failed:")


@pytest.mark.asyncio
@pytest.mark.parametrize("boundary", ["resolver", "fallback"])
async def test_kernel_normalizes_resolution_boundary_exceptions(boundary: str):
    def resolver(request: EventRequest, context: EventExecutionContext) -> dict[str, Any]:
        if boundary == "resolver":
            raise RuntimeError("resolver boom")
        return {}

    async def fallback(*args: Any) -> ResolvedEventArguments:
        raise RuntimeError("fallback boom")

    result = await EventKernel(
        EventRegistry([_definition(resolver=resolver)]),
        argument_fallback=fallback,
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.RESOLUTION_ERROR
    assert result.error == f"event argument {boundary} failed: {boundary} boom"


@pytest.mark.asyncio
@pytest.mark.parametrize("boundary", ["resolver", "fallback"])
async def test_kernel_normalizes_invalid_resolution_payloads(boundary: str):
    resolver = (
        (lambda request, context: None)
        if boundary == "resolver"
        else (lambda request, context: {})
    )

    async def fallback(*args: Any) -> Any:
        return {"query": "legacy bare mapping"}

    result = await EventKernel(
        EventRegistry([_definition(resolver=resolver)]),
        argument_fallback=fallback,
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.RESOLUTION_ERROR
    assert result.error == f"event argument {boundary} returned invalid payload"


@pytest.mark.asyncio
async def test_kernel_normalizes_non_json_resolver_arguments():
    result = await EventKernel(
        EventRegistry(
            [
                _definition(
                    resolver=lambda request, context: {"query": object()}
                )
            ]
        )
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.RESOLUTION_ERROR
    assert result.error.startswith("event argument resolver failed to serialize:")


@pytest.mark.asyncio
async def test_kernel_normalizes_non_object_fallback_arguments():
    async def fallback(*args: Any) -> ResolvedEventArguments:
        return ResolvedEventArguments(
            event_name="example.lookup",
            arguments=["not", "an", "object"],  # type: ignore[arg-type]
            raw_arguments='["not","an","object"]',
        )

    result = await EventKernel(
        EventRegistry([_definition(resolver=lambda request, context: {})]),
        argument_fallback=fallback,
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.RESOLUTION_ERROR
    assert result.error == "event argument fallback returned invalid payload"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("resolved", "expected_error"),
    [
        (
            ResolvedEventArguments(
                event_name="another.event",
                arguments={"query": "value"},
                raw_arguments='{"query":"value"}',
            ),
            "fallback returned tool another.event for example.lookup",
        ),
        (
            ResolvedEventArguments(
                event_name="example.lookup",
                arguments={"query": "value"},
                raw_arguments="not-json",
            ),
            "fallback raw arguments are not valid JSON",
        ),
        (
            ResolvedEventArguments(
                event_name="example.lookup",
                arguments={"query": "parsed"},
                raw_arguments='{"query":"raw"}',
            ),
            "fallback raw arguments do not match parsed arguments",
        ),
    ],
)
async def test_kernel_rejects_inconsistent_structured_fallback(
    resolved: ResolvedEventArguments,
    expected_error: str,
):
    async def fallback(*args: Any) -> ResolvedEventArguments:
        return resolved

    result = await EventKernel(
        EventRegistry([_definition(resolver=lambda request, context: {})]),
        argument_fallback=fallback,
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.RESOLUTION_ERROR
    assert result.error == expected_error


@pytest.mark.asyncio
async def test_kernel_passes_consistent_fallback_arguments_and_raw_json_to_handler():
    captured: list[EventRequest] = []

    async def fallback(*args: Any) -> ResolvedEventArguments:
        return ResolvedEventArguments(
            event_name="example.lookup",
            arguments={"query": "resolved"},
            raw_arguments='{"query":"resolved"}',
        )

    result = await EventKernel(
        EventRegistry(
            [
                _definition(
                    resolver=lambda request, context: {},
                    handler=lambda request: captured.append(request) or {"ok": True},
                )
            ]
        ),
        argument_fallback=fallback,
    ).execute(
        EventRequest(id="event-1", name="example.lookup"),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.SUCCESS
    assert captured[0].arguments == json.loads(captured[0].raw_arguments)
    assert result.raw_arguments == captured[0].raw_arguments


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [None, "text", 1, ["item"]])
async def test_kernel_normalizes_invalid_handler_payloads(payload: Any):
    result = await EventKernel(
        EventRegistry([_definition(handler=lambda request: payload)])
    ).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": "value"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.HANDLER_ERROR
    assert result.error == "event handler returned non-object payload"


@pytest.mark.asyncio
async def test_async_kernel_supports_async_handler():
    async def handler(request: EventRequest) -> dict[str, Any]:
        return {"query": request.arguments["query"]}

    result = await EventKernel(
        EventRegistry([_definition(handler=handler)])
    ).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": "async"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.SUCCESS
    assert result.payload == {"query": "async"}


@pytest.mark.asyncio
async def test_tool_registry_async_entry_points_support_async_handler():
    async def handler(event: ToolCallEvent) -> dict[str, Any]:
        return {"tool": event.name, "query": event.arguments["query"]}

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="async.lookup",
                description="Async lookup.",
                parameters={
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
                handler=handler,
            )
        ]
    )
    event = ToolCallEvent(
        id="call-1",
        name="async.lookup",
        arguments={"query": "value"},
        raw_arguments='{"query":"value"}',
    )

    assert await registry.handle_async(event) == {
        "tool": "async.lookup",
        "query": "value",
    }
    assert await registry.execute_async(event) == {
        "tool": "async.lookup",
        "query": "value",
    }


def test_tool_registry_sync_facade_rejects_async_handler_without_runtime_warning():
    called = False

    async def handler(event: ToolCallEvent) -> dict[str, Any]:
        nonlocal called
        called = True
        return {"tool": event.name}

    registry = ToolRegistry(
        [
            ToolDefinition(
                name="async.lookup",
                description="Async lookup.",
                parameters={"type": "object"},
                handler=handler,
            )
        ]
    )
    event = ToolCallEvent(
        id="call-1",
        name="async.lookup",
        arguments={},
        raw_arguments="{}",
    )

    with warnings.catch_warnings(record=True) as captured:
        warnings.simplefilter("always")
        payload = registry.execute(event)
        gc.collect()

    assert payload == {
        "tool": "async.lookup",
        "error": "tool handler failed: async event handlers require execute_async",
    }
    assert called is False
    assert not [warning for warning in captured if issubclass(warning.category, RuntimeWarning)]


@pytest.mark.asyncio
async def test_kernel_rejects_boolean_for_json_number_arguments():
    definition = _definition(
        parameters={
            "type": "object",
            "properties": {"query": {"type": "number"}},
            "required": ["query"],
        }
    )

    result = await EventKernel(EventRegistry([definition])).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": True},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.error == "invalid argument type for query: expected number"


@pytest.mark.asyncio
async def test_definition_metadata_survives_registration_and_result_creation():
    definition = _definition(
        result_policy=ResultPolicy.TEMPLATE_FOLLOW_UP,
        confirmation_policy=ConfirmationPolicy.REQUIRED,
        risk_level=RiskLevel.HIGH,
        idempotency_key_fields=("session_id", "event_id"),
        concurrency_class="device-write",
        conflict_keys=("device",),
        timeout_seconds=1.5,
        terminal=True,
    )
    registry = EventRegistry([definition])

    result = await EventKernel(registry).execute(
        EventRequest(
            id="event-1",
            name="example.lookup",
            arguments={"query": "value"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=["example.lookup"],
    )

    registered = registry.definition("example.lookup")
    assert registered is definition
    assert result.result_policy is ResultPolicy.TEMPLATE_FOLLOW_UP
    assert result.confirmation_policy is ConfirmationPolicy.REQUIRED
    assert result.risk_level is RiskLevel.HIGH
    assert result.idempotency_key_fields == ("session_id", "event_id")
    assert result.concurrency_class == "device-write"
    assert result.conflict_keys == ("device",)
    assert result.timeout_seconds == 1.5
    assert result.terminal is True


@pytest.mark.asyncio
@pytest.mark.parametrize("name", ["alpha.one", "beta-two", "任意.事件"])
async def test_kernel_applies_identical_behavior_to_arbitrary_event_names(name: str):
    definition = _definition(name=name)

    result = await EventKernel(EventRegistry([definition])).execute(
        EventRequest(
            id="event-1",
            name=name,
            arguments={"query": "value"},
            source=EventSource.PROVIDER_RESOLVED,
        ),
        enabled_names=[name],
    )

    assert result.status is EventStatus.SUCCESS
    assert result.payload == {"event": name, "query": "value"}