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.events.models import EventArgumentResolution 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_registry_does_not_expose_mutable_validator_instances(): registry = EventRegistry([_definition()]) assert not hasattr(registry, "validator") errors = list( registry.iter_validation_errors( "example.lookup", {"query": 42}, ) ) assert errors errors[0].schema["type"] = "integer" assert list( registry.iter_validation_errors( "example.lookup", {"query": 42}, ) ) 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 @pytest.mark.parametrize( "arguments", [ {"query": 42}, {"query": "unsupported"}, {"query": "valid", "unexpected": True}, ], ) async def test_kernel_does_not_fallback_for_complete_invalid_arguments( arguments: dict[str, Any], ): fallback_calls = 0 async def fallback(*args: Any) -> ResolvedEventArguments: nonlocal fallback_calls fallback_calls += 1 return ResolvedEventArguments( event_name="example.lookup", arguments={"query": "valid"}, raw_arguments='{"query":"valid"}', ) definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string", "enum": ["valid"]}}, "required": ["query"], "additionalProperties": False, }, resolver=lambda request, context: arguments, ) result = await EventKernel( EventRegistry([definition]), argument_fallback=fallback ).execute( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.INVALID_ARGUMENTS assert result.used_fallback is False assert fallback_calls == 0 @pytest.mark.asyncio @pytest.mark.parametrize("composition", ["anyOf", "oneOf"]) async def test_kernel_falls_back_for_required_missing_inside_composed_schema( composition: str, ): fallback_calls = 0 async def fallback(*args: Any) -> ResolvedEventArguments: nonlocal fallback_calls fallback_calls += 1 return ResolvedEventArguments( event_name="example.lookup", arguments={"choice": {"mode": "alpha"}}, raw_arguments='{"choice":{"mode":"alpha"}}', ) definition = _definition( parameters={ "type": "object", "properties": { "choice": { composition: [ { "type": "object", "properties": {"mode": {"const": "alpha"}}, "required": ["mode"], }, { "type": "object", "properties": {"mode": {"const": "beta"}}, "required": ["mode"], }, ] } }, "required": ["choice"], }, resolver=lambda request, context: {"choice": {}}, handler=lambda request: {"ok": True}, ) result = await EventKernel( EventRegistry([definition]), argument_fallback=fallback ).execute( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.SUCCESS assert result.used_fallback is True assert fallback_calls == 1 @pytest.mark.asyncio @pytest.mark.parametrize("composition", ["anyOf", "oneOf"]) async def test_kernel_does_not_fallback_for_complete_invalid_composed_schema( composition: str, ): fallback_calls = 0 async def fallback(*args: Any) -> ResolvedEventArguments: nonlocal fallback_calls fallback_calls += 1 raise AssertionError("fallback should not run") definition = _definition( parameters={ "type": "object", "properties": { "choice": { composition: [ { "type": "object", "properties": {"mode": {"const": "alpha"}}, "required": ["mode"], }, { "type": "object", "properties": {"mode": {"const": "beta"}}, "required": ["mode"], }, ] } }, "required": ["choice"], }, resolver=lambda request, context: {"choice": {"mode": "other"}}, handler=lambda request: {"ok": True}, ) result = await EventKernel( EventRegistry([definition]), argument_fallback=fallback ).execute( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.INVALID_ARGUMENTS assert result.used_fallback is False assert fallback_calls == 0 @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_structured_resolution_can_mark_optional_arguments_incomplete(): fallback_calls = 0 async def fallback(*args: Any) -> ResolvedEventArguments: nonlocal fallback_calls fallback_calls += 1 return ResolvedEventArguments( event_name="example.lookup", arguments={"query": "resolved"}, raw_arguments='{"query":"resolved"}', ) definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string"}}, "additionalProperties": False, }, resolver=lambda request, context: EventArgumentResolution( arguments={}, complete=False, ), ) result = await EventKernel( EventRegistry([definition]), 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"} assert result.used_fallback is True assert fallback_calls == 1 @pytest.mark.asyncio async def test_structured_incomplete_optional_arguments_fail_without_fallback(): definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string"}}, }, resolver=lambda request, context: EventArgumentResolution( arguments={}, complete=False, ), handler=lambda request: {"ok": True}, ) result = await EventKernel(EventRegistry([definition])).execute( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.INVALID_ARGUMENTS assert result.error == "event arguments incomplete" @pytest.mark.asyncio async def test_plain_dict_resolution_remains_complete_for_optional_schema(): fallback_calls = 0 async def fallback(*args: Any) -> ResolvedEventArguments: nonlocal fallback_calls fallback_calls += 1 raise AssertionError("fallback should not run") definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string"}}, }, resolver=lambda request, context: {}, handler=lambda request: {"ok": True}, ) result = await EventKernel( EventRegistry([definition]), argument_fallback=fallback ).execute( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.SUCCESS assert fallback_calls == 0 def test_sync_kernel_rejects_structured_incomplete_optional_arguments(): handler_calls = 0 def handler(request: EventRequest) -> dict[str, Any]: nonlocal handler_calls handler_calls += 1 return {"ok": True} definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string"}}, }, resolver=lambda request, context: EventArgumentResolution( arguments={}, complete=False, ), handler=handler, ) result = EventKernel(EventRegistry([definition])).execute_sync( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.INVALID_ARGUMENTS assert result.error == "event arguments incomplete" assert handler_calls == 0 @pytest.mark.parametrize( "resolver", [ lambda request, context: EventArgumentResolution( arguments={}, complete=True, ), lambda request, context: {}, ], ) def test_sync_kernel_preserves_complete_compatible_resolvers(resolver: Any): handler_calls = 0 def handler(request: EventRequest) -> dict[str, Any]: nonlocal handler_calls handler_calls += 1 return {"ok": True} definition = _definition( parameters={ "type": "object", "properties": {"query": {"type": "string"}}, }, resolver=resolver, handler=handler, ) result = EventKernel(EventRegistry([definition])).execute_sync( EventRequest(id="event-1", name="example.lookup"), enabled_names=["example.lookup"], ) assert result.status is EventStatus.SUCCESS assert result.payload == {"ok": True} assert handler_calls == 1 @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.DEFINITION_ERROR assert result.error.startswith("event argument validation failed:") @pytest.mark.asyncio async def test_tool_registry_maps_definition_errors_to_tool_compatibility_payload(): registry = ToolRegistry( [ ToolDefinition( name="broken.lookup", description="Broken lookup.", parameters={"$ref": "urn:agent-lab:missing-schema"}, handler=lambda event: {"tool": event.name}, ) ] ) payload = await registry.execute_async( ToolCallEvent( id="call-1", name="broken.lookup", arguments={}, raw_arguments="{}", ) ) assert payload["tool"] == "broken.lookup" assert payload["error"].startswith("tool definition 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 @pytest.mark.parametrize( ("arguments", "raw_arguments"), [ ({"query": 1}, '{"query":true}'), ({"query": 1.0}, '{"query":1}'), ({"query": {"nested": [1]}}, '{"query":{"nested":[true]}}'), ({"query": float("nan")}, '{"query":NaN}'), ({"query": float("inf")}, '{"query":Infinity}'), ], ) async def test_kernel_rejects_noncanonical_fallback_json( arguments: dict[str, Any], raw_arguments: str, ): async def fallback(*args: Any) -> ResolvedEventArguments: return ResolvedEventArguments( event_name="example.lookup", arguments=arguments, raw_arguments=raw_arguments, ) 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 @pytest.mark.asyncio async def test_registry_schema_is_isolated_from_caller_mutation(): parameters = { "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], "additionalProperties": False, } registry = EventRegistry([_definition(parameters=parameters)]) parameters["properties"]["query"]["type"] = "integer" parameters["required"].clear() parameters["additionalProperties"] = True assert registry.catalog()[0]["parameters"] == { "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], "additionalProperties": False, } registered = registry.definition("example.lookup") assert registered is not None with pytest.raises(TypeError): registered.parameters["additionalProperties"] = True with pytest.raises(TypeError): registered.parameters["properties"]["query"]["type"] = "integer" with pytest.raises(AttributeError): registered.parameters["required"].append("unexpected") result = await EventKernel(registry).execute( EventRequest( id="event-1", name="example.lookup", arguments={"query": 42, "unexpected": True}, source=EventSource.PROVIDER_RESOLVED, ), enabled_names=["example.lookup"], ) assert result.status is EventStatus.INVALID_ARGUMENTS @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("execution", ["async", "sync"]) async def test_kernel_isolates_nested_handler_mutation_from_audit_values( execution: str, ): caller_arguments = {"nested": {"items": ["original"]}} handler_arguments: list[dict[str, Any]] = [] def handler(request: EventRequest) -> dict[str, Any]: handler_arguments.append(request.arguments) request.arguments["nested"]["items"].append("handler") return {"nested": request.arguments["nested"]} definition = _definition( parameters={ "type": "object", "properties": { "nested": { "type": "object", "properties": { "items": {"type": "array", "items": {"type": "string"}} }, "required": ["items"], } }, "required": ["nested"], }, handler=handler, ) request = EventRequest( id="event-1", name="example.lookup", arguments=caller_arguments, raw_arguments='{"nested":{"items":["original"]}}', source=EventSource.PROVIDER_RESOLVED, ) kernel = EventKernel(EventRegistry([definition])) result = ( await kernel.execute(request, enabled_names=["example.lookup"]) if execution == "async" else kernel.execute_sync(request, enabled_names=["example.lookup"]) ) assert result.status is EventStatus.SUCCESS assert caller_arguments == {"nested": {"items": ["original"]}} assert request.arguments == {"nested": {"items": ["original"]}} assert result.arguments == {"nested": {"items": ["original"]}} assert json.loads(result.raw_arguments) == result.arguments assert result.payload == {"nested": {"items": ["original", "handler"]}} handler_arguments[0]["nested"]["items"].append("later") assert result.payload == {"nested": {"items": ["original", "handler"]}} result.payload["nested"]["items"].append("result") assert result.arguments == {"nested": {"items": ["original"]}} assert caller_arguments == {"nested": {"items": ["original"]}} @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 @pytest.mark.parametrize("execution", ["async", "sync"]) @pytest.mark.parametrize( "invalid_value", [object(), {"set-item"}, float("nan"), float("inf")], ) async def test_kernel_rejects_non_json_handler_dictionary_payloads( execution: str, invalid_value: Any, ): definition = _definition( handler=lambda request: {"invalid": invalid_value}, ) request = EventRequest( id="event-1", name="example.lookup", arguments={"query": "value"}, raw_arguments='{"query":"value"}', source=EventSource.PROVIDER_RESOLVED, ) kernel = EventKernel(EventRegistry([definition])) result = ( await kernel.execute(request, enabled_names=["example.lookup"]) if execution == "async" else kernel.execute_sync(request, enabled_names=["example.lookup"]) ) assert result.status is EventStatus.HANDLER_ERROR assert result.error == "event handler returned non-JSON 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 not 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"}