import json from datetime import datetime, timezone import pytest import agent_lab.application.events.builtin_plugins as builtin_plugins from agent_lab.application.events import ( EventArgumentResolution, EventExecutionContext, EventRequest, ResultPolicy, RiskLevel, ) from agent_lab.application.tools import build_default_tool_registry from agent_lab.domain.events import ToolCallEvent from agent_lab.domain.messages import ChatMessage BUILTIN_EVENT_NAMES = [ "session.terminate", "device.volume.adjust", "calendar.schedule.create", "knowledge.web.search", ] def _event(event_id: str, name: str, arguments: dict): return ToolCallEvent( id=event_id, name=name, arguments=arguments, raw_arguments=json.dumps(arguments, ensure_ascii=False), ) def test_default_registry_preserves_legacy_tools_and_adds_builtin_event_plugins(): registry = build_default_tool_registry() assert [item["name"] for item in registry.available_tools()] == [ "handoff_note", "mock_search", "mock_ticket", *BUILTIN_EVENT_NAMES, ] @pytest.mark.parametrize( ( "name", "result_policy", "risk_level", "fallback_allowed", "idempotency_key_fields", "concurrency_class", "conflict_keys", "timeout_seconds", "terminal", ), [ ( "session.terminate", ResultPolicy.TERMINATE, RiskLevel.HIGH, False, ("event_id",), "session-lifecycle", ("session",), 5.0, True, ), ( "device.volume.adjust", ResultPolicy.SILENT_SUCCESS, RiskLevel.MEDIUM, True, ("event_id",), "device-volume", ("device.volume",), 5.0, False, ), ( "calendar.schedule.create", ResultPolicy.TEMPLATE_FOLLOW_UP, RiskLevel.MEDIUM, True, ("event_id",), "schedule-write", ("calendar.schedule",), 10.0, False, ), ( "knowledge.web.search", ResultPolicy.LLM_FOLLOW_UP, RiskLevel.LOW, False, (), "read-only", (), 10.0, False, ), ], ) def test_builtin_definition_metadata_is_owned_by_each_flat_plugin( name, result_policy, risk_level, fallback_allowed, idempotency_key_fields, concurrency_class, conflict_keys, timeout_seconds, terminal, ): definition = build_default_tool_registry().event_registry.definition(name) assert definition is not None assert definition.result_policy is result_policy assert definition.risk_level is risk_level assert definition.fallback_allowed is fallback_allowed assert definition.idempotency_key_fields == idempotency_key_fields assert definition.concurrency_class == concurrency_class assert definition.conflict_keys == conflict_keys assert definition.timeout_seconds == timeout_seconds assert definition.terminal is terminal def test_session_terminate_schema_accepts_optional_reason_only(): registry = build_default_tool_registry().event_registry assert registry.iter_validation_errors("session.terminate", {}) == () assert registry.iter_validation_errors( "session.terminate", {"reason": "user requested"} ) == () assert registry.iter_validation_errors( "session.terminate", {"reason": 1} ) assert registry.iter_validation_errors( "session.terminate", {"unexpected": True} ) @pytest.mark.parametrize( "arguments", [ {"mode": "absolute", "value": 0}, {"mode": "absolute", "value": 100}, {"mode": "relative", "delta": -100}, {"mode": "relative", "delta": 100}, {"mode": "mute"}, {"mode": "unmute"}, ], ) def test_volume_schema_accepts_mode_specific_valid_arguments(arguments): registry = build_default_tool_registry().event_registry assert registry.iter_validation_errors("device.volume.adjust", arguments) == () @pytest.mark.parametrize( "arguments", [ {"mode": "absolute"}, {"mode": "absolute", "value": -1}, {"mode": "absolute", "value": 101}, {"mode": "absolute", "value": 20, "delta": 5}, {"mode": "relative"}, {"mode": "relative", "delta": 0}, {"mode": "relative", "delta": 101}, {"mode": "relative", "delta": 5, "value": 20}, {"mode": "mute", "value": 0}, {"mode": "unmute", "delta": 1}, ], ) def test_volume_schema_rejects_invalid_or_cross_mode_arguments(arguments): registry = build_default_tool_registry().event_registry assert registry.iter_validation_errors("device.volume.adjust", arguments) @pytest.mark.parametrize( ("content", "expected"), [ ("把音量调到 35", {"mode": "absolute", "value": 35}), ("set volume to 42", {"mode": "absolute", "value": 42}), ("音量增加 15", {"mode": "relative", "delta": 15}), ("decrease volume by 20", {"mode": "relative", "delta": -20}), ("请静音", {"mode": "mute"}), ("please unmute", {"mode": "unmute"}), ], ) def test_volume_resolver_recognizes_clear_chinese_and_english(content, expected): resolved = _resolve_text_event("device.volume.adjust", content) assert resolved == expected @pytest.mark.parametrize("content", ["把音量调高", "静音还是取消静音"]) def test_volume_resolver_marks_incomplete_or_ambiguous_requests_for_fallback(content): resolved = _resolve_text_event("device.volume.adjust", content) assert isinstance(resolved, EventArgumentResolution) assert resolved.complete is False def test_schedule_schema_requires_explicit_rfc3339_and_timezone(): registry = build_default_tool_registry().event_registry valid = { "title": "Design review", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", "recurrence": "FREQ=WEEKLY", "reminder_minutes": 15, } assert registry.iter_validation_errors("calendar.schedule.create", valid) == () assert registry.iter_validation_errors( "calendar.schedule.create", {**valid, "start_at": "tomorrow at nine"}, ) assert registry.iter_validation_errors( "calendar.schedule.create", {**valid, "timezone": "Shanghai"}, ) assert registry.iter_validation_errors( "calendar.schedule.create", {**valid, "reminder_minutes": -1}, ) @pytest.mark.parametrize( ("content", "expected"), [ ( 'Schedule "Design review" at 2026-07-14T09:30:00+08:00 ' "timezone Asia/Shanghai", { "title": "Design review", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ), ( "标题：项目评审；开始时间：2026-07-14T09:30:00+08:00；" "时区：Asia/Shanghai", { "title": "项目评审", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ), ], ) def test_schedule_resolver_accepts_only_explicit_datetime_details(content, expected): resolved = _resolve_text_event("calendar.schedule.create", content) assert resolved == expected def test_schedule_resolver_does_not_guess_relative_time(): resolved = _resolve_text_event( "calendar.schedule.create", "Schedule standup tomorrow at nine" ) assert isinstance(resolved, EventArgumentResolution) assert resolved.complete is False def test_web_search_resolver_uses_latest_relevant_user_request(): registry = build_default_tool_registry() definition = registry.event_registry.definition("knowledge.web.search") assert definition is not None assert definition.resolver is not None resolved = definition.resolver( EventRequest(id="search-1", name="knowledge.web.search"), EventExecutionContext( history=( ChatMessage(role="user", content="old query"), ChatMessage(role="assistant", content="assistant summary"), ChatMessage(role="user", content="latest focused query"), ChatMessage(role="assistant", content="I will search"), ) ), ) assert resolved == {"query": "latest focused query"} def test_web_search_schema_rejects_empty_query_and_out_of_range_result_limit(): registry = build_default_tool_registry().event_registry assert registry.iter_validation_errors( "knowledge.web.search", {"query": "focused", "max_results": 1} ) == () assert registry.iter_validation_errors( "knowledge.web.search", {"query": "focused", "max_results": 5} ) == () assert registry.iter_validation_errors( "knowledge.web.search", {"query": ""} ) assert registry.iter_validation_errors( "knowledge.web.search", {"query": "focused", "max_results": 0} ) assert registry.iter_validation_errors( "knowledge.web.search", {"query": "focused", "max_results": 6} ) def test_catalog_and_direct_provider_modes_share_the_same_builtin_schemas(): registry = build_default_tool_registry() catalog = {item["name"]: item for item in registry.available_tools()} direct = { item["function"]["name"]: item["function"] for item in registry.provider_tool_schemas(BUILTIN_EVENT_NAMES) } for name in BUILTIN_EVENT_NAMES: assert direct[name]["description"] == catalog[name]["description"] assert direct[name]["parameters"] == catalog[name]["parameters"] class RecordingSessionPort: def __init__(self): self.calls = [] def terminate(self, event_id, *, reason=None): self.calls.append((event_id, reason)) return {"port": "session", "event_id": event_id, "reason": reason} class RecordingVolumePort: def __init__(self): self.calls = [] def adjust(self, event_id, *, mode, value=None, delta=None): self.calls.append((event_id, mode, value, delta)) return { "port": "volume", "event_id": event_id, "mode": mode, "value": value, "delta": delta, } class RecordingCalendarPort: def __init__(self): self.calls = [] def create( self, event_id, *, title, start_at, timezone, recurrence=None, reminder_minutes=None, ): self.calls.append( ( event_id, title, start_at, timezone, recurrence, reminder_minutes, ) ) return {"port": "calendar", "event_id": event_id, "title": title} class RecordingSearchPort: def __init__(self): self.calls = [] def search(self, event_id, *, query, max_results=3): self.calls.append((event_id, query, max_results)) return {"port": "search", "query": query, "max_results": max_results} def test_injected_sync_ports_receive_validated_arguments_and_event_ids(): session = RecordingSessionPort() volume = RecordingVolumePort() calendar = RecordingCalendarPort() search = RecordingSearchPort() registry = build_default_tool_registry( session_termination_port=session, device_volume_port=volume, calendar_schedule_port=calendar, web_search_port=search, ) assert registry.execute( _event("session-1", "session.terminate", {"reason": "done"}) ) == {"port": "session", "event_id": "session-1", "reason": "done"} assert registry.execute( _event("volume-1", "device.volume.adjust", {"mode": "absolute", "value": 30}) )["port"] == "volume" assert registry.execute( _event( "calendar-1", "calendar.schedule.create", { "title": "Review", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", "reminder_minutes": 10, }, ) )["port"] == "calendar" assert registry.execute( _event( "search-1", "knowledge.web.search", {"query": "agent kernels", "max_results": 2}, ) ) == {"port": "search", "query": "agent kernels", "max_results": 2} assert session.calls == [("session-1", "done")] assert volume.calls == [("volume-1", "absolute", 30, None)] assert calendar.calls == [ ( "calendar-1", "Review", "2026-07-14T09:30:00+08:00", "Asia/Shanghai", None, 10, ) ] assert search.calls == [("search-1", "agent kernels", 2)] @pytest.mark.asyncio async def test_injected_async_ports_are_awaited_for_all_builtin_plugins(): class AsyncSessionPort(RecordingSessionPort): async def terminate(self, event_id, *, reason=None): return super().terminate(event_id, reason=reason) class AsyncVolumePort(RecordingVolumePort): async def adjust(self, event_id, *, mode, value=None, delta=None): return super().adjust( event_id, mode=mode, value=value, delta=delta ) class AsyncCalendarPort(RecordingCalendarPort): async def create(self, event_id, **arguments): return super().create(event_id, **arguments) class AsyncSearchPort(RecordingSearchPort): async def search(self, event_id, *, query, max_results=3): return super().search( event_id, query=query, max_results=max_results ) registry = build_default_tool_registry( session_termination_port=AsyncSessionPort(), device_volume_port=AsyncVolumePort(), calendar_schedule_port=AsyncCalendarPort(), web_search_port=AsyncSearchPort(), ) events = [ _event("session-1", "session.terminate", {}), _event( "volume-1", "device.volume.adjust", {"mode": "relative", "delta": -10}, ), _event( "calendar-1", "calendar.schedule.create", { "title": "Review", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ), _event("search-1", "knowledge.web.search", {"query": "agent kernels"}), ] payloads = [await registry.execute_async(event) for event in events] assert [payload["port"] for payload in payloads] == [ "session", "volume", "calendar", "search", ] @pytest.mark.parametrize( "arguments", [ {"mode": "absolute", "value": 101}, {"mode": "relative", "delta": 0}, {"mode": "mute", "value": 1}, ], ) def test_invalid_volume_arguments_never_call_the_port(arguments): volume = RecordingVolumePort() registry = build_default_tool_registry(device_volume_port=volume) payload = registry.execute(_event("volume-invalid", "device.volume.adjust", arguments)) assert payload["tool"] == "device.volume.adjust" assert "invalid" in payload["error"] assert volume.calls == [] def test_ambiguous_volume_request_without_fallback_never_calls_the_port(): volume = RecordingVolumePort() registry = build_default_tool_registry(device_volume_port=volume) payload = registry.handle( _event("volume-ambiguous", "device.volume.adjust", {}), EventExecutionContext( history=(ChatMessage(role="user", content="静音还是取消静音"),) ), ) assert payload == { "tool": "device.volume.adjust", "error": "missing required arguments: mode", } assert volume.calls == [] @pytest.mark.parametrize( ("port_name", "event"), [ ("session_termination_port", _event("s", "session.terminate", {})), ( "device_volume_port", _event("v", "device.volume.adjust", {"mode": "mute"}), ), ( "calendar_schedule_port", _event( "c", "calendar.schedule.create", { "title": "Review", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ), ), ( "web_search_port", _event("w", "knowledge.web.search", {"query": "agent kernels"}), ), ], ) def test_port_failures_are_normalized_by_the_generic_kernel(port_name, event): class FailingPort: def terminate(self, *args, **kwargs): raise RuntimeError("port unavailable") adjust = terminate create = terminate search = terminate registry = build_default_tool_registry(**{port_name: FailingPort()}) payload = registry.execute(event) assert payload == { "tool": event.name, "error": "tool handler failed: port unavailable", } def test_default_stateful_adapters_are_idempotent_by_event_id(): registry = build_default_tool_registry() first_session = registry.execute( _event("same-session", "session.terminate", {"reason": "first"}) ) second_session = registry.execute( _event("same-session", "session.terminate", {"reason": "second"}) ) first_volume = registry.execute( _event( "same-volume", "device.volume.adjust", {"mode": "absolute", "value": 20}, ) ) second_volume = registry.execute( _event( "same-volume", "device.volume.adjust", {"mode": "absolute", "value": 80}, ) ) first_schedule = registry.execute( _event( "same-schedule", "calendar.schedule.create", { "title": "First", "start_at": "2026-07-14T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ) ) second_schedule = registry.execute( _event( "same-schedule", "calendar.schedule.create", { "title": "Second", "start_at": "2026-07-15T09:30:00+08:00", "timezone": "Asia/Shanghai", }, ) ) assert second_session == first_session assert first_session["reason"] == "first" assert second_volume == first_volume assert first_volume["value"] == 20 assert second_schedule == first_schedule assert first_schedule["schedule"]["title"] == "First" def test_default_web_search_is_deterministic_compact_and_uses_injected_clock(): registry = build_default_tool_registry( clock=lambda: datetime(2030, 1, 2, 3, 4, 5, tzinfo=timezone.utc) ) payload = registry.execute( _event( "search-1", "knowledge.web.search", {"query": "agent kernels", "max_results": 2}, ) ) assert payload["tool"] == "knowledge.web.search" assert payload["query"] == "agent kernels" assert payload["retrieved_at"] == "2030-01-02T03:04:05Z" assert len(payload["sources"]) == 2 assert all(source["url"].startswith("https://example.invalid/") for source in payload["sources"]) assert all(set(source) == {"title", "url", "snippet"} for source in payload["sources"]) def test_builtin_port_protocols_and_default_adapter_types_are_public(): assert builtin_plugins.SessionTerminationPort assert builtin_plugins.DeviceVolumePort assert builtin_plugins.CalendarSchedulePort assert builtin_plugins.WebSearchPort assert builtin_plugins.InMemorySessionTerminationAdapter assert builtin_plugins.InMemoryDeviceVolumeAdapter assert builtin_plugins.InMemoryCalendarScheduleAdapter assert builtin_plugins.InMemoryWebSearchAdapter def _resolve_text_event(name: str, content: str): registry = build_default_tool_registry() definition = registry.event_registry.definition(name) assert definition is not None assert definition.resolver is not None return definition.resolver( EventRequest(id="event-1", name=name), EventExecutionContext(history=(ChatMessage(role="user", content=content),)), )